diff --git a/tests/onboarding/conftest.py b/tests/onboarding/conftest.py index 32bc86f..4e2f9f5 100644 --- a/tests/onboarding/conftest.py +++ b/tests/onboarding/conftest.py @@ -223,57 +223,3 @@ def _is_external_range(cidr: str) -> bool: return not any(net.subnet_of(private) for private in rfc1918) -def test_hostnossl_reject_precedes_permissive_host_rules() -> None: - """T1.4 — hostnossl reject must appear before any permissive host rule - that could match external IPs. - - Regression detector for the externalTrafficPolicy: Cluster→Local bug from - Phase 1: if a permissive `host` rule (non-SSL, non-reject) for an external - CIDR appears before the `hostnossl ... reject` line, an external plaintext - connection could match the permissive rule and bypass the TLS requirement. - - This is a static analysis test — no live cluster required. It runs on - every commit as part of the fast (non-integration) subset. - """ - if not GKE_KNOE_DB_YAML.exists(): - pytest.skip(f"pg_hba source not found: {GKE_KNOE_DB_YAML}") - - rules = _parse_pg_hba_from_yaml(GKE_KNOE_DB_YAML) - assert rules, f"No pg_hba rules found in {GKE_KNOE_DB_YAML}" - - hostnossl_reject_seen = False - violations: list[str] = [] - - for rule in rules: - tokens = rule.split() - if not tokens or tokens[0].startswith("#"): - continue - - conn_type = tokens[0].lower() - - # Track when we see a hostnossl ... reject line covering 0.0.0.0/0 - if conn_type == "hostnossl" and tokens[-1].lower() == "reject": - # Any hostnossl reject counts — we're conservative here. - hostnossl_reject_seen = True - continue - - # A permissive `host` rule (not hostssl, not local, not reject) that - # covers an external CIDR before hostnossl reject has been seen is a - # violation. - if conn_type == "host" and not hostnossl_reject_seen: - # tokens: conn_type db user address [mask] auth-method [options] - # address is tokens[3] for the standard 4-field form. - if len(tokens) >= 5: - address = tokens[3] - auth_method = tokens[-1].lower() - if auth_method != "reject" and _is_external_range(address): - violations.append(rule) - - assert not violations, ( - "pg_hba has permissive `host` rule(s) for external IPs appearing BEFORE " - "the `hostnossl ... reject` line. An external plaintext connection could " - "match these rules and bypass the TLS requirement.\n\n" - "Violating rules:\n" + "\n".join(f" {r}" for r in violations) + "\n\n" - "Fix: move `hostnossl all all 0.0.0.0/0 reject` above any permissive " - "`host` rule that covers non-RFC1918 addresses." - ) diff --git a/tests/onboarding/test_hostnossl_precedence.py b/tests/onboarding/test_hostnossl_precedence.py new file mode 100644 index 0000000..955d9d8 --- /dev/null +++ b/tests/onboarding/test_hostnossl_precedence.py @@ -0,0 +1,62 @@ +""" +T1.4 — hostnossl reject precedence (static analysis, no cluster needed). + +Regression detector for the externalTrafficPolicy: Cluster→Local bug from +Phase 1. See docs/plans/onboarding-tdd.md §3 T1.4 for the full spec. +""" +from __future__ import annotations + +from tests.onboarding.conftest import ( + GKE_KNOE_DB_YAML, + _is_external_range, + _parse_pg_hba_from_yaml, +) + +import pytest + + +def test_hostnossl_reject_precedes_permissive_host_rules() -> None: + """T1.4 — hostnossl reject must appear before any permissive host rule + that could match external IPs. + + If a permissive `host` rule (non-SSL, non-reject) for an external CIDR + appears before the `hostnossl ... reject` line, an external plaintext + connection could match the permissive rule and bypass the TLS requirement. + + Static analysis — no live cluster required. Runs on every commit. + """ + if not GKE_KNOE_DB_YAML.exists(): + pytest.skip(f"pg_hba source not found: {GKE_KNOE_DB_YAML}") + + rules = _parse_pg_hba_from_yaml(GKE_KNOE_DB_YAML) + assert rules, f"No pg_hba rules found in {GKE_KNOE_DB_YAML}" + + hostnossl_reject_seen = False + violations: list[str] = [] + + for rule in rules: + tokens = rule.split() + if not tokens or tokens[0].startswith("#"): + continue + + conn_type = tokens[0].lower() + + if conn_type == "hostnossl" and tokens[-1].lower() == "reject": + hostnossl_reject_seen = True + continue + + if conn_type == "host" and not hostnossl_reject_seen: + if len(tokens) >= 5: + address = tokens[3] + auth_method = tokens[-1].lower() + if auth_method != "reject" and _is_external_range(address): + violations.append(rule) + + assert not violations, ( + "pg_hba has permissive `host` rule(s) for external IPs appearing BEFORE " + "the `hostnossl ... reject` line. An external plaintext connection could " + "match these rules and bypass the TLS requirement.\n\n" + "Violating rules:\n" + "\n".join(f" {r}" for r in violations) + "\n\n" + "Fix: move `hostnossl all all 0.0.0.0/0 reject` above any permissive " + "`host` rule that covers non-RFC1918 addresses." + )