appfactory-portal/app/docker_logs.py

"""Diagnostické čtení Docker logů kontejneru služby.

Portál si logy nikam nekopíruje ani nezakládá vlastní úložiště - jen čte to, co už
Docker drží pro daný kontejner (`docker logs`). Slouží k dohledání příčiny HTTP 500
bez SSH přístupu na server.

Bezpečnost:
* docker se volá vždy přes argument array (subprocess bez ``shell=True``), takže se
  uživatelský vstup nemůže dostat do příkazové řádky,
* název kontejneru se odvozuje z ``app_id`` služby existující v databázi, nikdy se
  nebere přímo z URL,
* core služby AppFactory jsou z tohoto čtení vyloučené (CORE_CONTAINERS),
* ``since`` a ``tail`` mají allowlist, cokoli jiného spadne na bezpečný default.
"""

import re
import subprocess

# Docker odpovídá lokálně přes socket; delší čekání by jen blokovalo request portálu.
DOCKER_TIMEOUT_SECONDS = 10

# Povolené hodnoty parametru --since. Klíč je hodnota z URL, hodnota jde do dockeru.
ALLOWED_LOG_PERIODS = {
    "15m": "15m",
    "30m": "30m",
    "1h": "1h",
    "6h": "6h",
    "24h": "24h",
}

# Popisky pro UI v pořadí, v jakém se mají nabídnout.
LOG_PERIOD_LABELS = (
    ("15m", "posledních 15 minut"),
    ("30m", "posledních 30 minut"),
    ("1h", "poslední hodina"),
    ("6h", "posledních 6 hodin"),
    ("24h", "posledních 24 hodin"),
)

# Povolené hodnoty parametru --tail.
ALLOWED_LOG_TAILS = (100, 300, 500, 1000)

DEFAULT_LOG_PERIOD = "30m"
DEFAULT_LOG_TAIL = 500

# Core služby AppFactory. Diagnostika uživatelských aplikací k nim nemá přístup - pro
# jejich logy by muselo vzniknout samostatné administrační oprávnění.
CORE_CONTAINERS = frozenset(
    {
        "appfactory-gitea",
        "appfactory-portal",
        "appfactory-worker",
        "appfactory-monitor",
        "appfactory-webhook",
        "appfactory-caddy",
        "appfactory-registry",
    }
)

# Docker povoluje v názvu kontejneru [a-zA-Z0-9][a-zA-Z0-9_.-]*. Držíme se toho i tady,
# ať se do argumentu nikdy nedostane nic jiného (obrana navíc, název stejně vzniká z app_id).
CONTAINER_NAME_RE = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9_.-]{0,127}$")

# Typické chybové indikátory - slouží jen ke zvýraznění řádku, obsah logu se nemění.
ERROR_LINE_RE = re.compile(r"error|fail|exception|fatal|\b50[0234]\b", re.IGNORECASE)

# Stavy kontejneru, jak je vrací `docker inspect --format {{.State.Status}}`.
CONTAINER_STATE_LABELS = {
    "created": "Created",
    "restarting": "Restarting",
    "running": "Running",
    "removing": "Removing",
    "paused": "Paused",
    "exited": "Exited",
    "dead": "Dead",
}


class DockerLogsError(Exception):
    """Chyba čtení Docker logů. Text je určený k zobrazení uživateli."""


def normalize_period(value: str | None) -> str:
    """Vrátí povolenou hodnotu --since, jinak bezpečný default."""
    return ALLOWED_LOG_PERIODS.get((value or "").strip(), DEFAULT_LOG_PERIOD)


def normalize_tail(value) -> int:
    """Vrátí povolený počet řádků --tail, jinak bezpečný default."""
    try:
        number = int(str(value).strip())
    except (TypeError, ValueError):
        return DEFAULT_LOG_TAIL
    return number if number in ALLOWED_LOG_TAILS else DEFAULT_LOG_TAIL


def state_label(status: str | None) -> str:
    if not status:
        return ""
    return CONTAINER_STATE_LABELS.get(status, status)


def container_candidates(app_id: str) -> list[str]:
    """Názvy kontejnerů, pod kterými může služba běžet (stejné schéma jako u restartu .env).

    Core kontejnery a syntakticky nemožné názvy se z kandidátů rovnou vyhazují.
    """
    base = (app_id or "").strip()
    normalized = base.replace("_", "-")
    candidates: list[str] = []
    for candidate in (base, normalized, f"appfactory-{normalized}"):
        if not candidate or candidate in candidates:
            continue
        if candidate in CORE_CONTAINERS:
            continue
        if not CONTAINER_NAME_RE.match(candidate):
            continue
        candidates.append(candidate)
    return candidates


