mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-06 02:54:26 +00:00
kevinveenbirkenbachandClaude Opus 5
5b1ac73dd9
services.tor.enabled is Jinja in 99 of the 105 roles that declare it. A lookup('tls') or lookup('domains') inside a meta/*.yml value runs while the applications render is in flight, the re-entry guard hands it the unrendered tree, and _as_bool read the flag's own source text as False. The onion injection then did nothing for any app and the clearnet domain was returned.
fa3797540a already described this and quarantined it: the render state joined the domains cache key so the poisoned map answers no later read. The value stayed wrong, which is enough until a call site bakes it into applications. roles/web-svc-asset/meta/services.yml:15 is the only one in the repo that does, and in run 34922994823 it froze https://file.infinito.test/assets into the payload while every vhost of that round was published under the node onion. web-app-dashboard then waited 60 times on a host openresty answers with ssl_reject_handshake, and the deploy died on TLSV1_UNRECOGNIZED_NAME.
_resolved_flag renders enabled, exclusive and primary and raises when template text survives, the shape plugins/lookup/proxy_pass.py already uses. plugins/lookup/config.py and plugins/lookup/email.py render before coercing for the same reason; this module was the one that did not.
Measured across the whole map: 44 of 12819 leaves change, none added, none removed, 17 roles. 42 are clearnet to onion. The two others are repairs as well: keycloak's csp frame-ancestors moves from https to http, which is what an onion node serves, and the nextcloud sociallogin provider name stops being a hostname and becomes the bounded name that test_sociallogin_provider_name.py documents.
The new tests drive the same function with Jinja instead of the literal booleans the existing ones pass, which is why those never saw this. Three of the four fail without the fix.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
408 lines
15 KiB
Python
408 lines
15 KiB
Python
"""Focused unit tests for ``utils.cache.domains``.
|
|
|
|
`get_merged_domains` is a thin derivation on top of
|
|
`get_merged_applications` plus the
|
|
`plugins.filter.canonical_domains_map` filter. We pin: cache-keying
|
|
behaviour, the missing-DOMAIN_PRIMARY validation, and the dispatch to
|
|
the upstream applications view.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
import tempfile
|
|
import textwrap
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from utils.cache import _reset_cache_for_tests
|
|
from utils.cache import domains as cache_domains
|
|
from utils.cache.base import _RENDER_GUARD
|
|
from utils.roles.mapping import ROLE_FILE_META_SERVICES, ROLE_FILE_META_USERS
|
|
|
|
from . import PROJECT_ROOT
|
|
|
|
|
|
def _write(path: Path, content: str) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(textwrap.dedent(content), encoding="utf-8")
|
|
|
|
|
|
def _seed_minimal_role(tmp: Path) -> Path:
|
|
role = tmp / "roles" / "web-app-foo"
|
|
_write(
|
|
role / ROLE_FILE_META_SERVICES,
|
|
"""
|
|
server:
|
|
domains:
|
|
canonical:
|
|
- foo.{{ DOMAIN_PRIMARY }}
|
|
aliases: []
|
|
""",
|
|
)
|
|
_write(role / ROLE_FILE_META_USERS, "users: {}\n")
|
|
return tmp / "roles"
|
|
|
|
|
|
class TestMissingPrimaryDomain(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
_reset_cache_for_tests()
|
|
|
|
def test_raises_when_domain_primary_missing(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = _seed_minimal_role(Path(tmp))
|
|
with self.assertRaisesRegex(ValueError, "DOMAIN_PRIMARY"):
|
|
cache_domains.get_merged_domains(
|
|
variables={}, roles_dir=roles, templar=None
|
|
)
|
|
|
|
def test_falls_back_to_system_email_domain(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = _seed_minimal_role(Path(tmp))
|
|
with patch(
|
|
"utils.cache.applications.get_merged_applications",
|
|
return_value={},
|
|
):
|
|
result = cache_domains.get_merged_domains(
|
|
variables={"SYSTEM_EMAIL_DOMAIN": "infinito.test"},
|
|
roles_dir=roles,
|
|
templar=None,
|
|
)
|
|
self.assertIsInstance(result, dict)
|
|
|
|
|
|
class TestCachingPerVariablesSignature(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
_reset_cache_for_tests()
|
|
|
|
def test_caches_result_for_identical_variables(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = _seed_minimal_role(Path(tmp))
|
|
with patch(
|
|
"utils.cache.applications.get_merged_applications",
|
|
return_value={},
|
|
) as mocked:
|
|
first = cache_domains.get_merged_domains(
|
|
variables={"DOMAIN_PRIMARY": "infinito.test"},
|
|
roles_dir=roles,
|
|
templar=None,
|
|
)
|
|
second = cache_domains.get_merged_domains(
|
|
variables={"DOMAIN_PRIMARY": "infinito.test"},
|
|
roles_dir=roles,
|
|
templar=None,
|
|
)
|
|
self.assertEqual(mocked.call_count, 1)
|
|
self.assertEqual(first, second)
|
|
|
|
def test_different_domain_primary_misses_cache(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = _seed_minimal_role(Path(tmp))
|
|
with patch(
|
|
"utils.cache.applications.get_merged_applications",
|
|
return_value={},
|
|
) as mocked:
|
|
cache_domains.get_merged_domains(
|
|
variables={"DOMAIN_PRIMARY": "a.example"},
|
|
roles_dir=roles,
|
|
templar=None,
|
|
)
|
|
cache_domains.get_merged_domains(
|
|
variables={"DOMAIN_PRIMARY": "b.example"},
|
|
roles_dir=roles,
|
|
templar=None,
|
|
)
|
|
self.assertEqual(mocked.call_count, 2)
|
|
|
|
def test_a_map_built_during_the_render_never_becomes_the_finished_one(self):
|
|
"""Under the guard the applications view is still unrendered, so its
|
|
map must not answer a later read; it may still serve the render."""
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = _seed_minimal_role(Path(tmp))
|
|
variables = {"DOMAIN_PRIMARY": "infinito.test"}
|
|
with patch(
|
|
"utils.cache.applications.get_merged_applications",
|
|
return_value={},
|
|
) as mocked:
|
|
_RENDER_GUARD.applications = True
|
|
try:
|
|
cache_domains.get_merged_domains(
|
|
variables=variables, roles_dir=roles, templar=None
|
|
)
|
|
cache_domains.get_merged_domains(
|
|
variables=variables, roles_dir=roles, templar=None
|
|
)
|
|
self.assertEqual(mocked.call_count, 1)
|
|
finally:
|
|
_RENDER_GUARD.applications = False
|
|
cache_domains.get_merged_domains(
|
|
variables=variables, roles_dir=roles, templar=None
|
|
)
|
|
self.assertEqual(mocked.call_count, 2)
|
|
self.assertEqual(len(cache_domains._MERGED_DOMAINS_CACHE), 2)
|
|
self.assertEqual(
|
|
sorted(key[-1] for key in cache_domains._MERGED_DOMAINS_CACHE),
|
|
[False, True],
|
|
)
|
|
|
|
|
|
class TestResetClearsDomainsCache(unittest.TestCase):
|
|
def test_reset_evicts_cached_entry(self):
|
|
_reset_cache_for_tests()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
roles = _seed_minimal_role(Path(tmp))
|
|
with patch(
|
|
"utils.cache.applications.get_merged_applications",
|
|
return_value={},
|
|
):
|
|
cache_domains.get_merged_domains(
|
|
variables={"DOMAIN_PRIMARY": "infinito.test"},
|
|
roles_dir=roles,
|
|
templar=None,
|
|
)
|
|
self.assertEqual(len(cache_domains._MERGED_DOMAINS_CACHE), 1)
|
|
_reset_cache_for_tests()
|
|
self.assertEqual(len(cache_domains._MERGED_DOMAINS_CACHE), 0)
|
|
|
|
|
|
class TestImportableWithoutAnsible(unittest.TestCase):
|
|
"""`utils.cache.domains` MUST stay ansible-free at import time so
|
|
callers in CLI/runner-host paths can pull the module without
|
|
needing ansible. Calling `get_merged_domains` requires ansible
|
|
transitively (via `canonical_domains_map`) and is NOT pinned here
|
|
— the import-only invariant is what matters for the
|
|
ansible-less host. Subprocess isolation avoids the in-process
|
|
sys.modules / namespace-package hazards that bit earlier shims.
|
|
"""
|
|
|
|
def test_module_imports_without_ansible(self):
|
|
import subprocess
|
|
|
|
repo_root = PROJECT_ROOT
|
|
snippet = (
|
|
"import sys\n"
|
|
f"sys.path.insert(0, {str(repo_root)!r})\n"
|
|
"class _Block:\n"
|
|
" def find_spec(self, name, path=None, target=None):\n"
|
|
" if name == 'ansible' or name.startswith('ansible.'):\n"
|
|
" raise ImportError(f'blocked: {name}')\n"
|
|
" return None\n"
|
|
"sys.meta_path.insert(0, _Block())\n"
|
|
"from utils.cache.domains import get_merged_domains\n"
|
|
"assert callable(get_merged_domains)\n"
|
|
"print('OK')\n"
|
|
)
|
|
result = subprocess.run(
|
|
[sys.executable, "-c", snippet],
|
|
capture_output=True,
|
|
text=True,
|
|
cwd=str(repo_root),
|
|
timeout=60,
|
|
check=False,
|
|
)
|
|
self.assertEqual(
|
|
result.returncode,
|
|
0,
|
|
msg=f"stderr=\n{result.stderr}\nstdout=\n{result.stdout}",
|
|
)
|
|
self.assertIn("OK", result.stdout)
|
|
|
|
|
|
class TestOnionDomainInjection(unittest.TestCase):
|
|
ONION = "abc123def456ghij789klmno000pqrstuvwx111yz222abc333def444gh.onion"
|
|
|
|
def _apps(self, **tor):
|
|
return {
|
|
"web-app-x": {
|
|
"services": {
|
|
"tor": {
|
|
"enabled": True,
|
|
"exclusive": False,
|
|
"primary": False,
|
|
**tor,
|
|
}
|
|
},
|
|
}
|
|
}
|
|
|
|
def test_dual_stack_appends_onion(self):
|
|
merged = {"web-app-x": ["x.infinito.test"]}
|
|
out = cache_domains._inject_onion_domains(
|
|
merged, self._apps(), "infinito.test", self.ONION
|
|
)
|
|
self.assertEqual(out["web-app-x"], ["x.infinito.test", f"x.{self.ONION}"])
|
|
|
|
def test_exclusive_replaces_clearnet(self):
|
|
merged = {"web-app-x": ["x.infinito.test"]}
|
|
out = cache_domains._inject_onion_domains(
|
|
merged, self._apps(exclusive=True), "infinito.test", self.ONION
|
|
)
|
|
self.assertEqual(out["web-app-x"], [f"x.{self.ONION}"])
|
|
|
|
def test_primary_puts_onion_first(self):
|
|
merged = {"web-app-x": ["x.infinito.test"]}
|
|
out = cache_domains._inject_onion_domains(
|
|
merged, self._apps(primary=True), "infinito.test", self.ONION
|
|
)
|
|
self.assertEqual(out["web-app-x"], [f"x.{self.ONION}", "x.infinito.test"])
|
|
|
|
def test_disabled_app_untouched(self):
|
|
merged = {"web-app-x": ["x.infinito.test"]}
|
|
apps = {"web-app-x": {"services": {"tor": {"enabled": False}}}}
|
|
out = cache_domains._inject_onion_domains(
|
|
merged, apps, "infinito.test", self.ONION
|
|
)
|
|
self.assertEqual(out["web-app-x"], ["x.infinito.test"])
|
|
|
|
def test_exclusive_primary_default_from_provider(self):
|
|
"""A consumer that omits exclusive/primary inherits the svc-net-tor
|
|
provider defaults (exclusive: true -> onion only)."""
|
|
merged = {"web-app-x": ["x.infinito.test"]}
|
|
apps = {
|
|
"web-app-x": {"services": {"tor": {"enabled": True, "shared": True}}},
|
|
"svc-net-tor": {
|
|
"services": {
|
|
"tor": {
|
|
"enabled": True,
|
|
"shared": True,
|
|
"exclusive": True,
|
|
"primary": True,
|
|
}
|
|
}
|
|
},
|
|
}
|
|
out = cache_domains._inject_onion_domains(
|
|
merged, apps, "infinito.test", self.ONION
|
|
)
|
|
self.assertEqual(out["web-app-x"], [f"x.{self.ONION}"])
|
|
|
|
def test_consumer_override_beats_provider_default(self):
|
|
"""A consumer that pins exclusive: false overrides the provider's
|
|
exclusive: true and stays dual-stack."""
|
|
merged = {"web-app-x": ["x.infinito.test"]}
|
|
apps = {
|
|
"web-app-x": {
|
|
"services": {
|
|
"tor": {"enabled": True, "shared": True, "exclusive": False}
|
|
}
|
|
},
|
|
"svc-net-tor": {
|
|
"services": {
|
|
"tor": {
|
|
"enabled": True,
|
|
"shared": True,
|
|
"exclusive": True,
|
|
"primary": False,
|
|
}
|
|
}
|
|
},
|
|
}
|
|
out = cache_domains._inject_onion_domains(
|
|
merged, apps, "infinito.test", self.ONION
|
|
)
|
|
self.assertEqual(out["web-app-x"], ["x.infinito.test", f"x.{self.ONION}"])
|
|
|
|
def test_consumer_exclusive_false_inherits_provider_primary_dual(self):
|
|
"""A consumer pinning only exclusive: false inherits the provider's
|
|
primary: true and stays dual-stack, onion-first."""
|
|
merged = {"web-app-x": ["x.infinito.test"]}
|
|
apps = {
|
|
"web-app-x": {
|
|
"services": {
|
|
"tor": {"enabled": True, "shared": True, "exclusive": False}
|
|
}
|
|
},
|
|
"svc-net-tor": {
|
|
"services": {
|
|
"tor": {
|
|
"enabled": True,
|
|
"shared": True,
|
|
"exclusive": True,
|
|
"primary": True,
|
|
}
|
|
}
|
|
},
|
|
}
|
|
out = cache_domains._inject_onion_domains(
|
|
merged, apps, "infinito.test", self.ONION
|
|
)
|
|
self.assertEqual(out["web-app-x"], [f"x.{self.ONION}", "x.infinito.test"])
|
|
|
|
def test_bare_primary_domain_maps_to_node_onion(self):
|
|
merged = {"web-app-x": ["infinito.test"]}
|
|
out = cache_domains._inject_onion_domains(
|
|
merged, self._apps(), "infinito.test", self.ONION
|
|
)
|
|
self.assertEqual(out["web-app-x"], ["infinito.test", self.ONION])
|
|
|
|
|
|
class TestOnionFlagsAreRendered(unittest.TestCase):
|
|
"""`services.tor.*` reaches this map as Jinja in 99 of the 105 roles that
|
|
declare it, and as its own source text whenever the applications render is
|
|
still in flight. Reading that text as a boolean answers False and drops the
|
|
onion domains for every app, which is what froze a clearnet asset URL into
|
|
the applications payload of run 34922994823.
|
|
"""
|
|
|
|
ONION = TestOnionDomainInjection.ONION
|
|
JINJA_ON = "{{ 'svc-net-tor' in group_names }}"
|
|
|
|
_apps = TestOnionDomainInjection._apps
|
|
|
|
def _templar(self, **variables):
|
|
from ansible.parsing.dataloader import DataLoader
|
|
from ansible.template import Templar
|
|
|
|
templar = Templar(loader=DataLoader(), variables=dict(variables))
|
|
templar.available_variables = dict(variables)
|
|
return templar
|
|
|
|
def _inject(self, apps, **variables):
|
|
return cache_domains._inject_onion_domains(
|
|
{"web-app-x": ["x.infinito.test"]},
|
|
apps,
|
|
"infinito.test",
|
|
self.ONION,
|
|
templar=self._templar(**variables),
|
|
variables=dict(variables),
|
|
)
|
|
|
|
def test_a_templated_enabled_flag_still_injects_the_onion(self):
|
|
out = self._inject(
|
|
self._apps(enabled=self.JINJA_ON),
|
|
group_names=["svc-net-tor", "web-app-x"],
|
|
)
|
|
|
|
self.assertEqual(out["web-app-x"], ["x.infinito.test", f"x.{self.ONION}"])
|
|
|
|
def test_a_templated_flag_resolving_false_leaves_the_clearnet_domain(self):
|
|
out = self._inject(self._apps(enabled=self.JINJA_ON), group_names=["web-app-x"])
|
|
|
|
self.assertEqual(out["web-app-x"], ["x.infinito.test"])
|
|
|
|
def test_templated_exclusive_replaces_the_clearnet_domain(self):
|
|
out = self._inject(
|
|
self._apps(enabled=self.JINJA_ON, exclusive=self.JINJA_ON),
|
|
group_names=["svc-net-tor", "web-app-x"],
|
|
)
|
|
|
|
self.assertEqual(out["web-app-x"], [f"x.{self.ONION}"])
|
|
|
|
def test_a_flag_that_cannot_be_rendered_is_an_error(self):
|
|
"""Without a templar the text cannot resolve, and answering False
|
|
would drop the onion domains without saying so."""
|
|
with self.assertRaisesRegex(ValueError, "services.tor.enabled"):
|
|
cache_domains._inject_onion_domains(
|
|
{"web-app-x": ["x.infinito.test"]},
|
|
self._apps(enabled=self.JINJA_ON),
|
|
"infinito.test",
|
|
self.ONION,
|
|
templar=None,
|
|
variables={"group_names": ["svc-net-tor"]},
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|