Files
computer-playbook/roles/sys-ctl-hlth-csp/lookup_plugins/csp_accept_status.py
kevinveenbirkenbachandClaude Opus 5 f247b29fd0 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>
2026-09-16 15:19:58 +02:00

84 lines
2.6 KiB
Python

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())
]
]