feat(hlth-csp): accept a declared status code instead of dropping the vhost

The probe dropped every canonical and alias domain of a role whose server.status_codes.default named anything at or above 400, so a single by-design 4xx removed the whole vhost from CSP inspection. The checker now takes --accept-status <domain>=<code>, which keeps the page navigated and its policy evaluated, so the declaration buys coverage instead of costing it.

csp_accept_status reads the existing SPOT in utils/roles/applications/status_codes.py rather than restating the rule. That resolves per vhost, because server.status_codes is keyed like domains.canonical, so two domains of one role can differ; the old filter looked only at default and was blind to seaweedfs declaring its api, filer and master keys separately. The threshold is the checker's own: anything below 300 passes unflagged, so a declared redirect is named too.

csp_skip_domains keeps only the half that has no vhost to probe at all, a canonical domain owned by a disabled service, and gains the tests that half never had.

Measured: 14 domains move from skipped to probed, among them jenkins, mattermost, n8n, prometheus, mirror and xmpp. 28 domains carry an entry in total. Across the CI diagnostics of this run 473 of 476 observed responses were 200, the single infinito vhost answering 403 is api.seaweedfs, and it declares that code, so it stays green.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

roles/sys-ctl-hlth-csp/lookup_plugins/csp_accept_status.py

@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+from ansible.plugins.loader import lookup_loader
+from ansible.plugins.lookup import LookupBase
+
+from utils.roles.applications.config import get
+from utils.roles.applications.status_codes import declared_status_codes
+
+_CHECKER_ACCEPTS_BELOW = 300
+
+
+def _domains(value: Any) -> list[str]:
+    if isinstance(value, str):
+        return [value] if value else []
+    if isinstance(value, Mapping):
+        out: list[str] = []
+        for entry in value.values():
+            out.extend(_domains(entry))
+        return out
+    if isinstance(value, (list, tuple, set)):
+        out = []
+        for entry in value:
+            out.extend(_domains(entry))
+        return out
+    return []
+
+
+class LookupModule(LookupBase):
+    """Return ``<domain>=<code>[,<code>]`` for every declared non-2xx status.
+
+    The checker treats anything outside 2xx as unreachable, so a vhost that
+    serves a 4xx by design needs its code named. Naming it keeps the page
+    under CSP inspection, which excluding the domain would not.
+
+    The declaration is per vhost: ``server.status_codes`` is keyed like
+    ``domains.canonical``, so two domains of one role can differ.
+    """
+
+    def run(
+        self,
+        terms: list[Any] | None,
+        variables: dict[str, Any] | None = None,
+        **kwargs: Any,
+    ) -> list[list[str]]:
+        applications = lookup_loader.get(
+            "applications",
+            loader=self._loader,
+            templar=getattr(self, "_templar", None),
+        ).run(
+            [],
+            variables=variables
+            or getattr(self._templar, "available_variables", {})
+            or {},
+        )[0]
+
+        if not isinstance(applications, Mapping):
+            return [[]]
+
+        accepted: dict[str, set[int]] = {}
+        for app_id in applications:
+            domains = _domains(
+                get(applications, app_id, "domains.canonical", strict=False, default=[])
+            ) + _domains(
+                get(applications, app_id, "domains.aliases", strict=False, default=[])
+            )
+            for domain in domains:
+                codes = {
+                    code
+                    for code in declared_status_codes(applications, app_id, domain)
+                    if code >= _CHECKER_ACCEPTS_BELOW
+                }
+                if codes:
+                    accepted.setdefault(domain, set()).update(codes)
+
+        return [
+            [
+                f"{domain}={','.join(str(code) for code in sorted(codes))}"
+                for domain, codes in sorted(accepted.items())
+            ]
+        ]

roles/sys-ctl-hlth-csp/lookup_plugins/csp_skip_domains.py

@@ -32,21 +32,6 @@ def _to_list(x: Any) -> list[str]:
     return []
 
 
