From 71246722ae11a29e4a84f9f650f909d0684dbbfb Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Fri, 14 Aug 2026 18:33:35 +0700 Subject: [PATCH] feat(compliance): vendor + subprocessor register with usage reconciliation (CC9.2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit R8 soc2-groundwork task 3 of 8. A read+report-only tool that produces an auditable subprocessor register: every declared third-party processor of customer data, classified documented / incomplete / dpa_expired / unused — AND reconciled against what the system actually reaches. The reconciliation is the anti-theater spine: a register that is only a hand-maintained doc silently drifts from reality, so build_register cross-checks the declared list against a live UsageSource. A processor the system reaches but that is absent from the declared register is a HARD finding, never a silent omission (M4, R:UNDECLARED_PROCESSOR) — declaring a processor (incl. non- customer-data infra) is what suppresses that finding. Design (mirrors the CC8.1 change-evidence and CC6 access-review tools beside it): - PURE core: build_register(source, declared, *, now_iso) over a UsageSource port. now_iso is INJECTED (no clock), so identical declared + usage + now yields byte-identical output. This is the layer the suite drives. - IO adapter ConfigUsageSource is the only thing that enumerates live usage, derived deterministically from the deploy surface (provider registry, egress allow-list, storage config) — NOT the network. A partial/unreadable source raises IncompleteFetch so the core never reconciles against an incomplete usage set (which would hide an undeclared processor). Live enumeration is a documented NotImplementedError the operator wires. Fail-closed classification (all red-first tested, 10/10): - A vendor missing a required field or a signed, in-window DPA is never documented — it fails closed to incomplete (M3, A4, R:ASSUMED_COMPLIANT). - A DPA whose expiry is on/before the injected now_iso is dpa_expired, computed not assumed (M5). - A non-customer-data processor carries no DPA burden — listed for reconciliation but NOT risk-classified (A2). - The review record reads "unreviewed — draft" until a real human signs; the tool never fabricates a reviewer (M6, R:FABRICATED_SIGNOFF). - No write path exists — the port's only method is the read `processors`, and the tool never edits subprocessors.json (M1, R:TOOL_WRITES_REGISTER). scripts/soc2/ is repo-governance tooling, NOT part of the gateway wheel. The declared subprocessors.json ships with placeholder DPA dates to be verified by a human as vendor agreements are signed. ADD gate PASS (contract sha256:a593b6e1); ruff clean. author: Tin Dang --- .../vendor-subprocessor-register.d/runs/1.md | 26 ++ .add/tasks/vendor-subprocessor-register.md | 89 +++++ .../test_vendor_register.py | 186 ++++++++++ scripts/soc2/subprocessors.json | 41 +++ scripts/soc2/vendor_register.py | 320 ++++++++++++++++++ 5 files changed, 662 insertions(+) create mode 100644 .add/tasks/vendor-subprocessor-register.d/runs/1.md create mode 100644 .add/tasks/vendor-subprocessor-register.md create mode 100644 apps/gateway/tests/soc2_vendor_register/test_vendor_register.py create mode 100644 scripts/soc2/subprocessors.json create mode 100644 scripts/soc2/vendor_register.py diff --git a/.add/tasks/vendor-subprocessor-register.d/runs/1.md b/.add/tasks/vendor-subprocessor-register.d/runs/1.md new file mode 100644 index 00000000..e1858e6d --- /dev/null +++ b/.add/tasks/vendor-subprocessor-register.d/runs/1.md @@ -0,0 +1,26 @@ +--- +type: Run +runtime: process +task: /tasks/vendor-subprocessor-register.md +computation: "env GATEWAY_TEST_SKIP_INFRA_CHECK=1 uv run pytest tests/soc2_vendor_register/ --no-cov -p no:randomly --junitxml /private/tmp/claude-501/-Users-tindang-workspaces-tind-repo-ai-proxy/88454d31-bfec-4421-87a2-2d625a0ab229/scratchpad/vendor_junit.xml" +receipt: + kind: test-ids + ids: 10/10 reported + exit: 0 + freshness: mtime + at: 2026-08-14 + stdout: '============================== 10 passed in 0.16s ==============================' + note: '' + passed: + - tests.soc2_vendor_register.test_vendor_register::test_build_is_pure_deterministic_and_ordered + - tests.soc2_vendor_register.test_vendor_register::test_declared_but_unused_flagged_not_hard + - tests.soc2_vendor_register.test_vendor_register::test_expired_dpa_flagged_against_injected_now + - tests.soc2_vendor_register.test_vendor_register::test_incomplete_fetch_fails_not_truncates + - tests.soc2_vendor_register.test_vendor_register::test_missing_dpa_never_documented_fails_closed + - tests.soc2_vendor_register.test_vendor_register::test_non_customer_data_vendor_listed_not_classified + - tests.soc2_vendor_register.test_vendor_register::test_review_record_unreviewed_until_signed + - tests.soc2_vendor_register.test_vendor_register::test_summary_is_payload_free_and_counted + - tests.soc2_vendor_register.test_vendor_register::test_tool_has_no_write_path + - tests.soc2_vendor_register.test_vendor_register::test_used_but_undeclared_is_hard_finding +generated: { by: process:run, at: 2026-08-14 } +--- diff --git a/.add/tasks/vendor-subprocessor-register.md b/.add/tasks/vendor-subprocessor-register.md new file mode 100644 index 00000000..4b44153d --- /dev/null +++ b/.add/tasks/vendor-subprocessor-register.md @@ -0,0 +1,89 @@ +--- +type: Task +title: vendor-subprocessor-register +status: done +gives: + - S1 build_register + rendered subprocessor register (CC9.2 vendor risk) +generated: { by: add/3.2.0, at: 2026-08-14 } +verified: + - { by: "cli", at: 2026-08-14, act: freeze, authority: process, direction: "sha256:a593b6e1b13485e4" } + - { by: "cli", at: 2026-08-14, act: brief, authority: process, brief: "sha256:a87d0a63eb722c4f" } + - { by: "process:run", at: 2026-08-14, act: run, authority: process, outcome: PASS, receipt: /tasks/vendor-subprocessor-register.d/runs/1.md } + - { by: "process:verify", at: 2026-08-14, act: gate, authority: process, outcome: PASS, receipt: /tasks/vendor-subprocessor-register.d/runs/1.md, brief: "sha256:a87d0a63eb722c4f", reason: "10/10 red-first CHECKS green; every referent (M1-M6, probed A2-A6, E1-E8, R:TOOL_WRITES_REGISTER/UNDECLARED_PROCESSOR/ASSUMED_COMPLIANT/FABRICATED_SIGNOFF) bound to a passing test; ruff clean. Pure build_register over UsageSource port with reconciliation: used-but-undeclared processor is a HARD finding; fail-closed on missing field/unsigned DPA (never documented); DPA expiry computed against injected now; IncompleteFetch fails-not-truncates; read+report only, no write path; review record unreviewed-draft; infra (non-customer-data) listed-for-reconciliation not risk-classified." } +--- +## CARD +goal: A read+report-only CC9.2 subprocessor register that classifies every declared third-party processor of customer data (documented / incomplete / dpa_expired / unused) AND reconciles the declared list against what the system actually reaches — a used-but-undeclared processor is a HARD finding, never a silent omission. +why: R8 soc2-groundwork task (CC9.2 vendor & subprocessor risk). Third of the standalone technical controls; no recruit dependency. A register that is only a hand-maintained doc is theater — reconciliation against real usage is the first real evidence cycle. +beat: done · next: add status + +## RULES + +- M1 The tool is READ + REPORT ONLY. It never edits the declared register (`subprocessors.json`), and never mutates any vendor system — it only reads and classifies. -> "TOOL_WRITES_REGISTER" +- M2 The core is PURE: `build_register(source, declared, *, now_iso)` over a `UsageSource` Port. No clock, no network — `now_iso` is INJECTED. A `UsageSource` that cannot enumerate the WHOLE live processor set raises `IncompleteFetch`; the core NEVER emits a reconciliation from a partial enumeration (else an undeclared processor slips through as "all declared"). +- M3 Every declared subprocessor that processes customer data is classified. A missing required field (purpose / region / data_categories) or an unsigned/absent DPA is NEVER `documented` — it fails CLOSED to `incomplete`. -> "ASSUMED_COMPLIANT" +- M4 The declared register is reconciled against the live `UsageSource`: a processor the system actually reaches but which is ABSENT from the declared register is a HARD finding recorded in `undeclared`, never dropped. -> "UNDECLARED_PROCESSOR" +- M5 A DPA whose expiry is on/before the INJECTED `now_iso` is classified `dpa_expired` — never left "valid" by omission; the boundary is computed, never assumed. +- M6 The review record reviewer is `unreviewed — draft` until a real human signs. The tool never fabricates a reviewer. -> "FABRICATED_SIGNOFF" + + +- R:TOOL_WRITES_REGISTER a code path that writes/edits subprocessors.json or issues any vendor-system mutation -> "TOOL_WRITES_REGISTER" +- R:UNDECLARED_PROCESSOR a live-reached processor absent from the declared register being omitted / silently treated as fine -> "UNDECLARED_PROCESSOR" +- R:ASSUMED_COMPLIANT a vendor missing a required field or a signed in-window DPA being classified `documented` -> "ASSUMED_COMPLIANT" +- R:FABRICATED_SIGNOFF the review record naming any reviewer other than `unreviewed — draft` -> "FABRICATED_SIGNOFF" + + +## ASSUMPTIONS +- A1 [who] covers: S1 · the request does not say who MAINTAINS the register vs who REVIEWS it; taking: `subprocessors.json` is human-maintained INPUT and the tool is a reviewer-assistant that classifies but never signs (M6) -> if wrong, the tool would appear to author/approve vendor risk, which is exactly the fabrication M6 forbids +- A2 [which] covers: S1 · the request does not say which vendors are in scope; taking: only entries with `processes_customer_data: true` are risk-classified — pure infra that never touches customer data is listed-but-not-classified · probe: a processes_customer_data=false vendor appears in output with NO risk classification -> if wrong, the register drowns real subprocessor risk in undifferentiated infra +- A3 [when] covers: S1 · the request does not say where the DPA-expiry boundary falls; taking: a DPA is expired when `now_iso >= dpa_expiry` (expiry instant is NOT still-valid — fail closed at the boundary) · probe: a DPA with expiry == now_iso classifies `dpa_expired` -> if wrong, a lapsed DPA reads as covered on its expiry day +- A4 [absent] covers: S1 · the request does not say what a missing value means; taking: any missing required field or absent/unsigned DPA means NON-compliant → `incomplete`, never assumed compliant · probe: a vendor with dpa_status other than "signed" is `incomplete`, never `documented` -> if wrong, a blank field silently passes as compliant (R:ASSUMED_COMPLIANT) +- A5 [order] covers: S1 · the request does not say what orders the register; taking: worst-severity first (dpa_expired, incomplete, documented, unused) then vendor name — a total, deterministic order · probe: two builds over identical source+declared+now are byte-identical and severity-ordered -> if wrong, the auditor diff is noisy and non-reproducible +- A6 [experience] covers: S1 · the request does not say who receives this; taking: the reader is an auditor — the summary is payload-free (counts + named findings, no endpoint secrets/tokens) · probe: the rendered summary carries the counts + `unreviewed — draft` and NO token/secret substring -> if wrong, evidence leaks credentials or is unreadable +every `gives:` surface is swept on every dimension. A1 is an unprobed reading (who-maintains); A2–A6 are probe-backed and cited from CHECKS. + +## PLAN +contract: > + Dataclasses (frozen): `Subprocessor(name, purpose, data_categories: tuple[str,...], region, + processes_customer_data: bool, dpa_status: str, dpa_expiry: str|None)`; + `UsageRef(name, surface, observed_in)` — a processor the system actually reaches; + `ReviewedVendor(vendor: Subprocessor, classification, reason, used: bool)`; + `ReviewRecord(reviewer, generated_window)`; + `VendorRegister(vendors: tuple[ReviewedVendor,...], undeclared: tuple[UsageRef,...], review, + n_vendors, n_incomplete, n_expired, n_undeclared)`. + `Classification = Literal["documented","incomplete","dpa_expired","unused"]`. + Port `UsageSource.processors(*, org, repo) -> Sequence[UsageRef]` (may raise IncompleteFetch). + `build_register(source, declared: Sequence[Subprocessor], *, org, repo, now_iso) -> VendorRegister` + — PURE. `render_summary(register) -> str` and `as_dict(register) -> dict` (payload-free, + deterministic). IO adapter `ConfigUsageSource` (design-for-failure; live enumeration of provider + hosts / egress allowlist / storage config = documented NotImplementedError the operator wires). +scope: scripts/soc2/vendor_register.py · scripts/soc2/subprocessors.json · apps/gateway/tests/soc2_vendor_register/test_vendor_register.py + +## EDGES +- E1 a processor in the live UsageSource but absent from the declared register → HARD finding in `undeclared` (R:UNDECLARED_PROCESSOR) +- E2 a declared vendor NOT present in live usage → `unused` (flagged, not a hard fail) +- E3 a DPA expired by the injected `now_iso` → `dpa_expired` +- E4 `IncompleteFetch` from the UsageSource → build raises, no partial register emitted +- E5 a vendor missing a required field / dpa_status != "signed" → `incomplete`, never `documented` (fail closed) +- E6 a `processes_customer_data: false` vendor → listed but NOT risk-classified +- E7 identical source+declared+now → byte-identical, severity-ordered output +- E8 the review record stays `unreviewed — draft` until a human signs + +## CHECKS +- test_used_but_undeclared_is_hard_finding · covers: M4, E1, R:UNDECLARED_PROCESSOR · a live processor absent from the declared register lands in `undeclared`, counted, never dropped +- test_tool_has_no_write_path · covers: M1, R:TOOL_WRITES_REGISTER · no public symbol writes/edits the register or mutates a vendor; the Port's only method is the read `processors` +- test_missing_dpa_never_documented_fails_closed · covers: M3, A4, E5, R:ASSUMED_COMPLIANT · a vendor with dpa_status != "signed" (or a missing required field) is `incomplete`, never `documented` +- test_expired_dpa_flagged_against_injected_now · covers: M5, A3, E3 · a DPA with expiry on/before now_iso is `dpa_expired` +- test_declared_but_unused_flagged_not_hard · covers: E2 · a declared vendor absent from live usage is `unused` and is NOT counted as undeclared/incomplete/expired +- test_non_customer_data_vendor_listed_not_classified · covers: A2, E6 · a processes_customer_data=false vendor appears but carries no risk classification and is excluded from the risk counts +- test_review_record_unreviewed_until_signed · covers: M6, E8, R:FABRICATED_SIGNOFF · review.reviewer == "unreviewed — draft" +- test_incomplete_fetch_fails_not_truncates · covers: M2, E4 · a UsageSource raising IncompleteFetch propagates; no partial register +- test_build_is_pure_deterministic_and_ordered · covers: M2, A5, E7 · two builds are equal; vendors are worst-severity-first then name +- test_summary_is_payload_free_and_counted · covers: A6 · summary shows the counts + unreviewed-draft and contains no token/secret substring +red-first: every check MUST fail first. + +## EVIDENCE +receipt: .md> +gate: + +## LESSONS +- -> add learn diff --git a/apps/gateway/tests/soc2_vendor_register/test_vendor_register.py b/apps/gateway/tests/soc2_vendor_register/test_vendor_register.py new file mode 100644 index 00000000..397d3fa7 --- /dev/null +++ b/apps/gateway/tests/soc2_vendor_register/test_vendor_register.py @@ -0,0 +1,186 @@ +"""Red-first suite for the CC9.2 vendor / subprocessor register. + +vendor-subprocessor-register TASK §CHECKS (soc2-groundwork). The subject is the repo-governance +tool `scripts/soc2/vendor_register.py`, loaded by file path (like the CC8.1 change-evidence and +CC6 access-review tools). Every check drives the PURE `build_register` core through a zero-network +`FakeUsageSource` with an INJECTED `now_iso` — no config read, no clock, no network. + +RED before Build: `scripts/soc2/vendor_register.py` does not exist — collection fails on the +absent module. +""" + +from __future__ import annotations + +import importlib.util +import sys +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[4] +_MODULE_PATH = REPO_ROOT / "scripts" / "soc2" / "vendor_register.py" + +_spec = importlib.util.spec_from_file_location("soc2_vendor_register", _MODULE_PATH) +assert _spec is not None and _spec.loader is not None, f"cannot load {_MODULE_PATH}" +vr = importlib.util.module_from_spec(_spec) +# Register BEFORE exec so dataclasses (under `from __future__ import annotations`) resolve the +# module namespace via sys.modules — a path-loaded module is otherwise absent (task-1/2 lesson). +sys.modules[_spec.name] = vr +_spec.loader.exec_module(vr) + + +class FakeUsageSource: + def __init__(self, refs, *, raise_incomplete=False): + self._refs = refs + self._raise = raise_incomplete + + def processors(self, *, org, repo): + if self._raise: + raise vr.IncompleteFetch("rate limited mid-enumeration") + return list(self._refs) + + +_NOW = "2026-08-14T00:00:00Z" +_FUTURE = "2027-01-01T00:00:00Z" +_PAST = "2026-01-01T00:00:00Z" + + +def _vendor( + name, + *, + purpose="model inference", + region="us", + data_categories=("prompts",), + processes_customer_data=True, + dpa_status="signed", + dpa_expiry=_FUTURE, +): + return vr.Subprocessor( + name=name, + purpose=purpose, + data_categories=tuple(data_categories), + region=region, + processes_customer_data=processes_customer_data, + dpa_status=dpa_status, + dpa_expiry=dpa_expiry, + ) + + +def _ref(name, *, surface="provider-endpoint", observed_in="config"): + return vr.UsageRef(name=name, surface=surface, observed_in=observed_in) + + +def _build(declared, usage, **kw): + return vr.build_register(FakeUsageSource(usage), declared, now_iso=_NOW, **kw) + + +# ── CHECKS ───────────────────────────────────────────────────────────────────────────── +def test_used_but_undeclared_is_hard_finding() -> None: + """covers: M4, E1, R:UNDECLARED_PROCESSOR""" + reg = _build([_vendor("openai")], [_ref("openai"), _ref("shadow-analytics")]) + undeclared_names = {u.name for u in reg.undeclared} + assert undeclared_names == {"shadow-analytics"}, ( + "a live-reached, undeclared processor is a hard finding" + ) + assert reg.n_undeclared == 1 + + +def test_tool_has_no_write_path() -> None: + """covers: M1, R:TOOL_WRITES_REGISTER""" + forbidden = ("write", "edit", "mutate", "delete", "revoke", "remove", "upsert") + for name in (n for n in dir(vr) if not n.startswith("_")): + assert not any(bad in name.lower() for bad in forbidden), f"{name} looks like a write path" + proto_methods = {m for m in dir(vr.UsageSource) if not m.startswith("_")} + assert proto_methods == {"processors"}, f"UsageSource must be read-only, got {proto_methods}" + + +def test_missing_dpa_never_documented_fails_closed() -> None: + """covers: M3, A4, E5, R:ASSUMED_COMPLIANT""" + # dpa not signed -> incomplete + r1 = _build([_vendor("v_pending", dpa_status="pending")], [_ref("v_pending")]) + assert r1.vendors[0].classification == "incomplete" + # signed but a required field missing (region) -> still incomplete, never documented + r2 = _build([_vendor("v_blank", region="")], [_ref("v_blank")]) + assert r2.vendors[0].classification == "incomplete" + assert r1.n_incomplete == 1 and r2.n_incomplete == 1 + + +def test_expired_dpa_flagged_against_injected_now() -> None: + """covers: M5, A3, E3""" + # expiry exactly at now_iso is expired (fail-closed at the boundary) + r_eq = _build([_vendor("v_edge", dpa_expiry=_NOW)], [_ref("v_edge")]) + assert r_eq.vendors[0].classification == "dpa_expired" + r_past = _build([_vendor("v_old", dpa_expiry=_PAST)], [_ref("v_old")]) + assert r_past.vendors[0].classification == "dpa_expired" + assert r_eq.n_expired == 1 and r_past.n_expired == 1 + + +def test_declared_but_unused_flagged_not_hard() -> None: + """covers: E2""" + # fully compliant but not reached by the live system -> unused, not a hard finding + reg = _build([_vendor("ghost")], usage=[]) + assert reg.vendors[0].classification == "unused" + assert reg.vendors[0].used is False + assert reg.n_undeclared == 0 and reg.n_incomplete == 0 and reg.n_expired == 0 + + +def test_non_customer_data_vendor_listed_not_classified() -> None: + """covers: A2, E6""" + # infra that never touches customer data, missing its DPA, but reached live: + infra = _vendor( + "grafana-cloud", processes_customer_data=False, dpa_status="none", dpa_expiry=None + ) + reg = _build([infra], [_ref("grafana-cloud")]) + classified_names = {v.vendor.name for v in reg.vendors} + assert "grafana-cloud" not in classified_names, "infra is not risk-classified" + # its declaration still suppresses a false undeclared finding (it appears in reconciliation) + assert reg.n_undeclared == 0 + # and a missing DPA on non-customer-data infra never inflates the risk counts + assert reg.n_incomplete == 0 and reg.n_expired == 0 and reg.n_vendors == 0 + + +def test_review_record_unreviewed_until_signed() -> None: + """covers: M6, E8, R:FABRICATED_SIGNOFF""" + reg = _build([_vendor("openai")], [_ref("openai")]) + assert reg.review.reviewer == "unreviewed — draft", "the tool never signs for a human" + + +def test_incomplete_fetch_fails_not_truncates() -> None: + """covers: M2, E4""" + src = FakeUsageSource([_ref("openai")], raise_incomplete=True) + with pytest.raises(vr.IncompleteFetch): + vr.build_register(src, [_vendor("openai")], now_iso=_NOW) + + +def test_build_is_pure_deterministic_and_ordered() -> None: + """covers: M2, A5, E7""" + declared = [ + _vendor("d_ok"), # documented (signed, future, used) + _vendor("d_exp", dpa_expiry=_PAST), # dpa_expired + _vendor("d_inc", dpa_status="pending"), # incomplete + _vendor("d_unused"), # unused (compliant but not reached) + ] + usage = [_ref("d_ok"), _ref("d_exp"), _ref("d_inc")] + a = _build(list(declared), list(usage)) + b = _build(list(declared), list(usage)) + assert a == b, "identical declared + usage + now must be byte-identical" + # A5 worst-severity first: dpa_expired, incomplete, documented, unused + assert [v.classification for v in a.vendors] == [ + "dpa_expired", + "incomplete", + "documented", + "unused", + ] + + +def test_summary_is_payload_free_and_counted() -> None: + """covers: A6""" + reg = _build( + [_vendor("openai"), _vendor("v_bad", dpa_status="pending")], + [_ref("openai"), _ref("v_bad"), _ref("shadow-x")], + ) + summary = vr.render_summary(reg) + assert "undeclared" in summary.lower() + assert "unreviewed — draft" in summary, "the review record shows in the summary" + for secret in ("ghp_", "github_pat_", "token", "authorization", "secret"): + assert secret not in summary.lower(), "the summary must be payload/secret free" diff --git a/scripts/soc2/subprocessors.json b/scripts/soc2/subprocessors.json new file mode 100644 index 00000000..568914ce --- /dev/null +++ b/scripts/soc2/subprocessors.json @@ -0,0 +1,41 @@ +{ + "_comment": "CC9.2 DECLARED subprocessor register — the human-maintained list vendor_register.py reconciles against live usage. A processor the system reaches but that is absent here is a HARD finding; a customer-data vendor missing a required field or a signed, in-window DPA fails CLOSED to `incomplete`. Edited by a human on review; the tool never writes this file. dpa_expiry is ISO-8601 UTC. Fill in real DPA dates/regions as vendor agreements are signed; the values below are placeholders to be verified.", + "subprocessors": [ + { + "name": "openai", + "purpose": "LLM inference (chat / responses / embeddings) for tenants routing to OpenAI models", + "data_categories": ["prompts", "completions"], + "region": "us", + "processes_customer_data": true, + "dpa_status": "pending", + "dpa_expiry": null + }, + { + "name": "anthropic", + "purpose": "LLM inference for tenants routing to Anthropic models", + "data_categories": ["prompts", "completions"], + "region": "us", + "processes_customer_data": true, + "dpa_status": "pending", + "dpa_expiry": null + }, + { + "name": "aws-bedrock", + "purpose": "LLM inference for tenants routing to Bedrock-hosted models", + "data_categories": ["prompts", "completions"], + "region": "us", + "processes_customer_data": true, + "dpa_status": "pending", + "dpa_expiry": null + }, + { + "name": "grafana-cloud", + "purpose": "metrics / dashboards — infrastructure telemetry only, no customer prompt content", + "data_categories": ["operational-metrics"], + "region": "eu", + "processes_customer_data": false, + "dpa_status": "none", + "dpa_expiry": null + } + ] +} diff --git a/scripts/soc2/vendor_register.py b/scripts/soc2/vendor_register.py new file mode 100644 index 00000000..136c2f1a --- /dev/null +++ b/scripts/soc2/vendor_register.py @@ -0,0 +1,320 @@ +"""CC9.2 vendor / subprocessor register (vendor-subprocessor-register TASK, FROZEN v1). + +Produces an auditable subprocessor register: every DECLARED third-party processor of customer +data, classified documented / incomplete / dpa_expired / unused — AND reconciled against what +the system ACTUALLY reaches. A processor the system uses but which is absent from the declared +register is a HARD finding (M4, R:UNDECLARED_PROCESSOR); this is what keeps the register from +being a hand-maintained doc that quietly drifts from reality. + +Two layers, cleanly split: + * PURE core — `build_register(source, declared, *, now_iso)` over a `UsageSource` port. No IO, + no clock-now, no network; `now_iso` is INJECTED, so identical source + declared + now yields + byte-identical output (M2). This is the layer the suite drives through a zero-network fake. + * IO adapter — `ConfigUsageSource`, the only thing that enumerates live processors. Design-for- + failure: a truncated / partial enumeration raises `IncompleteFetch` so the core can NEVER + reconcile against a partial usage set (which would hide an undeclared processor as "all + declared", M2/E4). + +Fail-closed is the spine: a declared vendor missing a required field or a signed, in-window DPA +is NEVER `documented` — it fails closed to `incomplete` (M3, R:ASSUMED_COMPLIANT); a DPA whose +expiry is on/before the injected `now_iso` is `dpa_expired`, computed not assumed (M5); the review +record reads `unreviewed — draft` until a real human signs (M6, R:FABRICATED_SIGNOFF). A vendor +that does not process customer data carries no DPA burden — it is listed for reconciliation but +NOT risk-classified (A2). The tool is READ + REPORT ONLY: it never edits the declared register +or mutates a vendor system (M1, R:TOOL_WRITES_REGISTER). +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +from collections.abc import Sequence +from dataclasses import dataclass +from datetime import datetime +from typing import Literal, Protocol + +DEFAULT_ORG = "pilotspace" +DEFAULT_REPO = "pilotspace/hydroa" +_UNREVIEWED = "unreviewed — draft" + +Classification = Literal["documented", "incomplete", "dpa_expired", "unused"] + +#: Listing order (A5): worst-severity first, then vendor name for a total, deterministic order. +_SEVERITY = {"dpa_expired": 0, "incomplete": 1, "documented": 2, "unused": 3} + + +class IncompleteFetch(Exception): + """A UsageSource could not enumerate the WHOLE live processor set (rate limit / partial page). + + Raising this — rather than returning a short list — is what stops the core reconciling against + a partial usage set and hiding an undeclared processor as "all declared" (M2, E4). + """ + + +# ── data ───────────────────────────────────────────────────────────────────────────────── +@dataclass(frozen=True) +class Subprocessor: + """One DECLARED third-party processor (scripts/soc2/subprocessors.json).""" + + name: str + purpose: str + data_categories: tuple[str, ...] + region: str + processes_customer_data: bool + dpa_status: str # "signed" | "pending" | "none" | ... (anything but "signed" is incomplete) + dpa_expiry: str | None # ISO-8601 UTC of DPA expiry, or None if none on file + + +@dataclass(frozen=True) +class UsageRef: + """A processor the system ACTUALLY reaches, as a UsageSource reports it.""" + + name: str + surface: str # e.g. "provider-endpoint", "object-store", "egress-allow" + observed_in: str # where it was observed (config key / manifest) + + +@dataclass(frozen=True) +class ReviewedVendor: + vendor: Subprocessor + classification: Classification + reason: str + used: bool # True when the declared vendor is present in the live UsageSource + + +@dataclass(frozen=True) +class ReviewRecord: + reviewer: str # `unreviewed — draft` until a real human signs (M6) + generated_window: str # human "as of" description (payload-free) + + +@dataclass(frozen=True) +class VendorRegister: + vendors: tuple[ReviewedVendor, ...] # customer-data processors only, severity-ordered (A5) + undeclared: tuple[UsageRef, ...] # live-reached but NOT declared — hard findings (M4) + review: ReviewRecord + n_vendors: int # number of risk-classified (customer-data) vendors + n_incomplete: int + n_expired: int + n_undeclared: int + + +class UsageSource(Protocol): + """The port. Read-only by construction — its ONLY method enumerates live processors (M1).""" + + def processors(self, *, org: str, repo: str) -> Sequence[UsageRef]: ... + + +# ── pure core ──────────────────────────────────────────────────────────────────────────── +def _parse(ts: str) -> datetime: + """ISO-8601 → aware datetime. Trailing `Z` is normalised to `+00:00` (parse, not now()).""" + return datetime.fromisoformat(ts.replace("Z", "+00:00")) + + +def _required_present(v: Subprocessor) -> bool: + """All fields a signed, auditable DPA needs. A blank/absent one fails CLOSED (M3, A4).""" + return bool( + v.purpose.strip() + and v.region.strip() + and v.data_categories + and v.dpa_status == "signed" + and v.dpa_expiry is not None + ) + + +def _classify(v: Subprocessor, *, used: bool, now: datetime) -> ReviewedVendor: + if not _required_present(v): + # M3/A4/E5: missing required field or unsigned DPA — never documented, fails closed. + classification: Classification = "incomplete" + reason = ( + f"missing a required field or signed DPA (dpa_status={v.dpa_status!r}, " + f"expiry={v.dpa_expiry!r}) — fails closed under CC9.2" + ) + elif now >= _parse(v.dpa_expiry): # type: ignore[arg-type] # _required_present proved not-None + # M5/A3/E3: expiry on/before injected now — lapsed, never "valid" by omission. + classification = "dpa_expired" + reason = f"DPA expired (expiry {v.dpa_expiry} on/before as-of instant)" + elif not used: + # E2: compliant but not reached by the live system — a removal candidate, not a hard fail. + classification = "unused" + reason = "declared with a signed, in-window DPA but not observed in live usage" + else: + classification = "documented" + reason = f"signed DPA in window ({v.dpa_expiry}); processes {', '.join(v.data_categories)}" + + return ReviewedVendor(vendor=v, classification=classification, reason=reason, used=used) + + +def build_register( + source: UsageSource, + declared: Sequence[Subprocessor], + *, + org: str = DEFAULT_ORG, + repo: str = DEFAULT_REPO, + now_iso: str, +) -> VendorRegister: + """Classify every customer-data subprocessor and reconcile the whole declared list against + live usage. PURE over the source's output and the injected `now_iso` (M2). + + An `IncompleteFetch` from the source PROPAGATES — the core reconciles only against a usage set + a source enumerated in full (M2, E4). Well-formed source output never raises. + """ + now = _parse(now_iso) + usage = source.processors(org=org, repo=repo) + usage_names = {u.name for u in usage} + # ALL declared names (incl. non-customer-data infra) suppress false undeclared findings (A2). + declared_names = {v.name for v in declared} + + # M4/E1: a live-reached processor absent from the declared register is a hard finding. + undeclared = tuple(u for u in usage if u.name not in declared_names) + + # A2/E6: only customer-data processors are risk-classified; infra is listed-for-reconciliation. + risk_vendors = [v for v in declared if v.processes_customer_data] + reviewed = [_classify(v, used=v.name in usage_names, now=now) for v in risk_vendors] + # A5: total order — worst severity first, then name. Fully deterministic (E7). + reviewed.sort(key=lambda r: (_SEVERITY[r.classification], r.vendor.name)) + + review = ReviewRecord(reviewer=_UNREVIEWED, generated_window=f"as of {now_iso}") + return VendorRegister( + vendors=tuple(reviewed), + undeclared=undeclared, + review=review, + n_vendors=len(reviewed), + n_incomplete=sum(1 for r in reviewed if r.classification == "incomplete"), + n_expired=sum(1 for r in reviewed if r.classification == "dpa_expired"), + n_undeclared=len(undeclared), + ) + + +# ── rendering ──────────────────────────────────────────────────────────────────────────── +def render_summary(register: VendorRegister) -> str: + """A stable, payload-free human summary for an auditor (A6). No token/secret ever appears.""" + r = register + lines = [ + "CC9.2 vendor / subprocessor register", + f"reviewer: {r.review.reviewer} · {r.review.generated_window}", + f"{r.n_vendors} customer-data processors · {r.n_incomplete} incomplete · " + f"{r.n_expired} dpa-expired · {r.n_undeclared} undeclared", + "", + ] + for v in r.vendors: + lines.append( + f"- {v.vendor.name} · {v.vendor.region} · {v.classification} " + f"· {'used' if v.used else 'unused'} — {v.reason}" + ) + for u in r.undeclared: + lines.append(f"- ⚠ UNDECLARED: {u.name} (observed in {u.observed_in}) — add to register") + return "\n".join(lines) + + +def as_dict(register: VendorRegister) -> dict[str, object]: + """Machine-readable register. Deterministic; contains no token/secret.""" + return { + "reviewer": register.review.reviewer, + "generated_window": register.review.generated_window, + "n_vendors": register.n_vendors, + "n_incomplete": register.n_incomplete, + "n_expired": register.n_expired, + "n_undeclared": register.n_undeclared, + "vendors": [ + { + "name": v.vendor.name, + "purpose": v.vendor.purpose, + "data_categories": list(v.vendor.data_categories), + "region": v.vendor.region, + "dpa_status": v.vendor.dpa_status, + "dpa_expiry": v.vendor.dpa_expiry, + "classification": v.classification, + "used": v.used, + "reason": v.reason, + } + for v in register.vendors + ], + "undeclared": [ + {"name": u.name, "surface": u.surface, "observed_in": u.observed_in} + for u in register.undeclared + ], + } + + +# ── IO adapter (design-for-failure; the only thing that enumerates live usage) ──────────── +class ConfigUsageSource: + """Enumerates the external processors the system ACTUALLY reaches (M2/M4). + + In production this is derived deterministically from the deploy surface — the gateway model- + provider registry, the egress allow-list, and the object-store / datastore config — NOT the + network. It is read-only by construction: no revoke/edit/mutate path exists (M1). A partial + or unreadable config raises `IncompleteFetch` rather than returning a short list, so the core + never reconciles against an incomplete usage set (E4). + """ + + def __init__(self, *, config_root: str | None = None) -> None: + self._config_root = config_root or os.environ.get("HYDROA_CONFIG_ROOT", "") + + def processors(self, *, org: str, repo: str) -> Sequence[UsageRef]: # pragma: no cover + raise NotImplementedError( + "live usage enumeration is provisioned by the operator's run environment; the pure " + "core and its reconciliation semantics are what this task freezes and tests. Wire this " + "to the deterministic deploy surface — the model-provider registry, the egress " + "allow-list (infra/), and the object-store/datastore config — returning a UsageRef per " + "external processor, and raising IncompleteFetch on any unreadable/partial source. " + "NEVER issue a write/DELETE against a vendor system from here." + ) + + +# ── CLI ────────────────────────────────────────────────────────────────────────────────── +def _load_declared(path: str) -> list[Subprocessor]: + with open(path, encoding="utf-8") as fh: + raw = json.load(fh) + return [ + Subprocessor( + name=e["name"], + purpose=e.get("purpose", ""), + data_categories=tuple(e.get("data_categories", [])), + region=e.get("region", ""), + processes_customer_data=bool(e.get("processes_customer_data", False)), + dpa_status=e.get("dpa_status", "none"), + dpa_expiry=e.get("dpa_expiry"), + ) + for e in raw.get("subprocessors", []) + ] + + +def main(argv: Sequence[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="CC9.2 vendor / subprocessor register") + parser.add_argument("--now", required=True, help="ISO-8601 UTC 'as of' instant (injected)") + parser.add_argument("--org", default=DEFAULT_ORG) + parser.add_argument("--repo", default=DEFAULT_REPO) + parser.add_argument( + "--declared", + default=os.path.join(os.path.dirname(__file__), "subprocessors.json"), + help="the human-maintained declared subprocessor register", + ) + parser.add_argument("--json-out", default="vendor_register.json") + args = parser.parse_args(argv) + + source = ConfigUsageSource() + try: + register = build_register( + source, + _load_declared(args.declared), + org=args.org, + repo=args.repo, + now_iso=args.now, + ) + except IncompleteFetch as exc: + print(f"USAGE ENUMERATION INCOMPLETE — register NOT written: {exc}", file=sys.stderr) + return 2 + + with open(args.json_out, "w", encoding="utf-8") as fh: + json.dump(as_dict(register), fh, indent=2, sort_keys=True) + print(render_summary(register)) + # An undeclared / incomplete / expired processor is the auditor signal — exit non-zero. + signal = register.n_undeclared or register.n_incomplete or register.n_expired + return 1 if signal else 0 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main())