"""Domain-name cache: canonical-domains map derived from merged apps. Owns `_MERGED_DOMAINS_CACHE`. Public API: `get_merged_domains`. The domain map is intentionally derived from the applications view rather than living in a parallel top-level overrides path. Per-app domain declarations live in `applications..domains` and flow through the regular applications-merge pipeline. """ from __future__ import annotations from typing import TYPE_CHECKING, Any from .base import ( _RENDER_GUARD, _cache_key, _resolve_roles_dir, _stable_variables_signature, ) if TYPE_CHECKING: import os _MERGED_DOMAINS_CACHE: dict[tuple, dict[str, Any]] = {} def _as_bool(value: Any) -> bool: if isinstance(value, bool): return value if isinstance(value, str): return value.strip().lower() in ("true", "1", "yes", "on") return bool(value) def _resolved_flag( value: Any, *, app: str, name: str, templar: Any, variables: Any, apps: Any, ) -> bool: """A tor service flag as a boolean, never as the text of its own template. ``services.tor.*`` is declared as Jinja in 99 of the 105 roles that carry it. When this map is built from an applications tree that is still being rendered (``utils.cache.applications`` returns the raw tree while its own render is in flight) the flag arrives as its source text, which every boolean coercion reads as False. The onion injection would then silently not happen and the clearnet domain would be frozen into whatever value the render was producing. Args: value: the declared flag, rendered or not. app: application id, for the error message. name: flag name under ``services.tor``, for the error message. templar: templar to render with; ``None`` renders nothing. variables: scope the template resolves against. apps: applications tree seeded into the nested render. Returns: The flag as a boolean. Raises: ValueError: the flag still carries template text after rendering. """ if isinstance(value, str) and ("{{" in value or "{%" in value): from utils.cache.base import _render_with_templar value = _render_with_templar( value, templar=templar, variables=variables, raw_applications=apps ) if isinstance(value, str) and ("{{" in value or "{%" in value): raise ValueError( f"get_merged_domains: '{app}' services.tor.{name} stayed template " f"text ({value!r}). Reading it as a boolean would drop the onion " "domains without saying so." ) return _as_bool(value) def _onion_of(domain: str, primary: str, node_onion: str) -> str | None: """Return ``domain`` with the clearnet ``primary`` suffix swapped for the node onion, or None when the domain is not under ``primary``.""" if domain == primary: return node_onion suffix = "." + primary if domain.endswith(suffix): return domain[: -len(primary)] + node_onion return None def _inject_onion_domains( merged: dict[str, Any], apps: dict[str, Any], primary: str, node_onion: str, *, templar: Any = None, variables: Any = None, ) -> dict[str, Any]: """Add ``.`` domains for apps opting into onion routing. Driven by ``services.tor`` per app: ``enabled`` gates it (consumer-owned, no provider fallback — an app without a tor block opts out). ``exclusive`` replaces the clearnet domain (onion only), ``primary`` puts the onion first; both resolve consumer-override -> ``svc-net-tor`` provider default via :func:`resolve_service_config`. """ from utils.roles.applications.services.config import resolve_service_config from utils.roles.applications.services.registry import ( build_service_registry_from_applications, ) registry = build_service_registry_from_applications(apps) def _flag(value: Any, app: str, name: str) -> bool: return _resolved_flag( value, app=app, name=name, templar=templar, variables=variables, apps=apps, ) out: dict[str, Any] = {} for app, domains in merged.items(): tor = ((apps.get(app) or {}).get("services") or {}).get("tor") or {} if not isinstance(domains, (list, dict)) or not _flag( tor.get("enabled"), app, "enabled" ): out[app] = domains continue exclusive = _flag( resolve_service_config( apps, app, "tor", "exclusive", default=False, service_registry=registry ), app, "exclusive", ) is_primary = _flag( resolve_service_config( apps, app, "tor", "primary", default=False, service_registry=registry ), app, "primary", ) if isinstance(domains, dict): out[app] = _inject_onion_into_dict( domains, primary, node_onion, exclusive, is_primary ) continue onion = [o for d in domains if (o := _onion_of(str(d), primary, node_onion))] if exclusive: out[app] = onion or domains elif is_primary: out[app] = onion + [d for d in domains if d not in onion] else: out[app] = list(domains) + [o for o in onion if o not in domains] return out def _inject_onion_into_dict( domains: dict[str, Any], primary: str, node_onion: str, exclusive: bool, is_primary: bool, ) -> dict[str, Any]: """Onionize a named-canonical dict (``{key: domain}``) while keeping every value a plain string, so ``get_primary_domain`` still returns a string. ``exclusive`` swaps each value for its onion; otherwise the onion domains are added under ``_onion`` siblings (before the clearnet keys when ``primary`` is set) so ``generate_all_domains`` emits an onion vhost per named canonical without changing which value is primary. """ if exclusive: return { key: (_onion_of(str(value), primary, node_onion) or value) for key, value in domains.items() } onion_pairs: dict[str, Any] = {} for key, value in domains.items(): onion = _onion_of(str(value), primary, node_onion) if onion and onion != value: onion_pairs[f"{key}_onion"] = onion if is_primary: return {**onion_pairs, **domains} return {**domains, **onion_pairs} def get_merged_domains( *, variables: dict[str, Any] | None = None, roles_dir: str | os.PathLike[str] | None = None, templar: Any = None, ) -> dict[str, Any]: """Build the canonical-domain map lazily from the merged applications view. The result is canonical_domains_map(applications, DOMAIN_PRIMARY). Per-app domain declarations live in `applications..domains` (canonical/aliases) and flow through the regular applications-merge pipeline. Cached keyed on (roles_dir, variables_signature, render_state). A read issued while the applications render is in progress sees the unrendered tree, whose service flags are still Jinja strings; its map is cached under its own key so the render's own call sites reuse it without it ever being served as the finished map. The tor flags are rendered before they are read (``_resolved_flag``), so that mid-render map carries the same onion domains the finished one does. """ from plugins.filter.canonical_domains_map import ( FilterModule as _CanonicalDomainsFilter, ) from .applications import get_merged_applications variables = variables or {} resolved_roles_dir = _resolve_roles_dir(roles_dir=roles_dir) cache_key = ( _cache_key(resolved_roles_dir), _stable_variables_signature(variables), bool(getattr(_RENDER_GUARD, "applications", False)), ) cached = _MERGED_DOMAINS_CACHE.get(cache_key) if cached is not None: return cached primary_domain = ( variables.get("DOMAIN_PRIMARY") or variables.get("SYSTEM_EMAIL_DOMAIN") or "" ) if not primary_domain: raise ValueError( "get_merged_domains: DOMAIN_PRIMARY (or SYSTEM_EMAIL_DOMAIN fallback) " "must be set in variables." ) from utils.templating.ansible import render_ansible_strict def _render_raw(raw: Any, name: str) -> Any: if isinstance(raw, str) and ("{{" in raw or "{%" in raw): return render_ansible_strict( templar=None, raw=raw, var_name=name, err_prefix="get_merged_domains", variables={}, ) return raw primary_domain = _render_raw(primary_domain, "DOMAIN_PRIMARY") apps = get_merged_applications( variables=variables, roles_dir=roles_dir, templar=templar, ) filter_instance = _CanonicalDomainsFilter() merged = filter_instance.canonical_domains_map(apps, primary_domain) node_raw = ( ((apps.get("svc-net-tor") or {}).get("services") or {}).get("tor") or {} ).get("node") node_onion = str(_render_raw(node_raw or "", "services.tor.node") or "").strip() group_names = variables.get("group_names") or [] if node_onion and "svc-net-tor" in group_names: merged = _inject_onion_domains( merged, apps, str(primary_domain), node_onion, templar=templar, variables=variables, ) _MERGED_DOMAINS_CACHE[cache_key] = merged return merged def _reset() -> None: _MERGED_DOMAINS_CACHE.clear()