#!/usr/bin/env python3
"""
Проверка Unix-сокетов Docker и Podman: есть ли доступ без EACCES и отвечает ли API (GET /_ping).

Печатает таблицу и рекомендуемый ``DOCKER_HOST`` для рабочих вариантов.
Запуск на хосте (не внутри контейнера веб-UI).

Важно: вывод ``podman info`` (RemoteSocket.Path) на macOS часто указывает путь **внутри VM**
Podman Machine (например ``/run/podman/...``) — на диске macOS такого файла нет. «Настоящий»
сокет для монтирования в compose — первый **существующий** ``*.sock`` из той же логики, что
``scripts/detect_podman_socket.py`` (как у ``make podman``).

Автор: Сергей Антропов
Сайт: https://devops.org.ru
"""

from __future__ import annotations

import argparse
import importlib.util
import logging
import os
import shutil
import socket
import stat
import subprocess
import sys
from pathlib import Path
from typing import Iterable

logger = logging.getLogger(__name__)

# Минимальный запрос к Docker-compatible API (Podman отвечает так же).
_PING_REQUEST = b"GET /_ping HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n"


def _docker_host_unix_from_env() -> str | None:
    raw = (os.environ.get("DOCKER_HOST") or "").strip()
    if raw.startswith("unix://"):
        return raw[len("unix://") :].strip() or None
    return None


def _normalize_socket_path(raw: str) -> str:
    """
    Путь к сокету на диске: убираем префикс ``unix://`` / ``unix:`` (как в выводе ``podman info``).
    """
    p = (raw or "").strip()
    if p.startswith("unix://"):
        p = p[7:]
    elif p.startswith("unix:"):
        rest = p[5:].lstrip("/")
        p = "/" + rest if rest else p
    return p.strip() or raw.strip()


def _detect_podman_socket_path_via_project_script() -> str | None:
    """
    Тот же поиск «живого» сокета на **этой** ФС, что и ``make podman`` / compose.

    Возвращает ``None``, если модуль не загрузился.
    """
    script_path = Path(__file__).resolve().parent / "detect_podman_socket.py"
    try:
        spec = importlib.util.spec_from_file_location("_kind_k8s_detect_podman", script_path)
        if spec is None or spec.loader is None:
            return None
        mod = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(mod)
        fn = getattr(mod, "detect_podman_socket_path", None)
        if callable(fn):
            return str(fn())
    except Exception as e:
        logger.debug("detect_podman_socket.py: %s", e)
    return None


def _podman_default_path() -> str:
    xdg = (os.environ.get("XDG_RUNTIME_DIR") or "").strip()
    if xdg:
        return f"{xdg.rstrip('/')}/podman/podman.sock"
    try:
        return f"/run/user/{os.getuid()}/podman/podman.sock"
    except AttributeError:
        return "/run/user/1000/podman/podman.sock"


def _collect_candidates() -> list[tuple[str, str]]:
    """
    Список (метка, путь). Порядок: переменная окружения, CLI, стандартные пути.
    Дубликаты путей убираем, метки склеиваем через « + ».
    """
    # Ключ — нормализованный путь; для несуществующих файлов не вызываем resolve().
    buckets: dict[str, tuple[list[str], str]] = {}
    order_keys: list[str] = []

    def add(label: str, path: str | None) -> None:
        if not path:
            return
        expanded = str(Path(_normalize_socket_path(path)).expanduser())
        key = os.path.normpath(expanded)
        if key not in buckets:
            order_keys.append(key)
            buckets[key] = ([label], expanded)
        else:
            buckets[key][0].append(label)

    env_u = _docker_host_unix_from_env()
    if env_u:
        add("DOCKER_HOST", env_u)

    # Реальный путь к файлу сокета на хосте (не «логический» путь из VM в podman info).
    det = _detect_podman_socket_path_via_project_script()
    if det:
        add("хост: detect_podman_socket.py (как make podman)", det)

    podman_bin = shutil.which("podman")
    if podman_bin:
        try:
            proc = subprocess.run(
                [podman_bin, "info", "-f", "{{.Host.RemoteSocket.Path}}"],
                capture_output=True,
                text=True,
                timeout=15,
            )
            if proc.returncode == 0:
                line = (proc.stdout or "").strip().splitlines()
                cand = (line[0] if line else "").strip()
                if cand and "<no value>" not in cand.lower():
                    add("podman info", cand)
        except (OSError, subprocess.TimeoutExpired) as e:
            logger.debug("podman info: %s", e)

    docker_bin = shutil.which("docker")
    if docker_bin:
        try:
            proc = subprocess.run(
                [docker_bin, "context", "inspect", "--format", "{{.Endpoints.docker.Host}}"],
                capture_output=True,
                text=True,
                timeout=15,
            )
            if proc.returncode == 0:
                host = (proc.stdout or "").strip().splitlines()
                h = (host[0] if host else "").strip()
                if h.startswith("unix://"):
                    add("docker context", h[len("unix://") :])
        except (OSError, subprocess.TimeoutExpired) as e:
            logger.debug("docker context: %s", e)

    add("стандарт Docker", "/var/run/docker.sock")
    home = Path.home()
    add("Docker Desktop / Linux", str(home / ".docker/run/docker.sock"))
    add("Colima", str(home / ".colima/default/docker.sock"))
    add("Podman типичный", _podman_default_path())

    return [(" + ".join(buckets[k][0]), buckets[k][1]) for k in order_keys]