def _ensure_container_allowed(container_name: str) -> str:
    name = (container_name or "").strip()
    if not CONTAINER_NAME_RE.match(name):
        raise DockerLogsError("Neplatný název kontejneru.")
    if name in CORE_CONTAINERS:
        raise DockerLogsError("Logy core služeb AppFactory nejsou přes tuto stránku dostupné.")
    return name


def _run_docker(args: list[str], merge_stderr: bool = False) -> subprocess.CompletedProcess:
    """Spustí docker přes argument array. Nikdy shell=True, nikdy skládání příkazu z textu."""
    try:
        return subprocess.run(
            ["docker", *args],
            stdout=subprocess.PIPE,
            # U `docker logs` teče stdout kontejneru do stdout a stderr do stderr. Sloučením
            # do jednoho proudu zůstane zachované původní pořadí řádků, což je pro hledání
            # příčiny chyby podstatné.
            stderr=subprocess.STDOUT if merge_stderr else subprocess.PIPE,
            text=True,
            encoding="utf-8",
            errors="replace",
            timeout=DOCKER_TIMEOUT_SECONDS,
        )
    except FileNotFoundError as exc:
        raise DockerLogsError("Docker klient není v portálu dostupný.") from exc
    except subprocess.TimeoutExpired as exc:
        raise DockerLogsError(f"Docker neodpověděl do {DOCKER_TIMEOUT_SECONDS} sekund (timeout).") from exc
    except OSError as exc:
        raise DockerLogsError(f"Docker se nepodařilo spustit: {exc}") from exc


def docker_daemon_available() -> bool:
    """True, pokud portál dosáhne na Docker daemon (volá se jen při chybě, kvůli hlášce)."""
    try:
        result = _run_docker(["version", "--format", "{{.Server.Version}}"])
    except DockerLogsError:
        return False
    return result.returncode == 0


def inspect_container(container_name: str) -> str | None:
    """Stav kontejneru (running/exited/paused/...), nebo None když v Dockeru není."""
    name = _ensure_container_allowed(container_name)
    result = _run_docker(["inspect", "--format", "{{.State.Status}}", name])
    if result.returncode != 0:
        return None
    return (result.stdout or "").strip().lower() or None


def resolve_app_container(app_id: str) -> tuple[str, str | None]:
    """Najde kontejner služby. Vrací (název, stav); stav je None, když kontejner neexistuje."""
    candidates = container_candidates(app_id)
    if not candidates:
        raise DockerLogsError("Ze služby nelze odvodit platný název kontejneru.")

    for candidate in candidates:
        status = inspect_container(candidate)
        if status:
            return candidate, status

    return candidates[0], None


def get_container_logs(
    container_name: str,
    since: str = DEFAULT_LOG_PERIOD,
    tail: int = DEFAULT_LOG_TAIL,
) -> str:
    """Vrátí `docker logs --since <since> --tail <tail> <container>` jako text (stdout i stderr)."""
    name = _ensure_container_allowed(container_name)
    since_value = normalize_period(since)
    tail_value = normalize_tail(tail)

    result = _run_docker(
        ["logs", "--since", since_value, "--tail", str(tail_value), name],
        merge_stderr=True,
    )
    output = result.stdout or ""
    if result.returncode != 0:
        detail = short_detail(output)
        raise DockerLogsError(detail or f"Příkaz docker logs skončil s návratovým kódem {result.returncode}.")
    return output


def short_detail(value: str) -> str:
    """Zkrácený technický detail chyby pro zobrazení (escapuje se až v šabloně)."""
    text = " ".join((value or "").split())
    if len(text) > 500:
        return f"{text[:497]}..."
    return text


def split_log_lines(text: str) -> list[str]:
    if not text:
        return []
    normalized = text.replace("\r\n", "\n").replace("\r", "\n")
    lines = normalized.split("\n")
    while lines and not lines[-1]:
        lines.pop()
    return lines


def filter_log_lines(lines: list[str], query: str) -> list[str]:
    """Filtruje už načtené řádky v Pythonu - žádný grep, žádný shell."""
    needle = (query or "").strip().lower()
    if not needle:
        return list(lines)
    return [line for line in lines if needle in line.lower()]


def is_error_line(line: str) -> bool:
    return bool(ERROR_LINE_RE.search(line or ""))