KindClustersDashboard/app/kubeconfig_patch.py

"""Правка kubeconfig kind для доступа к API с хоста (после create из контейнера).

Kind внутри Docker видит другой адрес apiserver; для **kubectl на машине пользователя**
в ``kubeconfig.host`` подставляется ``https://<клиентский_хост>:<порт>`` (порт из
``docker port <cluster>-control-plane 6443/tcp``). Хост по умолчанию — **localhost**
(часто надёжнее ``127.0.0.1`` с TLS); переопределение: ``KIND_K8S_KUBECONFIG_CLIENT_HOST``
(например IP хоста в LAN).

**Два файла в ``clusters/<имя>/``:**

- ``kubeconfig`` — как выдал ``kind get kubeconfig``; для ``kubectl`` **в процессе веб-приложения**
  :func:`kubeconfig_path_for_container_kubectl` задаёт ``server=https://host.docker.internal:<порт>``
  и ``tls-server-name=localhost`` (иначе x509 не совпадает с SAN kind); см. ``KIND_K8S_KUBECONFIG_TLS_SERVER_NAME``.
  Запасной URL: ``https://<имя>-control-plane:6443`` (без доп. SNI).
- ``kubeconfig.host`` — копия с ``server=`` для kubectl на хосте (см.
  :func:`kubeconfig_download_client_hostname`).
- Скачивание через API **``GET …/kubeconfig/docker``** — тот же endpoint, что для kubectl в веб-контейнере:
  ``https://<KIND_K8S_APISERVER_GATEWAY_HOST>:<порт>`` (по умолчанию **host.docker.internal** и порт из ``docker port``)
  плюс **tls-server-name** (см. :func:`patch_kubeconfig_server_for_docker_network`), чтобы **любой** контейнер с
  ``extra_hosts: host.docker.internal:host-gateway`` достучался до API через хост, без общей сети с kind.

Если патч отключён или не удалился, ``kubeconfig.host`` может отсутствовать — тогда
отдаётся основной ``kubeconfig``.

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

from __future__ import annotations

import logging
import os
import shutil
import subprocess
import tempfile
import threading
from pathlib import Path

from kind_k8s_paths import container_cli_name

logger = logging.getLogger("kind_k8s.kubeconfig_patch")

# Имя файла kubeconfig с server для доступа с хоста (не использовать для kubectl в контейнере веб-приложения).
KUBECONFIG_HOST_BASENAME = "kubeconfig.host"


def _container_cli() -> str:
    """CLI для ``port`` к смонтированному сокету (в контейнере веб-UI — ``docker``)."""
    return container_cli_name()


def _control_plane_container_name(cluster_name: str) -> str:
    return f"{cluster_name}-control-plane"


def _parse_docker_port_line(line: str) -> tuple[str, str] | None:
    """Строка вида '0.0.0.0:32768' или '127.0.0.1:32768' -> (host, port)."""
    line = line.strip()
    if ":" not in line:
        return None
    # IPv6 [::]:port
    if line.startswith("["):
        rb = line.rfind("]")
        if rb == -1:
            return None
        host = line[1:rb]
        rest = line[rb + 1 :].lstrip(":")
        if not rest.isdigit():
            return None
        return host, rest
    host, _, port = line.rpartition(":")
    if not port.isdigit():
        return None
    host = host.strip()
    return host, port


def _host_bind_for_kubeconfig(host: str) -> str:
    if host in ("0.0.0.0", "::", ""):
        return "127.0.0.1"
    if host == "[::]":
        return "127.0.0.1"
    return host


def kubeconfig_download_client_hostname() -> str:
    """
    Имя хоста для поля ``server`` в kubeconfig, скачиваемом пользователем.

    По умолчанию ``localhost`` (совпадает с SAN в сертификате kind чаще, чем проблемный 127.0.0.1).
    Задаётся ``KIND_K8S_KUBECONFIG_CLIENT_HOST`` (например ``192.168.1.10`` для удалённого клиента).
    """
    raw = (os.environ.get("KIND_K8S_KUBECONFIG_CLIENT_HOST") or "localhost").strip()
    return raw or "localhost"


def get_apiserver_host_port(cluster_name: str) -> tuple[str, str] | None:
    """Узнать (host, port) привязки проброса 6443/tcp (host из вывода docker port, порт — число)."""
    cli = _container_cli()
    ctr = _control_plane_container_name(cluster_name)
    p = subprocess.run(
        [cli, "port", ctr, "6443/tcp"],
        capture_output=True,
        text=True,
    )
    if p.returncode != 0:
        logger.info(
            "Не удалось %s port %s 6443/tcp: %s",
            cli,
            ctr,
            (p.stderr or p.stdout or "").strip(),
        )
        return None
    for raw in (p.stdout or "").splitlines():
        parsed = _parse_docker_port_line(raw)
        if not parsed:
            continue
        h, port = parsed
        return _host_bind_for_kubeconfig(h), port
    return None


def _kubeconfig_cluster_name(kube_path: Path, logical_name: str) -> str:
    """Имя блока cluster в kubeconfig (обычно kind-<имя>)."""
    p = subprocess.run(
        [
            "kubectl",
            "--kubeconfig",
            str(kube_path),
            "config",
            "view",
            "-o",
            'jsonpath={range .clusters[*]}{.name}{"\n"}{end}',
        ],
        capture_output=True,
        text=True,
    )
    names = [x.strip() for x in (p.stdout or "").splitlines() if x.strip()]
    want = f"kind-{logical_name}"
    if want in names:
        return want
    for n in names:
        if n.startswith("kind-"):
            return n
    return want


def patch_kubeconfig_server_for_host(
    *,
    cluster_name: str,
    kube_path: Path,
) -> bool:
    """
    Подставить в kubeconfig server=https://<хост>:<порт> для доступа с хоста.

    Порт берётся из ``docker port`` / аналога к сокету; для Podman — тот же клиент
    при ``DOCKER_HOST`` на podman.sock.
    """
    hp = get_apiserver_host_port(cluster_name)
    if not hp:
        print(
            "Предупреждение: не удалось получить порт apiserver с хоста; "
            "kubeconfig оставлен как выдал kind (с хоста может не открываться).",
        )
        return False
    _bind_host, port = hp
    client_host = kubeconfig_download_client_hostname()
    server = f"https://{client_host}:{port}"
    cluster_id = _kubeconfig_cluster_name(kube_path, cluster_name)
    p = subprocess.run(
        [
            "kubectl",
            "--kubeconfig",
            str(kube_path),
            "config",
            "set-cluster",
            cluster_id,
            f"--server={server}",
        ],
        capture_output=True,
        text=True,
    )
    if p.returncode != 0:
        err = (p.stderr or p.stdout or "").strip()
        print(f"Предупреждение: kubectl config set-cluster не удался: {err}")
        return False
    print(f"Kubeconfig для хоста: apiserver → {server}")
    return True


def should_patch_after_create() -> bool:
    """Патчить после create, если задано явно или kind шёл из контейнера."""
    if os.environ.get("KIND_K8S_PATCH_KUBECONFIG", "").strip().lower() in ("1", "true", "yes", "да"):
        return True
    return os.environ.get("KIND_K8S_IN_CONTAINER", "").strip() == "1"


def in_kind_k8s_container() -> bool:
    """Процесс веб-приложения в Docker/Podman (``KIND_K8S_IN_CONTAINER=1``)."""
    return os.environ.get("KIND_K8S_IN_CONTAINER", "").strip() == "1"


def kubeconfig_host_file(cluster_data_dir: Path) -> Path:
    """Путь к kubeconfig для скачивания на хост: ``clusters/<имя>/kubeconfig.host``."""
    return cluster_data_dir / KUBECONFIG_HOST_BASENAME


def apiserver_url_docker_bridge(cluster_name: str) -> str:
    """
    URL apiserver по имени узла kind в общей Docker-сети (если она есть).

    Стандартное имя контрольной плоскости: ``<cluster>-control-plane``.
    """
    return f"https://{_control_plane_container_name(cluster_name)}:6443"


def _web_container_apiserver_endpoint(cluster_name: str) -> tuple[str, str | None]:
    """
    Пара ``(server URL, tls-server-name)`` для kubeconfig процесса в контейнере веб-UI.

    Через шлюз (``host.docker.internal:<порт>``) в сертификате kind нет этого имени;
    задаём ``tls-server-name`` (по умолчанию ``localhost``), которое есть в SAN.
    Без проброса порта — прямой ``https://<имя>-control-plane:6443``, ``tls-server-name`` не нужен.
    """
    hp = get_apiserver_host_port(cluster_name)
    if hp:
        _bind_host, port = hp
        gw = (os.environ.get("KIND_K8S_APISERVER_GATEWAY_HOST") or "host.docker.internal").strip()
        if not gw:
            gw = "host.docker.internal"
        server = f"https://{gw}:{port}"
        tls_name = (os.environ.get("KIND_K8S_KUBECONFIG_TLS_SERVER_NAME") or "localhost").strip() or "localhost"
        return server, tls_name
    return apiserver_url_docker_bridge(cluster_name), None


def apply_apiserver_endpoint_to_kubeconfig_file(
    *,
    cluster_name: str,
    kube_path: Path,
) -> tuple[bool, str, str | None, str]:
    """
    Выставить в kubeconfig ``server`` и при необходимости ``tls-server-name`` по
    :func:`_web_container_apiserver_endpoint` (шлюз хоста + порт или запасной ``*-control-plane:6443``).

    Возвращает ``(успех, server, tls_server_name|None, cluster_id_в_kubeconfig)``.
    """
    cluster_id = _kubeconfig_cluster_name(kube_path, cluster_name)
    server, tls_server_name = _web_container_apiserver_endpoint(cluster_name)
    p = subprocess.run(
        [
            "kubectl",
            "--kubeconfig",
            str(kube_path),
            "config",
            "set-cluster",
            cluster_id,
            f"--server={server}",
        ],
        capture_output=True,
        text=True,
    )
    if p.returncode != 0:
        err = (p.stderr or p.stdout or "").strip()
        logger.warning(
            "apply apiserver kubeconfig: set-cluster не удался для %s: %s",
            cluster_name,
            err[:500],
        )
        return False, server, tls_server_name, cluster_id

    if tls_server_name:
        p_tls = subprocess.run(
            [
                "kubectl",
                "--kubeconfig",
                str(kube_path),
                "config",
                "set-cluster",
                cluster_id,
                f"--tls-server-name={tls_server_name}",
            ],
            capture_output=True,
            text=True,
        )
        if p_tls.returncode != 0:
            err_tls = (p_tls.stderr or p_tls.stdout or "").strip()
            logger.warning(
                "apply apiserver kubeconfig: tls-server-name для %s: %s",
                cluster_name,
                err_tls[:400],
            )
    else:
        subprocess.run(
            [
                "kubectl",
                "--kubeconfig",
                str(kube_path),
                "config",
                "unset",
                f"clusters.{cluster_id}.tls-server-name",
            ],
            capture_output=True,
            text=True,
        )

    return True, server, tls_server_name, cluster_id


def patch_kubeconfig_server_for_docker_network(
    *,
    cluster_name: str,
    kube_path: Path,
) -> bool:
    """
    Kubeconfig для **kubectl из произвольного контейнера** (не обязательно в сети kind).

    При успешном ``docker port … 6443/tcp``: ``server=https://<шлюз>:<порт>`` (по умолчанию
    **host.docker.internal**, см. ``KIND_K8S_APISERVER_GATEWAY_HOST``) и **tls-server-name**
    (по умолчанию **localhost**, ``KIND_K8S_KUBECONFIG_TLS_SERVER_NAME``) — как у
    :func:`kubeconfig_path_for_container_kubectl`. Трафик идёт на хост и в проброшенный порт
    control plane, DNS **имени узла kind не нужен**.

    Если порт с хоста не получить — запасной ``https://<имя>-control-plane:6443`` без SNI
    (нужна общая Docker-сеть с узлом kind).
    """
    ok, server, tls_sni, cluster_id = apply_apiserver_endpoint_to_kubeconfig_file(
        cluster_name=cluster_name,
        kube_path=kube_path,
    )
    if ok:
        logger.info(
            "kubeconfig для стороннего контейнера: %s tls-server-name=%s (cluster=%s)",
            server,
            tls_sni or "—",
            cluster_id,
        )
    return ok


def apiserver_url_for_web_container_process(cluster_name: str) -> str:
    """Только URL apiserver для ``kubectl`` в контейнере веб-приложения (см. :func:`_web_container_apiserver_endpoint`)."""
    url, _tls = _web_container_apiserver_endpoint(cluster_name)
    return url


# Кэш временных kubeconfig для kubectl из контейнера: ключ → (mtime_ns исходника, путь).
# Версия в ключе — сброс кэша при смене логики (например tls-server-name).
_RUNTIME_KUBECONFIG_CACHE_VER = 2
_kubectl_runtime_cache: dict[str, tuple[int, Path]] = {}
_kubectl_runtime_lock = threading.Lock()


def kubeconfig_path_for_container_kubectl(*, cluster_name: str, kube_source_path: Path) -> Path:
    """
    Путь к kubeconfig для вызова ``kubectl`` из процесса в контейнере.

    Если ``KIND_K8S_IN_CONTAINER=1``, копия во временный файл: ``server`` через шлюз хоста
    и при необходимости ``tls-server-name`` (SAN kind: ``localhost``, …).
    Иначе возвращает ``kube_source_path`` без изменений.

    Поддерживает старые установки с одним «хостовым» kubeconfig на диске.
    """
    if not in_kind_k8s_container():
        return kube_source_path

    try:
        src = kube_source_path.resolve()
        mtime_ns = int(src.stat().st_mtime_ns)
    except OSError as e:
        logger.debug("kubeconfig runtime: нет исходника %s: %s", kube_source_path, e)
        return kube_source_path

    key = f"{cluster_name}\x00{src}\x00{_RUNTIME_KUBECONFIG_CACHE_VER}"
    with _kubectl_runtime_lock:
        hit = _kubectl_runtime_cache.get(key)
        if hit is not None and hit[0] == mtime_ns:
            return hit[1]

        try:
            fd, temp_name = tempfile.mkstemp(suffix=".kubeconfig", prefix="kind-k8s-kubectl-")
            os.close(fd)
            tmp_path = Path(temp_name)
            shutil.copyfile(src, tmp_path, follow_symlinks=True)
        except OSError as e:
            logger.warning("kubeconfig runtime: не удалось скопировать %s: %s", src, e)
            return kube_source_path

        ok, server, tls_server_name, cluster_id = apply_apiserver_endpoint_to_kubeconfig_file(
            cluster_name=cluster_name,
            kube_path=tmp_path,
        )
        if not ok:
            tmp_path.unlink(missing_ok=True)
            return kube_source_path

        prev = _kubectl_runtime_cache.pop(key, None)
        if prev is not None:
            try:
                prev[1].unlink(missing_ok=True)
            except OSError:
                pass

        _kubectl_runtime_cache[key] = (mtime_ns, tmp_path)
        logger.debug(
            "kubeconfig runtime: кластер «%s» server=%s tls-sni=%s (cluster=%s)",
            cluster_name,
            server,
            tls_server_name or "—",
            cluster_id,
        )
        return tmp_path