computer-playbook/utils/cleanup/nginx_vhosts.py

"""nginx vhost-file purge helper (entity-keyed).

Matrix-deploy variant transitions can leave per-domain nginx vhost
configs behind: when a service drops out of a round's include set,
``sys-svc-compose``'s aggressive docker prune removes the container,
but the vhost file under
``<NGINX_DIR>/conf.d/servers/{http,https}/<domain>.conf`` survives.
The next round's probe then gets HTTP 502 (nginx serves the vhost with
no upstream) instead of the expected TLS-error 000.

The entity-keyed purge primitive
``scripts/container/purge/entity/nginx.sh`` (driven by the app-keyed
orchestrator ``apps.sh`` as a sibling of ``db.sh``, ``compose.sh``,
``dir.sh``) invokes this module to remove the vhost files belonging to
a given compose entity. The mapping entity -> apps -> domains is
resolved via
``utils.roles.entity.apps.apps_for_entity`` plus
``utils.domains.list.build_applications_from_roles`` /
``utils.domains.application_domain_index.iter_app_domains`` so the
helper stays ansible-free and runnable inside the infinito container
without a templar context.
"""

from __future__ import annotations

import os
import sys
from pathlib import Path
from typing import TYPE_CHECKING

from utils.domains.application_domain_index import iter_app_domains
from utils.domains.default_primary import default_domain_primary
from utils.domains.list import ROLES_DIR, build_applications_from_roles
from utils.roles.entity.apps import apps_for_entity
from utils.tor_onion import identity_hs_dir

if TYPE_CHECKING:
    from collections.abc import Iterable

DEFAULT_NGINX_DIR = Path("/etc/nginx")
_PROTOCOLS: tuple[str, ...] = ("http", "https")


def _resolve_nginx_dir(nginx_dir: Path | None) -> Path:
    if nginx_dir is not None:
        return nginx_dir
    env_value = os.environ.get("NGINX_DIR")
    return Path(env_value) if env_value else DEFAULT_NGINX_DIR


def _resolve_domain_primary(domain_primary: str | None) -> str:
    if domain_primary is not None:
        return domain_primary
    return (
        os.environ.get("DOMAIN")
        or os.environ.get("INFINITO_DOMAIN")
        or default_domain_primary()
    )


def _read_node_onion(base_dir: Path) -> str:
    """The provisioned node onion address, read from the authoritative key file
    (``.onion-identity/hs/hostname`` — the same source ``services.tor.node`` is
    provisioned from); empty when no onion identity is present."""
    hostname = identity_hs_dir(base_dir) / "hostname"
    try:
        return hostname.read_text(encoding="ascii").strip()  # nocheck: cache-read
    except OSError:
        return ""


def _swap_to_onion(domain: str, primary: str, node_onion: str) -> str | None:
    """Dual-stack: the onion vhost of a clearnet domain, with the primary
    suffix swapped for the node onion. None when not under the primary."""
    if domain == primary:
        return node_onion
    suffix = "." + primary
    if domain.endswith(suffix):
        return domain[: -len(primary)] + node_onion
    return None


def iter_vhost_files_for_entity(
    entity: str,
    *,
    nginx_dir: Path,
    domain_primary: str,
    roles_dir: Path = ROLES_DIR,
) -> Iterable[Path]:
    """Yield existing vhost files (http + https) for every app belonging
    to *entity*, plus the ``www.<domain>.conf`` redirect variant and — when a
    node onion is provisioned (``services.tor.node`` / the key file) — the
    dual-stack ``<sub>.<node-onion>.conf`` variant, so a dropped app leaves no
    stale onion vhost either. Only yields files that exist.
    """
    apps = apps_for_entity(entity, roles_dir=roles_dir)
    if not apps:
        return

    applications = build_applications_from_roles(roles_dir, domain_primary)
    servers_dir = nginx_dir / "conf.d" / "servers"
    node_onion = _read_node_onion(roles_dir.parent)

    for app_id in apps:
        app_conf = applications.get(app_id)
        if not app_conf:
            continue
        for domain in iter_app_domains(app_conf, include_aliases=True):
            domain_variants = [domain]
            if not domain.startswith("www."):
                domain_variants.append(f"www.{domain}")
            if node_onion and (
                onion := _swap_to_onion(domain, domain_primary, node_onion)
            ):
                domain_variants.append(onion)
                if not onion.startswith("www."):
                    domain_variants.append(f"www.{onion}")
            for variant in domain_variants:
                for proto in _PROTOCOLS:
                    conf_file = servers_dir / proto / f"{variant}.conf"
                    if conf_file.is_file():
                        yield conf_file


def purge_vhost_files_for_entities(
    entities: list[str],
    *,
    nginx_dir: Path | None = None,
    domain_primary: str | None = None,
    roles_dir: Path = ROLES_DIR,
) -> list[Path]:
    """Remove vhost files for every entity in *entities*. Returns the
    list of paths that were actually removed (existed and unlink
    succeeded), in iteration order.
    """
    nd = _resolve_nginx_dir(nginx_dir)
    dp = _resolve_domain_primary(domain_primary)

    removed: list[Path] = []
    for entity in entities:
        for path in iter_vhost_files_for_entity(
            entity,
            nginx_dir=nd,
            domain_primary=dp,
            roles_dir=roles_dir,
        ):
            try:
                path.unlink()
            except FileNotFoundError:
                continue
            except OSError as exc:
                print(
                    f"!!! WARNING: failed to remove {path}: {exc}",
                    file=sys.stderr,
                )
                continue
            removed.append(path)
    return removed


def main(argv: list[str]) -> int:
    if not argv:
        print(
            "usage: python -m utils.cleanup.nginx_vhosts <ENTITY> [ENTITY ...]",
            file=sys.stderr,
        )
        return 2

    removed = purge_vhost_files_for_entities(argv)
    if removed:
        for path in removed:
            print(f">>> removed {path}")
    else:
        print(">>> No nginx vhost files to remove for entities: " + ", ".join(argv))
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))