def _stat_line(path: Path) -> str:
    try:
        if not path.exists():
            return "нет файла"
        if not path.is_socket():
            return "не сокет"
        st = path.stat()
        mode = stat.filemode(st.st_mode)
        return f"{mode} uid={st.st_uid} gid={st.st_gid}"
    except OSError as e:
        return f"stat: {e}"


def _probe_http_ping(sock_path: Path, timeout_sec: float) -> tuple[str, str]:
    """Статус: ok | denied | conn_err | bad_reply | missing."""
    try:
        if not sock_path.exists():
            return "missing", "файл отсутствует"
        if not sock_path.is_socket():
            return "missing", "не Unix-сокет"
    except OSError as e:
        return "missing", str(e)

    try:
        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        sock.settimeout(timeout_sec)
        sock.connect(os.fspath(sock_path))
        sock.sendall(_PING_REQUEST)
        chunks: list[bytes] = []
        while True:
            part = sock.recv(4096)
            if not part:
                break
            chunks.append(part)
            if len(chunks) > 8:  # достаточно для заголовков + OK
                break
        sock.close()
    except PermissionError:
        return "denied", "EACCES при connect (как у docker info)"
    except OSError as e:
        return "conn_err", str(e)

    raw = b"".join(chunks)
    if b" 200 " in raw and b"OK" in raw:
        return "ok", "HTTP 200, тело OK"
    head = raw[: min(120, len(raw))].decode("utf-8", errors="replace")
    return "bad_reply", head.replace("\r", " ").replace("\n", " ")


def _probe_docker_cli(unix_path: str, timeout_sec: int) -> tuple[str, str]:
    docker = shutil.which("docker")
    if not docker:
        return "skip", "docker CLI нет в PATH"
    uri = f"unix://{unix_path}"
    try:
        proc = subprocess.run(
            [docker, "-H", uri, "info"],
            capture_output=True,
            text=True,
            timeout=timeout_sec,
        )
        if proc.returncode == 0:
            return "ok", "docker -H … info → 0"
        err = (proc.stderr or proc.stdout or "").strip()[:200]
        if "permission denied" in err.lower() or "eacces" in err.lower():
            return "denied", err
        return "cli_err", err or f"код {proc.returncode}"
    except subprocess.TimeoutExpired:
        return "cli_err", "timeout"
    except OSError as e:
        return "cli_err", str(e)


def _run(candidates: Iterable[tuple[str, str]], timeout: float, with_docker_cli: bool) -> int:
    rows: list[tuple[str, str, str, str, str]] = []
    working: list[str] = []

    for label, path_str in candidates:
        p = Path(path_str).expanduser()
        st = _stat_line(p)
        http_st, http_msg = _probe_http_ping(p, timeout_sec=timeout)
        cli_st, cli_msg = ("—", "—")
        if with_docker_cli:
            cli_st, cli_msg = _probe_docker_cli(path_str, timeout_sec=int(timeout) + 5)
        rows.append((label, path_str, st, f"{http_st}: {http_msg}", f"{cli_st}: {cli_msg}"))
        if http_st == "ok":
            working.append(f"unix://{path_str}")

    w = 22
    print(
        f"{'Источник':<{w}} {'Путь':<40} {'Файл':<28} {'HTTP /_ping':<35} {'docker CLI':<30}",
    )
    print("-" * 145)
    for label, path_str, st, http_col, cli_col in rows:
        p_short = path_str if len(path_str) <= 40 else path_str[:37] + "..."
        print(f"{label:<{w}} {p_short:<40} {st:<28} {http_col:<35} {cli_col:<30}")

    me = ""
    try:
        me = f"{os.getuid()}:{os.getgid()}"
    except AttributeError:
        me = "?"

    print()
    print(f"Текущий процесс: uid:gid = {me}")
    if working:
        print("\nРабочие URI (HTTP /_ping OK), можно задать:")
        for u in working:
            print(f"  export DOCKER_HOST={u!r}")
        print("\nДля compose этого проекта см. CONTAINER_SOCKET / KIND_K8S_REMOTE_SOCKET_URI в .env.")
        return 0

    print(
        "\nНи один сокет не ответил на /_ping без отказа. "
        "Проверьте, что Docker Desktop / Podman machine запущены; для Podman на macOS часто помогает "
        "запуск веб-UI от 0:0 (см. Makefile, make podman) или нативный «podman compose».",
    )
    return 1


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Проверка сокетов Docker/Podman: доступ к API без permission denied.",
    )
    parser.add_argument(
        "-t",
        "--timeout",
        type=float,
        default=5.0,
        help="Таймаут connect/read для сокета (сек)",
    )
    parser.add_argument(
        "--no-docker-cli",
        action="store_true",
        help="Не вызывать «docker -H … info»",
    )
    parser.add_argument(
        "-v",
        "--verbose",
        action="store_true",
        help="Подробный журнал",
    )
    args = parser.parse_args()
    logging.basicConfig(level=logging.DEBUG if args.verbose else logging.WARNING)

    cands = _collect_candidates()
    if not cands:
        print("Нет кандидатов путей.", file=sys.stderr)
        return 2

    return _run(cands, timeout=args.timeout, with_docker_cli=not args.no_docker_cli)


if __name__ == "__main__":
    raise SystemExit(main())