computer-playbook/plugins/lookup/tor_ports.py

from __future__ import annotations

from typing import Any

from ansible.errors import AnsibleError
from ansible.plugins.loader import lookup_loader
from ansible.plugins.lookup import LookupBase

UDP_ONLY_CATEGORIES = frozenset({"relay", "media", "stun_turn", "stun_turn_tls"})
"""Categories a hidden service cannot carry.

``relay`` and ``media`` are UDP. ``stun_turn`` and its TLS variant do speak TCP,
but the allocation they hand out is a UDP relay address the onion client can
never receive on, so forwarding the signalling port buys nothing.
"""


def _as_bool(value: object) -> bool:
    if isinstance(value, bool):
        return value
    if isinstance(value, str):
        return value.strip().lower() in ("true", "1", "yes", "on")
    return bool(value)


def _mapping(value: object) -> dict[str, Any]:
    return value if isinstance(value, dict) else {}


def _collect_local_tcp_ports(port_categories: Any, into: set[int]) -> None:
    """Add single-int TCP ports of the loopback-published ``ports.local`` group
    only. An exposed service publishes just its local plaintext port to
    127.0.0.1, so the onion HiddenServicePort must target that; forwarding a
    ``public`` TLS port (e.g. ldaps 636) would be a dead loopback target (nothing
    listens there in the exposed variant) and plaintext-into-TLS."""
    if not isinstance(port_categories, dict):
        return
    local = port_categories.get("local")
    if not isinstance(local, dict):
        return
    for category, value in local.items():
        if category in UDP_ONLY_CATEGORIES:
            continue
        if isinstance(value, int):
            into.add(value)


def _provided_capabilities(
    applications: dict[str, Any], deployed: set[str]
) -> dict[str, set[str]]:
    """Capability key -> the deployed roles whose service ``provides:`` it.

    Args:
        applications: the variant-merged applications view.
        deployed: role ids present in this deploy.
    """
    provided: dict[str, set[str]] = {}
    for app_id, cfg in applications.items():
        if app_id not in deployed or not isinstance(cfg, dict):
            continue
        services = cfg.get("services")
        if not isinstance(services, dict):
            continue
        for entity in services.values():
            if not isinstance(entity, dict):
                continue
            key = entity.get("provides")
            if isinstance(key, str) and key:
                provided.setdefault(key, set()).add(app_id)
    return provided


def _is_inactive_coverer(
    entity: dict[str, Any], app_id: str, provided: dict[str, set[str]]
) -> bool:
    """Whether this service only covers a capability another deployed role provides."""
    covers = entity.get("covers")
    if not isinstance(covers, (list, tuple)):
        return False
    return any(
        isinstance(key, str) and (provided.get(key, set()) - {app_id}) for key in covers
    )


def collect_onion_ports(
    applications: dict[str, Any],
    deployed_roles: list[str],
) -> list[int]:
    """Single-int TCP ports a deployed role asks for by name under ``ports.onion``.

    ``ports.onion`` is a category-keyed map of booleans alongside ``internal`` /
    ``local`` / ``public``: it names which of a service's already-declared ports
    should answer on the node onion, and nothing else. The port number comes
    from ``ports.local`` when that category is declared there and from
    ``ports.public`` otherwise, which is the precedence ``container_ports``
    publishes by, so the forward always targets the port the service is actually
    reachable on.

    Naming the category is what makes this safe. Sweeping ``ports.public``
    wholesale forwards ports the deployment never publishes -- ldaps 636 is
    declared next to ``network.public: false``, and Mailu's implicit-TLS ports
    are dropped from the publish list whenever TLS is off, which an onion
    deployment always is.

    A service that only ``covers:`` a capability another deployed role
    ``provides:`` is skipped for the same reason: it is a legacy co-tenant, the
    provider owns the host ports, and forwarding the co-tenant's onion ports
    advertises endpoints nothing answers on. The Mailu-to-Stalwart migration
    round deploys both, and the ports Mailu alone would serve (110, 143, 587)
    are dead there.
    """
    ports: set[int] = set()
    deployed = set(deployed_roles)
    if not isinstance(applications, dict):
        return []
    provided = _provided_capabilities(applications, deployed)
    for app_id, cfg in applications.items():
        if app_id not in deployed:
            continue
        services = (cfg or {}).get("services") if isinstance(cfg, dict) else None
        if not isinstance(services, dict):
            continue
        for entity in services.values():
            if not isinstance(entity, dict):
                continue
            if _is_inactive_coverer(entity, app_id, provided):
                continue
            declared = entity.get("ports")
            if not isinstance(declared, dict):
                continue
            wanted = declared.get("onion")
            if not isinstance(wanted, dict):
                continue
            local = _mapping(declared.get("local"))
            public = _mapping(declared.get("public"))
            for category, flag in wanted.items():
                if category in UDP_ONLY_CATEGORIES or not _as_bool(flag):
                    continue
                value = local.get(category, public.get(category))
                if isinstance(value, int):
                    ports.add(value)
    return sorted(ports)


