tls v0.1.2 [base]

"""tls.certs — локальный CA и серверные сертификаты (openssl CLI).

Схема: долгоживущий CA (10 лет) + серверный серт (TLS_DAYS, по
умолчанию 375) с SAN из TLS_HOSTS. ensure() идемпотентен: выпускает
только если файлов нет, срок < RENEW_DAYS или SAN не покрывают хосты.
Перевыпуск при работающем сервере требует рестарта (флаг renewed).
"""
import logging
import os
import re
import subprocess
import threading

logger = logging.getLogger(__name__)

_lock = threading.RLock()
_cfg = {"dir": "", "hosts": "IP:127.0.0.1", "days": 375,
        "renew_days": 30, "ca_days": 3650}

_CA_KEY = "ca.key"
_CA_CRT = "ca.crt"
_SRV_KEY = "server.key"
_SRV_CSR = "server.csr"
_SRV_CRT = "server.crt"


def configure(directory="", hosts="", days=375, renew_days=30,
              ca_days=3650):
    with _lock:
        _cfg.update({"dir": (directory or "").strip(),
                     "hosts": (hosts or "IP:127.0.0.1").strip(),
                     "days": max(1, int(days or 375)),
                     "renew_days": max(1, int(renew_days or 30)),
                     "ca_days": max(30, int(ca_days or 3650))})