-def _has_code_ge_400(codes: Any) -> bool:
-    if codes is None:
-        return False
-    if not isinstance(codes, (list, tuple, set)):
-        codes = [codes]
-    for c in codes:
-        try:
-            n = int(c)
-        except (TypeError, ValueError):
-            continue
-        if n >= 400:
-            return True
-    return False
-
-
 def _selection_from(group_names: Any) -> set[str]:
     if isinstance(group_names, (list, set, tuple)):
         return {str(x) for x in group_names if str(x)}
@@ -58,11 +43,7 @@ def _selection_from(group_names: Any) -> set[str]:
 class LookupModule(LookupBase):
     """Return domains the CSP probe should skip.
 
-    Skips canonical + alias domains of every application whose
-    ``server.status_codes.default`` declares any HTTP code >= 400
-    (e.g. federation-only roles that legitimately serve 4xx at ``/``).
-
-    Additionally skips canonical domains owned by a disabled service: a
+    Skips canonical domains owned by a disabled service: a
     service entry may declare ``domains: [<canonical key>, ...]``; when its
     ``enabled`` resolves falsy those canonicals get no vhost (e.g. the
     seaweedfs filer/master frontend on an onion node), so probing them can
@@ -121,32 +102,4 @@ class LookupModule(LookupBase):
             if not selection or app_id in selection:
                 skip |= self._service_disabled_domains(applications, app_id)
 
-            default = get(
-                applications,
-                app_id,
-                "server.status_codes.default",
-                strict=False,
-                default=None,
-            )
-            if not _has_code_ge_400(default):
-                continue
-
-            canonical = get(
-                applications,
-                app_id,
-                "domains.canonical",
-                strict=False,
-                default=[],
-            )
-            aliases = get(
-                applications,
-                app_id,
-                "domains.aliases",
-                strict=False,
-                default=[],
-            )
-            for d in _to_list(canonical) + _to_list(aliases):
-                if d:
-                    skip.add(d)
-
         return [sorted(skip)]

roles/sys-ctl-hlth-csp/meta/services.yml

@@ -8,4 +8,4 @@ csp:
       categories: [web]
 csp-checker:
   image: ghcr.io/kevinveenbirkenbach/csp-checker
-  version: 2.3.0
+  version: 3.0.0

roles/sys-ctl-hlth-csp/tasks/00_core.yml

@@ -60,6 +60,10 @@
             '--skip-domain ' ~ (HEALTH_CSP_SKIP_DOMAINS | join(' ')),
             ''
           ),
+          (HEALTH_CSP_ACCEPT_STATUS | length > 0) | ternary(
+            '--accept-status ' ~ (HEALTH_CSP_ACCEPT_STATUS | join(' ')),
+            ''
+          ),
           (lookup('config', 'svc-net-tor', 'services.tor.node', '') | length > 0) | ternary(
             '--tor-proxy=socks5://' ~ lookup('tor_socks')
             ~ ' --onion-timeout=' ~ HEALTH_CSP_ONION_TIMEOUT_MS,

roles/sys-ctl-hlth-csp/vars/main.yml

@@ -3,6 +3,7 @@ system_service_id: sys-ctl-hlth-csp
 HEALTH_CSP_ONION_TIMEOUT_MS: "{{ (HEALTH_CSP_TIMEOUT_MS | int) * (lookup('config', 'svc-net-tor', 'services.tor.timeout_multipliers')[lookup('config', 'svc-net-tor',
   'services.tor.flavor')] | int) }}"
 HEALTH_CSP_ONION_DOMAIN_SECONDS: "{{ ((HEALTH_CSP_ONION_TIMEOUT_MS | int) // 1000) + (HEALTH_CSP_DOMAIN_RESERVE_SECONDS | int) }}"
+HEALTH_CSP_ACCEPT_STATUS: "{{ lookup('csp_accept_status') }}"
 HEALTH_CSP_SKIP_DOMAINS: >-
   {{ (
        lookup('csp_skip_domains', group_names=lookup('deployment').deployed)

tests/unit/python/roles/sys-ctl-hlth-csp/lookup_plugins/test_csp_accept_status.py

@@ -0,0 +1,124 @@
+from __future__ import annotations
+
+import importlib.util
+import sys
+import unittest
+import unittest.mock as mock
+from unittest.mock import patch
+
+from . import PROJECT_ROOT
+
+
+def _load_module(rel_path: str, name: str):
+    path = PROJECT_ROOT / rel_path
+    spec = importlib.util.spec_from_file_location(name, str(path))
+    mod = importlib.util.module_from_spec(spec)
+    assert spec and spec.loader
+    sys.modules[name] = mod
+    spec.loader.exec_module(mod)
+    return mod
+
+
+class _DummyTemplar:
+    def __init__(self, available_variables: dict | None = None):
+        self.available_variables = available_variables or {}
+
+
+class CspAcceptStatusLookupTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.module = _load_module(
+            "roles/sys-ctl-hlth-csp/lookup_plugins/csp_accept_status.py",
+            "csp_accept_status_lookup",
+        )
+
+    def _run(self, applications, **kwargs):
+        lm = self.module.LookupModule()
+        lm._templar = _DummyTemplar()
+        lm._loader = mock.MagicMock()
+        with patch.object(self.module, "lookup_loader") as loader_mock:
+            loader_mock.get.return_value = mock.MagicMock(
+                run=lambda *_a, **_k: [applications]
+            )
+            return lm.run(terms=[], variables={}, **kwargs)[0]
+
+    def test_empty_applications_returns_empty_list(self):
+        self.assertEqual(self._run({}), [])
+
+    def test_non_mapping_applications_returns_empty(self):
+        self.assertEqual(self._run([]), [])
+
+    def test_a_declared_4xx_is_named_with_its_domain(self):
+        apps = {
+            "web-svc-mirror": {
+                "domains": {"canonical": ["mirror.example.com"]},
+                "server": {"status_codes": {"default": [404]}},
+            }
+        }
+        self.assertEqual(self._run(apps), ["mirror.example.com=404"])
+
+    def test_aliases_are_named_too(self):
+        apps = {
+            "web-app-bridgy": {
+                "domains": {
+                    "canonical": ["bridgy.example.com"],
+                    "aliases": ["fed.example.com"],
+                },
+                "server": {"status_codes": {"default": [200, 404]}},
+            }
+        }
+        self.assertEqual(
+            self._run(apps),
+            ["bridgy.example.com=404", "fed.example.com=404"],
+        )
+
+    def test_a_2xx_only_declaration_names_nothing(self):
+        apps = {
+            "web-app-foo": {
+                "domains": {"canonical": ["foo.example.com"]},
+                "server": {"status_codes": {"default": [200]}},
+            }
+        }
+        self.assertEqual(self._run(apps), [])
+
+    def test_a_redirect_is_named_because_the_checker_rejects_it(self):
+        apps = {
+            "web-app-foo": {
+                "domains": {"canonical": ["foo.example.com"]},
+                "server": {"status_codes": {"default": [200, 301]}},
+            }
+        }
+        self.assertEqual(self._run(apps), ["foo.example.com=301"])
+
+    def test_two_vhosts_of_one_role_keep_their_own_codes(self):
+        apps = {
+            "web-app-split": {
+                "domains": {
+                    "canonical": {
+                        "api": "api.example.com",
+                        "web": "web.example.com",
+                    }
+                },
+                "server": {
+                    "status_codes": {
+                        "api": [405],
+                        "web": [200],
+                        "default": [200],
+                    }
+                },
+            }
+        }
+        self.assertEqual(self._run(apps), ["api.example.com=405"])
+
+    def test_several_codes_of_one_domain_are_joined(self):
+        apps = {
+            "web-app-foo": {
+                "domains": {"canonical": ["foo.example.com"]},
+                "server": {"status_codes": {"default": [200, 403, 404]}},
+            }
+        }
+        self.assertEqual(self._run(apps), ["foo.example.com=403,404"])
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)

tests/unit/python/roles/sys-ctl-hlth-csp/lookup_plugins/test_csp_skip_domains.py

@@ -24,6 +24,13 @@ class _DummyTemplar:
         self.available_variables = available_variables or {}
 
 
+def _app(service_enabled, domain="filer.example.com"):
+    return {
+        "server": {"domains": {"canonical": {"filer": domain}}},
+        "services": {"filer": {"enabled": service_enabled, "domains": ["filer"]}},
+    }
+
+
 class CspSkipDomainsLookupTests(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
@@ -45,89 +52,47 @@ class CspSkipDomainsLookupTests(unittest.TestCase):
     def test_empty_applications_returns_empty_list(self):
         self.assertEqual(self._run({}), [])
 
-    def test_app_with_status_code_ge_400_skips_canonical_and_aliases(self):
-        apps = {
-            "web-app-bridgy": {
-                "domains": {
-                    "canonical": ["bridgy.example.com"],
-                    "aliases": ["fed.example.com"],
-                },
-                "server": {
-                    "status_codes": {"default": [200, 404]},
-                },
-            }
-        }
-        self.assertEqual(self._run(apps), ["bridgy.example.com", "fed.example.com"])
+    def test_non_mapping_applications_returns_empty(self):
+        self.assertEqual(self._run([]), [])
 
-    def test_app_without_status_code_ge_400_is_not_skipped(self):
+    def test_a_disabled_service_hides_its_canonical_domain(self):
+        self.assertEqual(
+            self._run({"web-app-seaweedfs": _app(False)}),
+            ["filer.example.com"],
+        )
+
+    def test_an_enabled_service_keeps_its_domain_under_probe(self):
+        self.assertEqual(self._run({"web-app-seaweedfs": _app(True)}), [])
+
+    def test_a_service_without_domains_is_ignored(self):
         apps = {
             "web-app-foo": {
-                "domains": {"canonical": ["foo.example.com"]},
-                "server": {
-                    "status_codes": {"default": [200]},
-                },
+                "server": {"domains": {"canonical": {"web": "foo.example.com"}}},
+                "services": {"worker": {"enabled": False}},
             }
         }
         self.assertEqual(self._run(apps), [])
 
-    def test_app_without_status_codes_field_is_not_skipped(self):
-        apps = {"web-app-foo": {"domains": {"canonical": ["foo.example.com"]}}}
-        self.assertEqual(self._run(apps), [])
-
-    def test_a_declared_4xx_is_honoured_outside_the_selection(self):
+    def test_the_selection_filters_which_apps_are_inspected(self):
         apps = {
-            "web-app-bridgy": {
-                "domains": {"canonical": ["bridgy.example.com"]},
-                "server": {
-                    "status_codes": {"default": [404]},
-                },
-            },
-            "web-svc-mirror": {
-                "domains": {"canonical": ["mirror.example.com"]},
-                "server": {
-                    "status_codes": {"default": [404]},
-                },
-            },
+            "web-app-a": _app(False, "a.example.com"),
+            "web-app-b": _app(False, "b.example.com"),
         }
         self.assertEqual(
-            self._run(apps, group_names=["web-app-bridgy"]),
-            ["bridgy.example.com", "mirror.example.com"],
+            self._run(apps, group_names=["web-app-a"]),
+            ["a.example.com"],
         )
 
     def test_group_names_csv_string_is_accepted(self):
         apps = {
-            "a": {
-                "domains": {"canonical": ["a.example.com"]},
-                "server": {
-                    "status_codes": {"default": [404]},
-                },
-            },
-            "b": {
-                "domains": {"canonical": ["b.example.com"]},
-                "server": {
-                    "status_codes": {"default": [404]},
-                },
-            },
+            "web-app-a": _app(False, "a.example.com"),
+            "web-app-b": _app(False, "b.example.com"),
         }
         self.assertEqual(
-            self._run(apps, group_names="a,b"),
+            self._run(apps, group_names="web-app-a,web-app-b"),
             ["a.example.com", "b.example.com"],
         )
 
-    def test_non_4xx_codes_dont_count(self):
-        apps = {
-            "web-app-foo": {
-                "domains": {"canonical": ["foo.example.com"]},
-                "server": {
-                    "status_codes": {"default": [200, 301, 302]},
-                },
-            }
-        }
-        self.assertEqual(self._run(apps), [])
-
-    def test_non_mapping_applications_returns_empty(self):
-        self.assertEqual(self._run([]), [])
-
 
 if __name__ == "__main__":
     unittest.main(verbosity=2)