def collect_exposed_ports(
    applications: dict[str, Any],
    deployed_roles: list[str],
) -> list[int]:
    """Single-int TCP ports of every deployed service flagged ``exposed: true``
    in the variant-merged ``applications`` view, sorted.

    ``exposed`` is an explicit per-service opt-in (default false) that makes the
    service reachable over the node onion: its port gets a dedicated
    ``HiddenServicePort``. Because ``applications`` is the variant-merged config
    (base ``meta/services.yml`` deep-merged with the round's ``meta/variants.yml``
    override), a service is forwarded only in the variant/config where it sets
    ``exposed: true`` — which is exactly what lets the two DB variants be tested
    over Tor (v0 exposed -> reachable, v1 not -> refused).
    """
    ports: set[int] = set()
    deployed = set(deployed_roles)
    if not isinstance(applications, dict):
        return []
    for app_id, cfg in applications.items():
        if app_id not in deployed:
            continue
        services = (cfg or {}).get("services") if isinstance(cfg, dict) else None
        if not isinstance(services, dict):
            continue
        for entity in services.values():
            if not isinstance(entity, dict) or not _as_bool(entity.get("exposed")):
                continue
            _collect_local_tcp_ports(entity.get("ports"), ports)
    return sorted(ports)


class LookupModule(LookupBase):
    """
    Usage:
        {{ lookup('tor_ports') }}

    Returns the HiddenServicePort mappings for the roles in the current deploy
    (``group_names``): a list of ``{'onion_port': <port>, 'target':
    '127.0.0.1:<port>'}`` dicts, sorted by port. Two opt-ins are unioned:

      * every port a service names under ``ports.onion`` in meta/services.yml,
        and
      * every port of a service that opts in with ``exposed: true``.

    Both are read from the variant-merged applications view, so a variant that
    drops the opt-in drops the forward with it. The published port binds the
    host interface, so the loopback target reaches it from svc-net-tor's
    host-network container. Composed with the flagged forwards by
    ``lookup('tor_extra_ports')`` and rendered as ``HiddenServicePort`` lines in
    svc-net-tor's torrc.
    """

    def run(
        self,
        terms: list[Any] | None,
        variables: dict[str, Any] | None = None,
        **kwargs: Any,
    ) -> list[Any]:
        if terms:
            raise AnsibleError("lookup('tor_ports') expects no terms.")

        variables = variables or getattr(self._templar, "available_variables", {}) or {}
        group_names = variables.get("group_names") or []
        if not isinstance(group_names, list):
            group_names = []
        deployed = [str(g) for g in group_names]

        try:
            applications = lookup_loader.get(
                "applications",
                loader=getattr(self, "_loader", None),
                templar=getattr(self, "_templar", None),
            ).run([], variables=variables, roles_dir=kwargs.get("roles_dir"))[0]
        except Exception:  # noqa: BLE001  no merged view -> no ports to forward
            applications = {}
        ports = set(collect_onion_ports(applications, deployed))
        ports.update(collect_exposed_ports(applications, deployed))

        return [
            [
                {"onion_port": port, "target": f"127.0.0.1:{port}"}
                for port in sorted(ports)
            ]
        ]