KindClustersDashboard/scripts/detect_podman_socket.py

#!/usr/bin/env python3
"""Печатает в stdout путь к ``podman.sock`` или ``uid:gid`` владельца сокета.

Режимы:

* без аргументов — путь (для ``$(shell ...)`` в Makefile);
* ``--print-owner`` — ``uid:gid`` по ``stat`` найденного сокета (для ``KIND_K8S_CONTAINER_UIDGID``).

Логика пути: ``podman info`` (схема ``unix://`` снимается), ``podman machine inspect`` (без путей
в ``/var/folders/…`` на macOS — там API-сокет, **compose не смонтирует**: ``operation not supported``),
стабильные ``*.sock`` под ``~/.local/share/containers/podman/machine/``, затем типичные пути пользователя.

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

from __future__ import annotations

import json
import os
import shutil
import subprocess
import sys
from pathlib import Path


def _compose_cannot_bind_mount_socket(path_str: str) -> bool:
    """
    На macOS сокет в ``/var/folders/…/T/podman/`` (например ``*-api.sock`` из machine inspect)
    существует на диске, но bind-mount в контейнер через compose часто даёт
    ``mkdir … operation not supported`` — такой путь в ``CONTAINER_SOCKET`` использовать нельзя.
    """
    if sys.platform != "darwin":
        return False
    norm = os.path.normpath(path_str)
    return "/var/folders/" in norm


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


def _podman_machine_socket_paths(podman_bin: str) -> list[str]:
    """
    Пути к сокету на **хосте** из ``podman machine inspect`` (актуально для macOS/Windows).

    На Linux без machine обычно возвращается ``[]``.
    """
    try:
        proc = subprocess.run(
            [podman_bin, "machine", "inspect"],
            capture_output=True,
            text=True,
            timeout=20,
        )
        if proc.returncode != 0 or not (proc.stdout or "").strip():
            return []
        data = json.loads(proc.stdout)
        if not isinstance(data, list):
            return []
        out: list[str] = []
        for m in data:
            if not isinstance(m, dict):
                continue
            ci = m.get("ConnectionInfo") or {}
            if not isinstance(ci, dict):
                continue
            ps = ci.get("PodmanSocket")
            pth: str | None = None
            if isinstance(ps, dict):
                raw = ps.get("Path")
                pth = str(raw).strip() if raw else None
            elif isinstance(ps, str) and ps.strip():
                pth = ps.strip()
            if pth:
                cleaned = _strip_unix_scheme(pth)
                if not _compose_cannot_bind_mount_socket(cleaned):
                    out.append(cleaned)
        return out
    except (OSError, subprocess.TimeoutExpired, json.JSONDecodeError, TypeError, ValueError):
        return []


def _darwin_stable_machine_sockets() -> list[str]:
    """Подкаталог machine на macOS: сокеты вне временного ``/var/folders``."""
    base = Path.home() / ".local/share/containers/podman/machine"
    if sys.platform != "darwin" or not base.is_dir():
        return []
    out: list[str] = []
    try:
        for p in sorted(base.rglob("*.sock"), key=lambda x: (len(x.parts), str(x))):
            s = str(p)
            if _compose_cannot_bind_mount_socket(s):
                continue
            out.append(s)
    except OSError:
        return []
    return out


def _podman_socket_path_for_user() -> 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 podman_socket_owner_uidgid() -> str:
    """
    UID:GID владельца найденного ``podman.sock`` на диске (``stat``).

    Нужен для ``user:`` в compose: при Podman на macOS и в VM владелец сокета часто
    **не совпадает** с ``id -u`` / ``id -g`` процесса ``make`` на хосте — иначе
    ``permission denied`` на смонтированный сокет внутри контейнера.
    Если сокет недоступен — fallback на UID:GID текущего процесса.
    """
    path_str = detect_podman_socket_path()
    p = Path(path_str)
    try:
        if p.is_socket():
            st = p.stat()
            return f"{st.st_uid}:{st.st_gid}"
    except OSError:
        pass
    try:
        return f"{os.getuid()}:{os.getgid()}"
    except AttributeError:
        return "1000:1000"


def detect_podman_socket_path() -> str:
    """Путь к существующему сокету или типичный путь по умолчанию."""
    cands: list[str] = []
    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()
                candidate = (line[0] if line else "").strip()
                if candidate and "<no value>" not in candidate.lower():
                    cands.append(_strip_unix_scheme(candidate))
        except (OSError, subprocess.TimeoutExpired):
            pass
        # Реальный файл на хосте (без macOS temp api.sock — см. _compose_cannot_bind_mount_socket).
        cands.extend(_podman_machine_socket_paths(podman_bin))
    cands.extend(_darwin_stable_machine_sockets())
    cands.append(_podman_socket_path_for_user())
    for path in cands:
        cleaned = _strip_unix_scheme(path)
        if _compose_cannot_bind_mount_socket(cleaned):
            continue
        p = Path(cleaned).expanduser()
        try:
            if p.is_socket():
                return str(p.resolve())
        except OSError:
            continue
    return str(Path(_strip_unix_scheme(_podman_socket_path_for_user())).expanduser())


def main() -> None:
    if len(sys.argv) > 1 and sys.argv[1] == "--print-owner":
        sys.stdout.write(podman_socket_owner_uidgid())
        return
    sys.stdout.write(detect_podman_socket_path())


if __name__ == "__main__":
    main()