def _run(*args, input_text=None):
    """openssl CLI. input_text — для -extfile /dev/stdin (расширения SAN).

    '' — ок, иначе текст ошибки.
    """
    fixed = list(args)
    tmp = ""
    if input_text is not None and "/dev/stdin" in fixed:
        import tempfile as _tf
        try:
            with _tf.NamedTemporaryFile("w", suffix=".cnf",
                                        delete=False) as f:
                f.write("[SAN]\n%s\n" % input_text)
                tmp = f.name
        except OSError as e:
            return "нет временного файла: %s" % e
        fixed = [a if a != "/dev/stdin" else tmp for a in fixed]
    try:
        subprocess.run(fixed, check=True, timeout=120,
                       stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
        return ""
    except FileNotFoundError:
        return "нет бинарника openssl"
    except subprocess.CalledProcessError as e:
        err = (e.stderr or b"").decode("utf-8", "ignore").strip()
        return "openssl: %s" % (err or e)
    finally:
        if tmp:
            try:
                os.unlink(tmp)
            except OSError:
                pass


def _enddate(path):
    """Остаток дней жизни серта (None — не прочитался)."""
    try:
        p = subprocess.run(["openssl", "x509", "-in", path, "-noout",
                            "-enddate"], check=True, timeout=30,
                           stdout=subprocess.PIPE,
                           stderr=subprocess.DEVNULL)
    except Exception:  # noqa: BLE001
        return None
    m = re.search(r"notAfter=(.*)", p.stdout.decode("utf-8",
                                                   "ignore").strip())
    if not m:
        return None
    from datetime import datetime, timezone
    for fmt in ("%b %d %H:%M:%S %Y %Z", "%b %e %H:%M:%S %Y %Z"):
        try:
            dt = datetime.strptime(m.group(1).strip(), fmt)
            dt = dt.replace(tzinfo=timezone.utc)
            delta = dt - datetime.now(timezone.utc)
            return delta.total_seconds() / 86400.0
        except ValueError:
            continue
    return None


def _san_covered(path, hosts):
    """Покрывает ли серт все SAN из hosts (IP:/DNS: через запятую)."""
    try:
        p = subprocess.run(["openssl", "x509", "-in", path, "-noout",
                            "-ext", "subjectAltName"], check=True,
                           timeout=30, stdout=subprocess.PIPE,
                           stderr=subprocess.DEVNULL)
    except Exception:  # noqa: BLE001
        return False
    text = p.stdout.decode("utf-8", "ignore")
    have = set()
    for m in re.finditer(r"IP Address:([0-9a-fA-F:.]+)", text):
        have.add("IP:" + m.group(1))
    for m in re.finditer(r"DNS:([^\s,]+)", text):
        have.add("DNS:" + m.group(1))
    want = [h.strip() for h in (hosts or "").split(",") if h.strip()]
    return all(w in have for w in want)


def _paths():
    d = _cfg["dir"]
    return {n: os.path.join(d, n) for n in
            (_CA_KEY, _CA_CRT, _SRV_KEY, _SRV_CSR, _SRV_CRT)}


def ensure():
    """Гарантировать пару (cert, key). Возврат dict{ca,cert,key,renewed,expires_in,err}.

    renewed=True — файлы перевыпущены (нужен рестарт слушателей).
    """
    with _lock:
        d = _cfg["dir"]
        if not d:
            return {"err": "не задан каталог TLS"}
        try:
            os.makedirs(d, exist_ok=True)
        except OSError as e:
            return {"err": "нет каталога TLS: %s" % e}
        P = _paths()
        renewed = False

        def _need_server():
            if not (os.path.exists(P[_SRV_CRT])
                    and os.path.exists(P[_SRV_KEY])):
                return True
            left = _enddate(P[_SRV_CRT])
            if left is None or left < _cfg["renew_days"]:
                return True
            return not _san_covered(P[_SRV_CRT], _cfg["hosts"])

        if not (os.path.exists(P[_CA_KEY]) and os.path.exists(P[_CA_CRT])):
            err = _run("openssl", "req", "-x509", "-newkey", "rsa:2048",
                       "-nodes", "-keyout", P[_CA_KEY], "-out", P[_CA_CRT],
                       "-days", str(_cfg["ca_days"]), "-subj", "/CN=bot-local-CA")
            if err:
                return {"err": err}
            try:
                os.chmod(P[_CA_KEY], 0o600)
            except OSError:
                pass
            logger.info("tls: выпущен локальный CA")
        if _need_server():
            err = _run("openssl", "req", "-newkey", "rsa:2048", "-nodes",
                       "-keyout", P[_SRV_KEY], "-out", P[_SRV_CSR],
                       "-subj", "/CN=bot-local")
            if err:
                return {"err": err}
            try:
                os.chmod(P[_SRV_KEY], 0o600)
            except OSError:
                pass
            san = "subjectAltName=" + ",".join(
                h.strip() for h in _cfg["hosts"].split(",") if h.strip())
            err = _run("openssl", "x509", "-req", "-in", P[_SRV_CSR],
                       "-CA", P[_CA_CRT], "-CAkey", P[_CA_KEY],
                       "-CAcreateserial", "-days", str(_cfg["days"]),
                       "-out", P[_SRV_CRT],
                       "-extfile", "/dev/stdin", "-extensions", "SAN",
                       input_text=san)
            if err:
                return {"err": err}
            renewed = True
            logger.info("tls: выпущен серверный серт (%s)", _cfg["hosts"])
        try:
            os.unlink(P[_SRV_CSR])
        except OSError:
            pass
        left = _enddate(P[_SRV_CRT])
        return {"ca": P[_CA_CRT], "cert": P[_SRV_CRT], "key": P[_SRV_KEY],
                "renewed": renewed,
                "expires_in": round(left, 1) if left is not None else None,
                "err": ""}


def limits_hosts():
    """SAN-хосты конфига (для статуса без чтения привата)."""
    with _lock:
        return _cfg.get("hosts", "")


def ca_pem():
    """Текст CA-сертификата (для установки на устройства) или ''."""
    with _lock:
        p = _paths().get(_CA_CRT, "")
        try:
            with open(p, "r", encoding="utf-8") as f:
                return f.read()
        except OSError:
            return ""

"""tls.Module — локальный CA и серверные серты (лист, requires: []).

transport_web с WEB_TLS_AUTO=1 забирает пару через tls.ensure при старте
(модули грузятся раньше транспортов). Перевыпуск при живом сервере —
флаг renewed (нужен рестарт). CA ставится на устройства один раз —
текст в tls.status (ca_pem) и GET /api/tls/status.
"""
import logging
import os

from core.base_module import BaseModule
from core.paths import abs_data

logger = logging.getLogger(__name__)


class Module(BaseModule):
    def setup(self, ctx) -> None:
        self.config = ctx.module_config("tls")
        self.ctx = ctx
        from . import certs as _certs
        self._certs = _certs

        def _int(key, default):
            try:
                return int(str(self.config.get(key, default) or default))
            except (TypeError, ValueError):
                return default

        _certs.configure(
            directory=abs_data(ctx.data_dir,
                               (self.config.get("TLS_DIR", "") or "").strip(),
                               os.path.join(ctx.data_dir, "cache", "tls")),
            hosts=(self.config.get("TLS_HOSTS", "") or "IP:127.0.0.1").strip(),
            days=_int("TLS_DAYS", 375),
            renew_days=_int("TLS_RENEW_DAYS", 30),
            ca_days=_int("TLS_CA_DAYS", 3650))

        api = ctx.api.register
        api("tls", "ensure", _certs.ensure)
        api("tls", "status", self.api_status)

    # --- ctx.api ---

    def api_status(self):
        st = dict(self._certs.ensure())
        if st.get("err"):
            return {"configured": False, "err": st["err"]}
        st["configured"] = True
        st["hosts"] = self._certs.limits_hosts()
        st["ca_pem"] = self._certs.ca_pem()
        return st

    def health(self) -> dict:
        return {"ok": True, "module": self.name}

"""tls.ui_web — GET /api/tls/status (статус + CA-pem).

Право tls_manage. Ручной перевыпуск не нужен: ensure() при каждом
статусе сам продлевает (renewed=true — перезапустить веб).
"""

ROUTES = [
    ("GET", "/api/tls/status", "status", {"right": "tls_manage",
     "desc": "TLS: срок серта, хосты, CA-pem (renewed — нужен рестарт)"}),
]

SERVICE = []

PANELS = []

MONITOR_PANELS = [
    {"id": "tls", "title": "TLS: сертификат",
     "api": "/api/tls/status", "visibility": "right:tls_manage",
     "refresh_s": 300},
]


def _sess_uid(req):
    sess = req.get("session")
    if sess:
        return sess.get("uid")
    return None


def _check(ctx, sess, right) -> bool:
    try:
        return bool(ctx.api.call("transport_web.check", sess, right))
    except Exception:  # noqa: BLE001
        return False


def handle_api(ctx, config, method, req):
    from core.errors import UserError

    if method != "status":
        raise UserError("неизвестный метод: %s" % method)
    if _sess_uid(req) is None:
        return 401, {"ok": False, "error": "Требуется вход"}
    if not _check(ctx, req.get("session"), "tls_manage"):
        return 403, {"ok": False, "error": "Нет права tls_manage"}
    try:
        st = ctx.api.call("tls.status")
    except Exception as e:  # noqa: BLE001
        return 503, {"ok": False, "error": "TLS недоступен: %s" % e}
    st["ok"] = True
    return st