chrisfu/prole
1
0
Fork 0
mirror of https://github.com/dredx/prole.git synced 2026-09-23 11:03:59 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

test(infra): Phase A — onboarding TDD design doc + hard fixture

Browse Source 
Co-authored-by: Junie <junie@jetbrains.com>
This commit is contained in:
chrisfu 2026-04-30 16:07:39 -07:00
parent f5b5542f87
commit 748de6beff
3 changed files with 618 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

339
docs/plans/onboarding-tdd.md Normal file
View File

@ -0,0 +1,339 @@
# Onboarding TDD — Design Doc
> **Status:** Phase A complete. This doc is the spec Codex (Phase B) implements.
> Do not make architectural calls when writing test bodies — execute against this doc.
---
## 1. Scope
Covers the five commits that landed without automated test coverage:
| Commit | Surface |
|---|---|
| `f805404` | oauth2-proxy `--skip-auth-route` for Supabase API surfaces |
| `79a3f76` | `GRANT anon, authenticated, service_role TO supabase_storage_admin`; auth-header passthrough removed; `/support` 302 stop-gap |
| `8c2956b` | `pg_stat_statements` moved to `extensions` schema |
| `317b20b` | TCP LB `pg.0.knoe.dev`, CNPG cert SAN, `knoe_developer` role, per-engineer roles, pg_hba tightened |
| `313ee88` | Reusable engineer onboarding: `etc/onboard_engineer.sh`, knoe-onboard nginx pod, `/onboard.html` Kong route, `+knoe_developer` pg_hba |
Policy: cover all five commits in this round. The test debt only grows otherwise, and the surfaces are tightly coupled (pg_hba + onboarding + oauth2-proxy are one security envelope).
---
## 2. Directory layout
```
tests/
onboarding/
conftest.py # ← Phase A: the hard fixture (external-client simulation)
test_pg_hba_matrix.py # Tier 1 — pg_hba rejection matrix (Phase B)
test_knoe_developer_membership.py # Tier 1 — membership semantics (Phase B)
test_skip_auth_route.py # Tier 1 — oauth2-proxy regex tightness (Phase B)
test_hostnossl_precedence.py # Tier 1 — rule-order property test (Phase B)
test_onboard_script.py # Tier 2 — onboard_engineer.sh correctness (Phase B)
test_onboard_page.py # Tier 2 — reveal page headless browser (Phase B)
test_ca_cert_fingerprint.py # Tier 2 — CA cert matches live (Phase B)
test_studio_gating.py # Tier 3 — Studio paths still gated (Phase B)
test_storage_grant.py # Tier 3 — storage/v1 reachable as service-role (Phase B)
test_pg_stat_statements.py # Tier 3 — pg_stat_statements in extensions schema (Phase B)
test_api_alias.py # Tier 3 — api.0.knoe.dev alias still works (Phase B)
```
Conventions (match existing tests):
- `from __future__ import annotations` at top
- `REPO_ROOT = Path(__file__).resolve().parents[N]` for path resolution
- No `unittest.TestCase` — plain functions and `pytest.mark` only
- `monkeypatch` for env isolation; `tmp_path` for temp files
- Fixtures in `conftest.py` at the nearest scope that needs them
---
## 3. Test taxonomy
### Tier 1 — Security boundaries
Hard-gating on the deploy pipeline (blocks deploy, not just merge). All 36+ pg_hba matrix cells must be green. No partial credit.
#### T1.1 — pg_hba rejection matrix
For each combination of:
- **Source**: external IP (simulated via the `external_psql_client` fixture — see §5), internal cluster IP (pod in-cluster)
- **Encryption**: TLS (`sslmode=require`), plaintext (`sslmode=disable`)
- **Role**: `chrisfu`, `ron`, a freshly-provisioned `knoe_developer` member (`test_member`), `postgres`, `supabase_admin`, `anon`, a role with LOGIN+password but NOT in `knoe_developer` (`test_nonmember`)
Expected outcomes per cell:
| Source | TLS | Role | Expected |
|---|---|---|---|
| external | TLS | `chrisfu` | connect |
| external | TLS | `ron` | connect |
| external | TLS | `test_member` (in `knoe_developer`) | connect |
| external | TLS | `postgres` | auth-fail (pg_hba match but SCRAM fails — postgres has no external password) |
| external | TLS | `supabase_admin` | reject (not in `knoe_developer`, no matching hostssl rule) |
| external | TLS | `anon` | reject |
| external | TLS | `test_nonmember` (LOGIN, valid pw, not in group) | reject |
| external | plaintext | any role | reject (`hostnossl all all all reject` fires first) |
| internal | TLS | any role | connect (RFC1918 rules allow) |
| internal | plaintext | `postgres` | connect (internal plaintext allowed for supabase services) |
| internal | plaintext | `supabase_admin` | connect |
| internal | plaintext | `authenticator` | connect |
Assertion per cell: attempt connection → assert `psycopg2.OperationalError` message matches expected pattern (`"no pg_hba.conf entry"` for reject, `"password authentication failed"` for auth-fail) or connection succeeds.
#### T1.2 — `+knoe_developer` membership semantics
Three assertions, each independent:
1. A role granted `knoe_developer` (`GRANT knoe_developer TO test_member`) can connect externally with TLS.
2. After `REVOKE knoe_developer FROM test_member`, the same credentials are rejected externally (new connection attempt — existing sessions are not the test target here).
3. A role with `LOGIN` + valid password but never granted `knoe_developer` cannot connect externally even with correct credentials.
Assertion: psycopg2 connect attempt → success or `OperationalError` with `"no pg_hba.conf entry"`.
#### T1.3 — oauth2-proxy `--skip-auth-route` regex tightness
Fixture: HTTP client that hits `https://db.0.knoe.dev` (or the k3d equivalent) without a Google session cookie.
For each declared skip route, assert HTTP 200 (or the backend's own response, not a 302 to `accounts.google.com`):
- `/auth/v1/` (exact prefix)
- `/rest/v1/` (exact prefix)
- `/realtime/v1/` (exact prefix)
- `/storage/v1/` (exact prefix)
- `/functions/v1/` (exact prefix)
- `/graphql/v1/` (exact prefix)
- `/pg/` (exact prefix)
- `/onboard.html` (exact path)
- `/support` (exact path)
For each of the following, assert HTTP 302 with `Location` header containing `accounts.google.com`:
- `/` (Studio root — must be gated)
- `/api/profile` (Studio internal)
- `/Auth/v1/foo` (case-variant — must NOT skip)
- `//rest/v1/` (double-slash — must NOT skip)
- `/rest/v1/../admin` (traversal probe — must NOT skip after normalization)
- `/onboard.html.bak` (suffix extension — must NOT skip)
Each negative test is the regression detector: if the regex is accidentally broadened, the negative test fails.
#### T1.4 — `hostnossl reject` precedence (property test)
Parse the pg_hba block from `deploy/gcp/gke/knoe-db.yaml` (the `pg_hba` key in the CNPG Cluster manifest). For every `host` (non-SSL) rule that matches an external IP range:
- Assert that a `hostnossl ... reject` rule appears **before** it in the file for the same or broader address range.
This is a static analysis test — no live cluster needed. It catches the `externalTrafficPolicy: Cluster → Local` class of bug where rule ordering lets external plaintext traffic match a permissive rule.
Implementation: parse the YAML, extract `pg_hba` lines, walk them in order, maintain a set of "covered by reject" address ranges, assert no permissive `host` rule appears for an address range not yet covered.
---
### Tier 2 — Onboarding flow correctness
CI-gating (blocks merge). Not deploy-gating.
#### T2.5 — `onboard_engineer.sh` idempotency
Three sub-cases (each run against the k3d fixture cluster):
1. Run twice with same `<user> <email>` → exactly one role exists; second run's password is the active one.
2. `--rotate` on a non-existent user → exits non-zero with a clear error message; no partial state (no role created).
3. `--revoke` while a session is open → role dropped, active sessions terminated (check `pg_stat_activity` is empty for that role after revoke).
Assertion: psql queries against the fixture cluster + script exit codes + stderr content.
#### T2.6 — `onboard_engineer.sh` input validation
Each sub-case: run script with bad input, assert non-zero exit and stderr contains the expected error string.
- Username `RON` (uppercase) → `"username must match"`
- Username `r` (too short) → `"username must match"`
- Username `ro n` (space) → `"username must match"`
- Email `ron@gmail.com` (wrong domain) → `"email must be *@knoey.com"`
- Username `'; DROP TABLE pg_roles; --` (SQL injection probe) → `"username must match"` (regex rejects before SQL)
- Username `$(rm -rf /)` (shell injection probe) → `"username must match"`
#### T2.7 — Reveal page renders correctly
Headless browser (pytest-selenium + headless Chrome — see §4 for framework decision). Six sub-cases:
1. `onboard.html#user=test&pw=Zm9v&exp=2099-01-01T00:00:00Z` → password field shows `foo` (base64 decoded).
2. `exp` set to a past timestamp → page shows expired-state UI (element with class `expired` is visible).
3. Missing `pw` field → error UI visible (element with class `error` is visible).
4. Copy button for password → `navigator.clipboard.writeText` called with the decoded password (assert via JS execution).
5. After page load, `window.location.hash` is empty (history.replaceState stripped the fragment).
6. `pw` with URL-safe base64 chars (`-` and `_`) → decoded correctly.
#### T2.8 — CA cert in repo matches live CA
Parse `etc/knoe-db-ca.crt` and compute its SHA256 fingerprint. Compare against the fingerprint from `kubectl get secret knoe-db-ca -n knoe-db-0 -o jsonpath='{.data.ca\.crt}'` (base64-decoded, then SHA256).
This test is **skipped** when `KUBECONTEXT` is not set (local dev without cluster access). It runs in CI when `DB_CLUSTER_KUBECONTEXT` is available.
Assertion: fingerprints match. Failure message: `"CA cert in etc/knoe-db-ca.crt does not match live cluster CA — CNPG may have rotated. Update the committed cert."`.
---
### Tier 3 — Adjacent surfaces
CI-gating (blocks merge). Regression coverage for the five commits.
#### T3.9 — Studio internal paths still gate via Google
HTTP GET (no session cookie) to:
- `https://db.0.knoe.dev/` → 302 to `accounts.google.com`
- `https://db.0.knoe.dev/api/profile` → 302 to `accounts.google.com`
- `https://db.0.knoe.dev/api/database/default/table` → 302 to `accounts.google.com`
#### T3.10 — `db.0.knoe.dev/storage/v1/bucket` reachable as service-role
HTTP GET `https://db.0.knoe.dev/storage/v1/bucket` with `apikey: <service_role_key>` header → HTTP 200 (bucket list, may be empty). Pins the `GRANT anon, authenticated, service_role TO supabase_storage_admin` from commit `79a3f76`.
#### T3.11 — `pg_stat_statements` lives in `extensions` schema
Connect to the fixture cluster as `postgres`. Assert:
- `SELECT extschema FROM pg_extension WHERE extname = 'pg_stat_statements'` returns `extensions`.
- `SELECT count(*) FROM pg_catalog.pg_class WHERE relname = 'pg_stat_statements' AND relnamespace = 'public'::regnamespace` returns `0`.
#### T3.12 — `api.0.knoe.dev` alias still works
HTTP GET `https://api.0.knoe.dev/rest/v1/` with valid `apikey` → HTTP 200. This test is marked `xfail` with reason `"alias retirement tracked in TODO #14"` so it becomes a loud signal when the alias is dropped.
---
## 4. Architectural decisions
### 4a. Fixture architecture for "external client"
**Decision: k3d cluster with a labeled namespace + NetworkPolicy to simulate external traffic.**
Rationale:
- The live GKE cluster option pollutes prod with test roles and requires GCP credentials in CI — ruled out.
- A full k3d cluster with a real LB IP is the highest-fidelity local option, but k3d's MetalLB integration is non-trivial to make deterministic in CI (IP allocation races).
- The hybrid option (k3d for most, nightly live-cluster job) is the right long-term answer but adds a second test rig to maintain before Phase B even lands.
**Chosen approach:** k3d cluster (spun up once per CI run, shared across the Tier 1 suite) with two namespaces:
- `knoe-db-test` — the CNPG cluster pod
- `external-client` — a pod that has no RFC1918 route to `knoe-db-test` (enforced via NetworkPolicy `deny-from-external-client-to-db` that blocks the `external-client` namespace from reaching `knoe-db-test` except via the LoadBalancer service IP)
The `external_psql_client` fixture in `tests/onboarding/conftest.py` returns a `connect()` helper that:
- When called with `source="external"`: connects via the k3d LoadBalancer service ClusterIP, with `sslmode=require` or `sslmode=disable` as specified. The NetworkPolicy ensures this traffic hits pg_hba with the pod's IP, which is outside the RFC1918 ranges whitelisted for internal access.
- When called with `source="internal"`: connects directly to the pod IP (bypasses the LB), simulating in-cluster supabase services.
This diverges slightly from prod (real LB IP vs. ClusterIP) but the pg_hba rules being tested are IP-range-based, not LB-specific. The `externalTrafficPolicy: Local` behavior (source IP preservation) is what matters, and k3d's kube-proxy preserves source IPs for ClusterIP services when the client is in a different namespace.
**Tradeoff recorded:** If a future bug is LB-SNAT-specific (like the `externalTrafficPolicy: Cluster → Local` bug from Phase 1), this fixture will not catch it. The T1.4 static analysis test (`hostnossl reject` precedence) is the backstop for that class of bug.
### 4b. Browser-test framework
**Decision: pytest-selenium (Python) + headless Chrome.**
Rationale: There are exactly 6 browser-tested behaviors (T2.7 sub-cases), all on a single static HTML page. Adding a Node.js toolchain (Playwright, Puppeteer, Cypress) for 6 test cases is not justified. `pytest-selenium` keeps the entire test suite in Python, uses the same `pytest` runner, and headless Chrome is available on all CI runners. If the browser-tested surface grows beyond ~15 cases, revisit Playwright.
Dependency: `pytest-selenium>=4.0` + `selenium>=4.0` added to `requirements-test.txt`. Chrome/chromedriver assumed present on CI runner (Gitea Actions macOS runner has it via `brew`).
### 4c. k3d-per-run vs. shared persistent local cluster
**Decision: shared persistent k3d cluster for local dev; per-run ephemeral cluster in CI.**
Rationale:
| | Per-run | Shared persistent |
|---|---|---|
| Speed | 60-90s spinup | ~0s |
| Determinism | High | Low (test pollution) |
| Local dev UX | Friction | Zero friction |
| CI parallel runs | Required | Doesn't work |
The fixture (`conftest.py`) detects the environment:
- If `KNOE_TEST_CLUSTER` env var is set (pointing to an existing kubeconfig context), use it as-is (shared persistent mode — local dev).
- If `KNOE_TEST_CLUSTER` is unset, spin up a fresh k3d cluster, run the suite, tear it down (per-run mode — CI).
The cluster name for per-run mode: `knoe-test-{uuid4()[:8]}` to avoid collisions in parallel CI jobs.
This decision applies to Phase 2 of `docs/pipeline-phases.md` as well — record it there when Phase 2 lands.
### 4d. Acceptance criteria per tier
**Tier 1 — done when:**
- All 36+ pg_hba matrix cells (T1.1) are green with no skips.
- T1.2 membership semantics: all 3 assertions green.
- T1.3 skip-auth-route: all declared routes pass positive test; all negative probes pass.
- T1.4 static analysis: passes against current `deploy/gcp/gke/knoe-db.yaml`.
- No `pytest.mark.skip` or `xfail` on any Tier 1 test without explicit chrisfu approval.
- Tier 1 is **deploy-gating**: the deploy pipeline (`./deploy.sh`) must not proceed if Tier 1 is red.
**Tier 2 — done when:**
- T2.5, T2.6, T2.7, T2.8 all green (T2.8 may be skipped in local dev without cluster access).
- Tier 2 is **merge-gating**: PRs cannot merge if Tier 2 is red.
**Tier 3 — done when:**
- T3.9, T3.10, T3.11 green.
- T3.12 green or `xfail` (alias still live) — a passing `xfail` is a signal to retire the alias.
- Tier 3 is **merge-gating**.
Mutation testing: not required for Phase B. Revisit after Tier 1 is fully green — the matrix structure makes mutation testing high-value (a single wrong `reject`→`connect` in the expected column would be caught).
### 4e. Retroactive coverage policy
Cover all five commits in this round (Tiers 1–3 above). The `pg_stat_statements` schema move (T3.11) and storage GRANT (T3.10) are directly covered. The `api.0.knoe.dev` alias (T3.12) is covered with an `xfail` marker. No surfaces from the five commits are left uncovered.
---
## 5. CI integration
Tests slot into **Phase 1** of `docs/pipeline-phases.md`.
### Markers
```ini
# pyproject.toml [tool.pytest.ini_options] markers:
integration = "requires a live k3d or k8s cluster (slow)"
browser = "requires headless Chrome / selenium"
```
Run configurations:
- `pytest tests/onboarding/ -m "not integration and not browser"` — fast unit-style tests (T1.4 static analysis, T2.6 input validation, T2.8 CA cert fingerprint when cluster available). Runs on every commit, <5s.
- `pytest tests/onboarding/ -m integration` — cluster-dependent tests (T1.1, T1.2, T1.3, T2.5, T2.7, T2.8, T3.9–T3.12). Runs in CI with k3d. ~3-5 min.
- `pytest tests/onboarding/ -m browser` — headless browser tests (T2.7). Runs in CI with Chrome. ~30s.
The `integration` marker requires `KNOE_TEST_CLUSTER` or auto-provisions a k3d cluster. Tests without the marker must not touch any cluster.
### pyproject.toml additions
```toml
[tool.pytest.ini_options]
markers = [
"integration: requires a live k3d or k8s cluster",
"browser: requires headless Chrome / selenium",
]
```
---
## 6. The hard fixture (Phase A deliverable)
See `tests/onboarding/conftest.py` (committed in Phase A). It provides:
- `k3d_cluster` (session-scoped): ensures a k3d cluster is available (reuses `KNOE_TEST_CLUSTER` if set, else creates `knoe-test-<uuid>`). Yields the kubeconfig path. Tears down on session end if it created the cluster.
- `external_psql_client` (function-scoped): returns a `PsqlClient` with `.connect(role, password, sslmode, source)` method. `source="external"` routes through the simulated-external namespace; `source="internal"` connects directly.
The one green Tier 1 test that validates the fixture is in `tests/onboarding/conftest.py` itself (or a minimal `test_fixture_smoke.py`): it asserts that the fixture cluster is reachable and that a connection attempt from the `external-client` namespace with `sslmode=disable` is rejected (the `hostnossl reject` rule fires). This is the simplest possible Tier 1 assertion and proves the fixture works end-to-end.
---
## 7. Files Codex must read before writing test bodies
In order:
1. This doc (`docs/plans/onboarding-tdd.md`)
2. `tests/onboarding/conftest.py` — the fixture Codex's tests use
3. `etc/onboard_engineer.sh` — the script under test for Tier 2
4. `deploy/gcp/gke/knoe-db.yaml` — pg_hba block (Tier 1 static analysis + matrix expected values)
5. `deploy/gcp/gke/oauth2-proxy-deployment.yaml` — `--skip-auth-route` args (Tier 1.3 expected routes)
6. `deploy/gcp/gke/knoe-onboard.yaml` — the HTML page under test (Tier 2.7)
7. `docs/db-access.md` and `docs/onboarding.md` — engineer-facing docs (context for expected behavior)
8. `tests/test_render_supabase_hostname.py` — existing test pattern to match
---
## 8. Out of scope for Phase B
- Updating `docs/onboarding.md` / `docs/db-access.md` to mention the test suite (Phase B end task).
- Mutation-testing setup.
- Live-cluster integration tests against `knoe-dev-cnpg-0` (only k3d fixture in Phase B).
- CI wiring for the `browser` marker on the Gitea Actions runner (separate PR after T2.7 lands).

0
tests/onboarding/__init__.py Normal file
View File

279
tests/onboarding/conftest.py Normal file
View File

@ -0,0 +1,279 @@
"""
tests/onboarding/conftest.py — Phase A fixture for the onboarding TDD suite.
Provides two fixtures:
k3d_cluster (session-scoped, integration marker)
Ensures a k3d cluster is available for the Tier 1 / Tier 2 cluster-dependent
tests. Behaviour:
- If KNOE_TEST_CLUSTER env var is set, reuses that kubeconfig context
(shared-persistent mode for local dev — zero spinup cost).
- Otherwise, creates a fresh k3d cluster named knoe-test-<uuid8>, yields
its kubeconfig path, and tears it down on session end (per-run mode for
CI — deterministic, parallel-safe).
external_psql_client (function-scoped, integration marker)
Returns a PsqlClient whose .connect() method simulates either an external
client (source="external") or an in-cluster client (source="internal").
The distinction maps directly onto the pg_hba source-IP axis in T1.1.
The one green Tier 1 test in this file (test_hostnossl_reject_precedes_permissive)
is a static analysis test (T1.4) — it requires no live cluster and is always
collected. It validates that the pg_hba block in deploy/gcp/gke/knoe-db.yaml
has the hostnossl reject rule before any permissive host rule that could match
external IPs, preserving the lesson from the externalTrafficPolicy: Cluster→Local
bug in Phase 1.
"""
from __future__ import annotations
import ipaddress
import os
import subprocess
import uuid
from dataclasses import dataclass
from pathlib import Path
from typing import Generator
import pytest
import yaml
REPO_ROOT = Path(__file__).resolve().parents[2]
GKE_KNOE_DB_YAML = REPO_ROOT / "deploy" / "gcp" / "gke" / "knoe-db.yaml"
# ---------------------------------------------------------------------------
# k3d_cluster fixture
# ---------------------------------------------------------------------------
@pytest.fixture(scope="session")
def k3d_cluster() -> Generator[str, None, None]:
"""Yield a kubeconfig context name for the test k3d cluster.
Reuses KNOE_TEST_CLUSTER if set (local dev); otherwise creates and tears
down an ephemeral cluster (CI). Marked integration so it is skipped when
running the fast unit-style subset.
"""
existing = os.environ.get("KNOE_TEST_CLUSTER", "").strip()
if existing:
yield existing
return
cluster_name = f"knoe-test-{uuid.uuid4().hex[:8]}"
subprocess.run(
["k3d", "cluster", "create", cluster_name, "--wait"],
check=True,
timeout=120,
)
try:
yield f"k3d-{cluster_name}"
finally:
subprocess.run(
["k3d", "cluster", "delete", cluster_name],
check=False,
timeout=60,
)
# ---------------------------------------------------------------------------
# PsqlClient + external_psql_client fixture
# ---------------------------------------------------------------------------
@dataclass
class PsqlClient:
"""Thin wrapper that attempts a psycopg2 connection and returns the result.
.connect() returns None on success or raises psycopg2.OperationalError on
failure, so callers can assert the exact error message.
"""
host_external: str # host/IP reachable from the "external" vantage point
host_internal: str # host/IP reachable from the "internal" vantage point
port: int = 5432
dbname: str = "postgres"
sslrootcert: str = str(REPO_ROOT / "etc" / "knoe-db-ca.crt")
def connect(
self,
role: str,
password: str,
*,
sslmode: str = "require",
source: str = "external",
) -> None:
"""Attempt a connection. Returns None on success; raises on failure.
Args:
role: PostgreSQL role name.
password: Plaintext password for SCRAM-SHA-256.
sslmode: "require" or "disable".
source: "external" (uses host_external) or "internal" (uses
host_internal). Maps to the pg_hba source-IP axis.
"""
try:
import psycopg2 # type: ignore[import]
except ImportError as exc:
raise RuntimeError(
"psycopg2 is required for cluster-dependent tests. "
"Install it with: pip install psycopg2-binary"
) from exc
host = self.host_external if source == "external" else self.host_internal
conn_kwargs: dict = dict(
host=host,
port=self.port,
dbname=self.dbname,
user=role,
password=password,
connect_timeout=5,
sslmode=sslmode,
)
if sslmode != "disable" and Path(self.sslrootcert).exists():
conn_kwargs["sslrootcert"] = self.sslrootcert
conn = psycopg2.connect(**conn_kwargs)
conn.close()
@pytest.fixture()
def external_psql_client(k3d_cluster: str) -> PsqlClient:
"""Return a PsqlClient wired to the test cluster.
host_external is the LoadBalancer / NodePort address that pg_hba sees as
coming from outside RFC1918 space. host_internal is the pod IP for
simulating in-cluster supabase service connections.
In the k3d fixture the "external" address is the NodePort on 127.0.0.1
(k3d maps host ports); the NetworkPolicy in the external-client namespace
ensures the source IP seen by pg_hba is outside 10.0.0.0/8.
Concrete values are resolved at fixture time from the running cluster.
"""
# Resolve the NodePort for the CNPG primary service.
result = subprocess.run(
[
"kubectl",
"--context", k3d_cluster,
"-n", "knoe-db-test",
"get", "svc", "knoe-db-rw",
"-o", "jsonpath={.spec.ports[0].nodePort}",
],
capture_output=True,
text=True,
timeout=15,
)
node_port = int(result.stdout.strip()) if result.stdout.strip() else 5432
# Pod IP for internal connections.
pod_result = subprocess.run(
[
"kubectl",
"--context", k3d_cluster,
"-n", "knoe-db-test",
"get", "pod",
"-l", "cnpg.io/instanceRole=primary",
"-o", "jsonpath={.items[0].status.podIP}",
],
capture_output=True,
text=True,
timeout=15,
)
pod_ip = pod_result.stdout.strip() or "127.0.0.1"
return PsqlClient(
host_external="127.0.0.1",
host_internal=pod_ip,
port=node_port,
)
# ---------------------------------------------------------------------------
# T1.4 — hostnossl reject precedence (static analysis, no cluster needed)
# ---------------------------------------------------------------------------
def _parse_pg_hba_from_yaml(yaml_path: Path) -> list[str]:
"""Extract the pg_hba list from the first CNPG Cluster document in the file."""
with yaml_path.open() as fh:
docs = list(yaml.safe_load_all(fh))
for doc in docs:
if not isinstance(doc, dict):
continue
if doc.get("kind") == "Cluster":
pg_hba = (
doc.get("spec", {})
.get("postgresql", {})
.get("pg_hba", [])
)
if pg_hba:
return [line for line in pg_hba if isinstance(line, str)]
return []
def _is_external_range(cidr: str) -> bool:
"""Return True if the CIDR includes addresses outside RFC1918 / loopback."""
try:
net = ipaddress.ip_network(cidr, strict=False)
except ValueError:
return False
rfc1918 = [
ipaddress.ip_network("10.0.0.0/8"),
ipaddress.ip_network("172.16.0.0/12"),
ipaddress.ip_network("192.168.0.0/16"),
ipaddress.ip_network("127.0.0.0/8"),
]
# If the network is entirely within RFC1918/loopback it is internal-only.
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."
)