diff --git a/.env.example b/.env.example index 7e5fa4d..552dd85 100644 --- a/.env.example +++ b/.env.example @@ -27,6 +27,11 @@ GITHUB_TOKEN= # repo-scoped PAT for opening PRs (or use `gh auth login`) # VPCOPILOT_SIM_LOGS=traffic.jsonl # traffic sample the refiner also checks each candidate against # VPCOPILOT_MINISIGN_KEY=~/.minisign/vpcopilot.key # sign exported bundles (unencrypted key: minisign -G -W) # VPCOPILOT_MINISIGN_BIN= # path to minisign, if not on PATH +# VPCOPILOT_ESCALATION_WEBHOOK= # POST target for `reconcile` escalations (the audit record is the durable one) +# --- Audit event sink (J3): ship every audit entry off the box as it is written --- +# https://… | syslog://host:514 | syslog:///var/run/syslog | stdout | off (`vpcopilot audit-sink` to check) +# VPCOPILOT_AUDIT_SINK= # the local audit.log stays authoritative; a sink is a copy +# VPCOPILOT_AUDIT_SINK_TOKEN= # optional bearer token for an https sink # --- Validation auth (auth-protected targets: let the probe log in so it can demonstrate the exploit) --- # VPCOPILOT_PROBE_USER= # username the validation probe logs in with (cookie or token app) # VPCOPILOT_PROBE_PASS= # its password diff --git a/ROADMAP.md b/ROADMAP.md index 7e43c9b..eac3630 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -715,13 +715,134 @@ sees the trail. J1–J4 are the open `BACKLOG.md` evidence entries, scheduled; * `manifest.json` already SHA-256s every member and `docs/AUDIT.md` ships a runnable verification snippet. Signature checking is added when J1 lands. -- [ ] **J3** Audit event sink. (M, P2) - An optional sink on `audit.record` sending each entry to syslog, an HTTP webhook, or - stdout JSON, putting the trail somewhere other than the box making the change. - - Acceptance: fail-soft, a dead collector never fails an apply and logs one warning; - configured in `.env` and the ⚙ Setup page; the local `audit.log` stays authoritative. +- [x] **J3** Audit event sink. (M, P2) — **DONE:** `audit_sink.py`, hooked inside `audit.record` + itself, so all 30 call sites and all three surfaces inherit it with no call-site change. + `VPCOPILOT_AUDIT_SINK` picks the transport by scheme — `https://…` (POST), `syslog://host:514` + (RFC 3164 datagram), `syslog:///var/run/syslog` (unix datagram), `stdout`, or `off` — plus + `VPCOPILOT_AUDIT_SINK_TOKEN`. Both keys are on the ⚙ Setup page; `vpcopilot audit-sink [--send]` + and `GET`+`POST /api/audit-sink` are the check. No new dependency (httpx is already a base dep; + syslog is a raw socket). Verified live end to end. 68 tests; suite 785 → 854, coverage 79% → 80%. + - **Acceptance, as met:** a dead collector never fails an apply ✅ — verified live, a real + `audit-backfill` against a black-holed sink exited **0**, wrote its sidecar and its audit entry, + and emitted exactly **one** warning; configured in `.env` and the ⚙ Setup page ✅ — verified by + POSTing `/api/config` in a running console and watching it take effect with **no restart**; the + local `audit.log` stays authoritative ✅ — the line is serialized **once** and handed to both + destinations, so the shipped copy is byte-identical to the line on disk (verified live), and a + failed local write delivers nothing. + - **"Logs one warning" is a design constraint, not a log line.** A hung collector at a 5 s timeout + times the several entries one apply writes is a stall the criterion does not name but plainly + forbids. So a failure starts a 60 s cooldown: one timeout and one warning per outage, not per + entry. The warning goes to **stderr**, never stdout — `audit.record` takes no log callable and + adding one would land it in `**detail` and be serialized into the entry, and a warning that + relied on K1's stdout swap would corrupt the protocol stream anywhere the swap is not in force. + - **Refusing to guess, applied to a transport.** An unparseable sink value is reported `unusable` + with the reason on every surface — it is *not* silently equivalent to having no sink. That is the + H2 `unpinned`-vs-clean distinction, and it is the whole feature: a run succeeds either way, so + without it a misconfigured sink is invisible. + - **`off` is a value, not an empty box.** `console._write_env` drops falsy updates and the UI skips + empty inputs, so clearing the field cannot unset a key — without an explicit `off`, a sink + switched on from the Setup page could never be switched off from it. + - **The syslog size limit is asked of the kernel, never hardcoded — and it is not theoretical.** + Measured: a macOS `/var/run/syslog` unix socket has `SO_SNDBUF` **2048** and refuses more with + `EMSGSIZE`; UDP walls near 9216; a Linux `/dev/log` is far larger. A realistic `drift_detected` + carrying a 60-field diff measured **5,481 bytes** — 2.7× the macOS limit — so the entries that + overflow are exactly the ones worth shipping. The full entry is sent and *only* a kernel refusal + triggers a reduced envelope. Verified against a real kernel: a **16,139-byte** entry was refused + on UDP and a **363-byte** valid-JSON envelope arrived carrying the action, `finding_id`, the true + byte count and a pointer to the local log, while the on-disk entry kept all 120 changes. A + dropped record or a truncated fragment would both have read as nothing happening. + - **Decisions.** *Hooked inside `audit.record`*, the one chokepoint, so the MCP write tools inherit + it with no tool-list change (pinned by a test). *No new `record()` parameter of any kind* — + `**detail` swallows unknown kwargs and `json.dumps` has no `default=`, so a stray `log=` would be + serialized into the entry or raise after the LB was already mutated. *The sink never writes an + audit record of its own*: it lives inside `record`, so that would recurse, and an + entry-count-changing side effect is the bug J4's no-op check exists to prevent — delivery status + is reported by `audit-sink` and the Setup card instead. *The entry goes verbatim, with no + envelope and no `out_dir`* — a filesystem path is not something to put on the wire (the J1 leak + precedent) and `run_id` is already the join key. *One sink, not a list.* *Timeout hardcoded at + 5 s*, matching the house style of literal timeouts, but shorter than `reconcile.notify`'s 10 s + because this one sits on the critical path of an apply rather than at the end of a pass. + - **Deliberately no MCP tool.** The sink itself is inherited by every MCP write tool, which is the + point; the *check* is operator configuration, and sending a test event to an external collector + on a model's initiative is what K1's opt-in exists to prevent. + - **Found by self-review before the reviewers reported, each pinned by a test.** Two were real + leaks or dead ends rather than style: + - **A password in the sink URL would have been printed on every surface.** `urlsplit` puts + userinfo in `netloc`, and the redactor was rebuilding the origin from `netloc` — so + `https://svc:hunter2@collector/x` rendered the password on the CLI panel, the Setup page and + the API response. Rebuilt from `hostname`/`port`, with userinfo collapsed to `…@`. + - **An IPv6 collector could never be reached.** The socket family was hardcoded `AF_INET`, so + `syslog://[fe80::1]:514` was configured and delivered nothing, for ever — the exact + "configured but silently dead" failure the item exists to surface. Resolved via `getaddrinfo`. + - **`check(send=True)` erased the evidence it was called to diagnose.** It called `reset_state()` + to escape the cooldown, wiping a long-lived console's record of earlier failures: an operator + clicking the button to investigate deleted the symptom. Replaced by an explicit `force` that + bypasses only the cooldown; a test event that lands also re-arms a suppressed sink. + - Plus: the one-warning latch was a check-then-set across the console's per-apply threads (now + check-and-set under a lock); the oversize envelope could itself overflow, since it borrows + unbounded strings from the entry (now bounded); and whitespace in a hostname could split the + syslog header so a receiver read part of it as the tag. + - **Found by adversarial review, before shipping** (35 raised across six failure dimensions; 4 + confirmed after independent skeptics tried to refute each, and every one was the same shape the + item exists to prevent — *not delivering, rendering as delivering*): + - **The one input that produced silence was the one this module exists to make loud.** + `urlsplit` *raises* on some malformed values — a dropped bracket in an IPv6 host + (`https://[2001:db8::1/x`), a netloc failing NFKC validation — and unguarded that propagated + out of `status()` into a **CLI traceback and a console 500**, while `emit`'s catch-all + swallowed it with **no warning and no `last_error`**. Reproduced on all three surfaces. Made + reachable by this item's own IPv6 support, which is what invites that typo. Now returns the + ordinary invalid shape every surface already renders. + - **A 3xx counted as a successful delivery.** The client does not follow redirects, so the body + was never re-sent — a moved ingest path or a proxy bouncing to an SSO page would swallow the + entire audit stream while every surface read `delivered N/N`. Reproduced against a real 302 + server: `sent`, `delivered 1`, exit 0, and the redirect target never contacted. **Refused + rather than followed**, deliberately: the request carries the bearer token, and chasing a + `Location` to an unconfigured host is how a credential travels. + - **`sent` over UDP meant "handed to the kernel".** A datagram to a port with nothing bound + succeeds at the send call, so `audit-sink --send` against a dead collector printed `sent` and + exited 0. Now `sent, unconfirmed` on both surfaces, saying what was established and what was + not — the J2 `present-unverified` precedent, where reporting "I cannot check this" as "this + worked" was the one thing that would destroy the distinction that matters. + - **The test suite shipped fabricated audit records to a real collector.** With + `VPCOPILOT_AUDIT_SINK` exported, one full run sent **267 datagrams** of invented + `apply_waf` / `retire` / `rollback_failed` entries — indistinguishable, at the collector, from + records of real changes to a load balancer. The doc had said "unset it before running the + suite", which is a guard that depends on remembering; `tests/conftest.py` now clears it in an + autouse fixture. Verified with a counter: 1 deliberate record registers, the full suite adds + **zero**. + - **Two pre-existing races in `runmeta`, found because J3's concurrency test reproduced them, and + both fixed here.** They are called out rather than slipped in — neither is caused by this change, + but the first is a raise inside `audit.record`, which no audit-integrity item should ship past: + - **`_save`'s temp file was named by PID only**, so two threads writing a fresh out dir shared it + and the loser of the `os.replace` got `FileNotFoundError` **out of `audit.record`** — a change + already made to a load balancer that could not be recorded. Reproduced **12 times out of 12** + with 8 threads and *no sink configured at all*. The console starts every apply on its own + daemon thread with no job lock. Now namespaced by pid **and** thread id. + - **`run_id` minting was an unguarded read-modify-write**, so those same eight concurrent first + writes produced **eight different run_ids** — seven audit entries carrying a join key + `run.json` does not contain, which is the export's whole attribution path. Now thread-atomic; + single-threaded behaviour is byte-identical, and the remaining cross-*process* case is stated + rather than pretended away. + - **Fixed en route (pre-existing console defect, found while validating the Setup card):** + `class="bad"` marks a failure in the H2 dependency preview — a manifest that would not parse, a + package OSV could not be asked about — and **`.bad` was defined nowhere in the CSS**, so all of it + rendered as ordinary body text. A warning styled like content is precisely the defect that + preview exists to prevent. Defined, which repairs the three pre-existing call sites as well as + J3's own; pinned by a test that fails if any failure-marking class is used but undefined. + - **What a sink does not do, stated rather than left to be discovered** (the J1/J4 precedent, and + `docs/AUDIT.md`'s honest-limits list is amended rather than left contradicting the feature): it + does **not** make the log tamper-evident. A delivered copy raises the cost of editing the local + file afterwards; a *missing* one proves nothing, because the transport is allowed to fail. It is + best-effort by construction, and where the two disagree they disagree about delivery. + - **Could not be confirmed on this box, and says so rather than claiming it:** that an entry + reaches a real syslog *daemon's* store. macOS discards `local0.info` by default — `/etc/asl.conf` + routes only auth facilities and `/etc/syslog.conf` forwards only `install.*` — and the control + proves it is the box, not the frame: `/usr/bin/logger -p local0.info` also produces zero hits. + What *was* verified is the wire format, against a strict RFC 3164 parser: `<134>` (local0.info), + space-padded day, hostname, `vpcopilot[pid]`, and a JSON payload that survives framing intact. - **Reconciled:** there is no Admin tab — credential and `.env` editing lives on the ⚙ Setup - page (`GET`/`POST /api/config`). + page (`GET`/`POST /api/config`), which renders `MANAGED_KEYS` generically, so the two new keys + needed no HTML change to appear. - [x] **J4** Attribution backfill. (S, P2) — **DONE:** `vpcopilot audit-backfill` + `POST /api/audit-backfill`. `backfill.py` freezes the finding each audit entry belongs to into diff --git a/docs/AUDIT.md b/docs/AUDIT.md index 74f3c6d..4254167 100644 --- a/docs/AUDIT.md +++ b/docs/AUDIT.md @@ -59,7 +59,11 @@ zip can never imply more coverage than it has: - The log is written by the same process that makes the change, to a local file. It is **not tamper-evident** — anyone who can write the out dir can edit `audit.log`. The manifest's SHA-256s prove a bundle was not altered *after export*; they say nothing about the authenticity of the log - before it. If you need tamper-evidence, ship the out dir to append-only storage. + before it. If you need tamper-evidence, ship the out dir to append-only storage — or set an + **audit event sink** (§10) so a copy of each entry leaves the box as it is written, which is the + cheap half of the same answer. The sink is **best-effort**: a delivered copy raises the cost of + editing the local log afterwards, but a *missing* one proves nothing, because the transport is + allowed to fail. - `actor` is self-asserted (`VPCOPILOT_ACTOR`, else the OS user) — it is attribution, not authentication. - Nothing in `export.py` calls XC or GitHub. The trail says what the tool *did*; the LB itself is @@ -648,11 +652,130 @@ shipped verbatim in the bundle. If the sidecar and the log ever disagree, the lo --- +## 10. Audit event sink — a copy off the box (J3) + +The log lives on the machine that made the change, which is the one machine an attacker who made an +unauthorised change would want to edit. `VPCOPILOT_AUDIT_SINK` ships each entry to a collector **as +it is written**, so the local file stops being the only copy. + +```sh +VPCOPILOT_AUDIT_SINK=https://collector.example.com/ingest # POST the entry as the body +VPCOPILOT_AUDIT_SINK=syslog://10.0.0.9:514 # RFC 3164 datagram over UDP +VPCOPILOT_AUDIT_SINK=syslog:///var/run/syslog # …or a local unix datagram socket +VPCOPILOT_AUDIT_SINK=stdout # one JSON line, for a log-scraping runtime +VPCOPILOT_AUDIT_SINK=off # deliberately disabled +VPCOPILOT_AUDIT_SINK_TOKEN=… # optional: sent as `Authorization: Bearer …` +``` + +Both keys are on the ⚙ **Setup** page. `off` is a *value* rather than an empty box on purpose: the +console's `.env` writer drops blank updates, so clearing the field cannot unset a key — without an +explicit `off`, a sink switched on from the Setup page could never be switched off from it. + +### What it is, and what it is not + +**The local `audit.log` stays authoritative.** The line is appended to disk first and the sink is +handed *the same string*, so the two cannot disagree about what happened. If the local write fails, +nothing is delivered — an off-box event with no local counterpart could never be reconciled against +the run it belongs to. + +**It never fails the run it observes.** Every delivery path returns a status and nothing raises. +`audit.record` is called on the `rollback_failed` path, so a sink that could raise would turn *"the +LB may be left in a changed state"* into an unrecorded event. A dead collector costs **one warning +on stderr and one timeout**, not one per entry: after a failure the sink goes quiet for 60 s rather +than paying the 5 s timeout again on the next record, because an apply writes several entries and +their sum would be a real stall. + +**Misconfigured never renders as clean.** A value that cannot be parsed is reported as *unusable* +with the reason, and is not silently equivalent to having no sink. This is the same distinction +`dependencies.json` draws between `unpinned` and clean, applied to a transport. That includes values +`urlsplit` itself refuses — a dropped bracket in an IPv6 host (`https://[2001:db8::1/x`) is the typo +that syntax invites, and it is reported rather than raised. + +**A redirect is a failure, not a delivery.** The client does not follow redirects, so a `3xx` means +the body was never re-sent: a moved ingest path, or a proxy bouncing to an SSO page, would otherwise +swallow the whole stream while every surface read `delivered N/N`. It is refused rather than +followed on purpose — the request carries the bearer token, and chasing a `Location` to a host you +did not configure is how a credential travels. Point the sink at the final URL. + +**`sent` over UDP means "handed to the kernel".** A datagram to a port with nothing bound *succeeds* +at the send call, so the syslog-UDP transport reports **`sent, unconfirmed`** and says why. The HTTP +and unix-socket transports get an answer from the other end and report `sent` unqualified. This is +the same three-state honesty as `export --verify`'s `present-unverified`: reporting "I cannot check +this" the same way as "this worked" would destroy the distinction that matters most. + +**A sink is a copy, not a second source of truth.** It attests nothing on its own: the collector +receives what this tool sent, exactly as the local log records what this tool did. Where the two +disagree, they disagree about *delivery*, and the bundle ships the log. + +**What the collector receives.** The entry verbatim — the same JSON, including `actor`, `host`, +`run_id` and every per-action detail field. There is no envelope and no `out_dir`: a filesystem +path is not something to put on the wire (the J1 precedent, where the signature's trusted comment +leaked the exporter's absolute path), and `run_id` is already the join key. Read §2 for what those +details contain before pointing this at a collector outside your network. + +### Checking it + +A sink that is configured and silently not delivering is the failure this feature has to make +visible, because the run succeeds either way: + +```sh +vpcopilot audit-sink # configuration only — no network +vpcopilot audit-sink --send # deliver one test event; exits non-zero if it does not land +``` + +The Setup page has the same readout and a **Send test event** button (`GET`/`POST /api/audit-sink`). +The test event is deliberately **not** shaped like an audit entry — it carries `kind: +"vpcopilot-audit-sink-test"` where a real entry carries `action`, so a collector alerting on actions +cannot be tripped by a connectivity check — and it is written to no `audit.log`: asking whether the +sink works must not add to the evidence it carries. + +Delivery counters (`attempted` / `delivered` / `failed` / `suppressed`) are **per process**. A +one-shot CLI apply reports its own run; a long-lived console or MCP session accumulates. A CLI run +that failed said so on stderr, and there is nowhere else it could have been recorded — writing a +delivery record into `audit.log` would recurse, and an entry-count-changing side effect is the bug +J4's no-op check exists to prevent. + +### Syslog size, measured + +A syslog datagram has a hard size limit and the kernel enforces it by **refusing the write** +(`EMSGSIZE`) — measured at 2048 bytes on a macOS `/var/run/syslog` unix socket and around 9216 for +UDP; a Linux `/dev/log` is far larger. The entries that overflow are the interesting ones, because +`drift_detected` carries a whole field-level diff. So nothing is hardcoded: the full entry is sent, +and *only* if the kernel refuses it is a reduced envelope sent instead — + +```json +{"ts":"…","action":"drift_detected","run_id":"…","actor":"…","finding_id":"…", + "audit_sink_oversize":true,"bytes":4213, + "note":"entry too large for one syslog datagram — the full record is in the run's audit.log"} +``` + +— valid JSON that says an entry happened, identifies it, and says it did not fit, rather than a +dropped record or a truncated fragment. Note also that a *remote* receiver may impose its own +smaller limit (RFC 3164 only requires 1024 bytes to be accepted); the HTTP sink has no such ceiling. + +The syslog header timestamp is local time, because the RFC says so — it is the transport's clock. +The authoritative one is `ts` **inside** the JSON, which is UTC. + +### One more thing worth knowing + +The sink is process-global and reads its configuration from the environment on every entry, so a +Setup-page save takes effect on the next record with no restart. + +The test suite is immune to that by construction rather than by convention: `tests/conftest.py` +clears both variables in an autouse fixture. It has to, because most of the suite exercises +`audit.record` — with the variable exported, one full run shipped **267 fabricated audit records** +(`apply_waf`, `retire`, `rollback_failed`) to the collector, where nothing distinguishes them from +records of real changes to a load balancer. + +--- + ## See also - `docs/USAGE.md` — the full apply / PR / retire workflow - `DESIGN.md` — where the audit sink sits in the architecture - `src/vpcopilot/export.py` — `COLUMNS`, `CATEGORY`, `CONTROL`, `BUNDLE_FILES` +- `src/vpcopilot/audit_sink.py` — the off-box sink (§10) - Tests: `tests/test_audit_provenance.py` (what every entry must carry), `tests/test_export.py` (normalization, CSV, bundle, multi-run), - `tests/test_console_audit_export.py` (the endpoints) + `tests/test_console_audit_export.py` (the endpoints), + `tests/test_audit_sink.py` (the sink: fail-soft, redaction, transports, size) diff --git a/docs/USAGE.md b/docs/USAGE.md index 65d7b29..3adc5bb 100644 --- a/docs/USAGE.md +++ b/docs/USAGE.md @@ -365,6 +365,36 @@ nothing here touches XC or GitHub. Dry runs are not in it: nothing changed, so nothing is logged. The bundle is evidence for a human reviewer, not a compliance certification. Full reference: **[AUDIT.md](AUDIT.md)**. +### Shipping the trail off the box (J3) + +The log is written by the machine that made the change, which is the one machine someone who made +an unauthorised change would want to edit. Point `VPCOPILOT_AUDIT_SINK` at a collector and each +entry is copied there as it is written: + +```sh +VPCOPILOT_AUDIT_SINK=https://collector.example.com/ingest # POST the entry as the body +VPCOPILOT_AUDIT_SINK=syslog://10.0.0.9:514 # …or syslog:///var/run/syslog +VPCOPILOT_AUDIT_SINK=stdout # …or a JSON line for a log-scraping runtime +VPCOPILOT_AUDIT_SINK=off # deliberately disabled +vpcopilot audit-sink --send # prove it lands (exits non-zero if it does not) +``` + +Both keys are on the console's ⚙ **Setup** page, which has the same readout and a **Send test +event** button. `off` is a value rather than a blank field because the `.env` writer drops empty +updates — a sink switched on from that page has to be switchable off from it. + +**The local `audit.log` stays authoritative.** The line is written to disk first and the sink gets +the same string, so the two cannot disagree; if the local write fails, nothing is delivered. +Delivery is fail-soft and can never change the outcome of the action being recorded — a dead +collector costs one warning on stderr and one timeout, then goes quiet for a minute rather than +stalling every subsequent entry. A sink that is *misconfigured* reports as unusable with the +reason, because "we are shipping nothing" must never read the same as "nothing is configured". + +What a sink does **not** do is make the log tamper-evident. A delivered copy raises the cost of +editing the local file afterwards; a missing one proves nothing, because the transport is allowed +to fail. See **[AUDIT.md §10](AUDIT.md)** for what the collector receives, the measured syslog size +limit, and what the sink attests. + **Signing a bundle (optional).** Point `VPCOPILOT_MINISIGN_KEY` at an *unencrypted* minisign secret key and every export gains `manifest.json.minisig` beside the manifest: diff --git a/src/vpcopilot/audit.py b/src/vpcopilot/audit.py index 5b44766..3959549 100644 --- a/src/vpcopilot/audit.py +++ b/src/vpcopilot/audit.py @@ -6,13 +6,17 @@ Identity (`run_id` / `actor` / `host` / `tool_version`) is stamped **here** rather than at each call site, so no mutating path can forget it and no caller can override it — an entry that cannot say who -made the change, from where, and as part of which run is not an audit record.""" +made the change, from where, and as part of which run is not an audit record. + +J3: when `VPCOPILOT_AUDIT_SINK` is set, the same line is also shipped off-box (see `audit_sink`). +The local file is written first and is authoritative; delivery is fail-soft and cannot change the +outcome of the action being recorded.""" from __future__ import annotations import json from pathlib import Path -from . import __version__, runmeta +from . import __version__, audit_sink, runmeta _STAMPED = ("ts", "run_id", "actor", "host", "tool_version") @@ -21,10 +25,17 @@ def record(out_dir, action: str, **detail) -> None: detail = {k: v for k, v in detail.items() if k not in _STAMPED} entry = {"ts": runmeta.utc_now(), "action": action, "run_id": runmeta.run_id(out_dir), "actor": runmeta.actor(), "host": runmeta.host(), "tool_version": __version__, **detail} + # Serialized ONCE and handed to both destinations, so the shipped copy and the line on disk + # cannot drift. A non-serializable detail value still raises here, as it always has — before + # anything is written or sent (`ledger.py` pre-serializes datetimes for exactly this reason). + line = json.dumps(entry) p = Path(out_dir) p.mkdir(parents=True, exist_ok=True) with open(p / "audit.log", "a") as f: - f.write(json.dumps(entry) + "\n") + f.write(line + "\n") + # After the local write, never instead of it: an off-box event with no local counterpart could + # never be reconciled against the run it belongs to. + audit_sink.emit(line, entry) def load(out_dir) -> list[dict]: diff --git a/src/vpcopilot/audit_sink.py b/src/vpcopilot/audit_sink.py new file mode 100644 index 0000000..38a2f84 --- /dev/null +++ b/src/vpcopilot/audit_sink.py @@ -0,0 +1,401 @@ +"""J3 — optional off-box sink for the audit trail. + +`audit.log` is written by the same process that makes the change, to a local file, so anyone who +can write the out dir can edit it (`docs/AUDIT.md`, honest limits). Shipping each entry to a +collector as it is written is the cheap half of the answer: the copy leaves the box immediately, +so altering the local log afterwards no longer alters the only record. + +Three transports, chosen by the scheme of one variable, `VPCOPILOT_AUDIT_SINK`: + + https://collector.example.com/ingest POST the entry as the request body + syslog://10.0.0.9:514 RFC 3164 datagram over UDP + syslog:///var/run/syslog …or to a local unix datagram socket + stdout one JSON line on stdout, for a log-scraping runtime + off deliberately disabled (distinct from never configured) + +Four properties are the whole point, and each is pinned by a test: + +- **The local log stays authoritative.** The line is written to disk *first* and the sink is handed + the exact same string, so the two cannot disagree about what happened. If the local write fails, + nothing is delivered — an off-box event with no local counterpart would be unreconcilable. +- **It never fails the run it observes.** Every path returns a status string; nothing raises. A + sink hangs off `audit.record`, which is called on the rollback-failed path, so a sink that could + raise would turn "the LB may be left changed" into an unrecorded event. +- **A dead collector costs one warning and one timeout, not one per entry.** After a failure the + sink goes quiet for `COOLDOWN_S` rather than paying the timeout again on the next record — an + apply writes several entries and a hung collector would otherwise stall it for their sum. +- **Misconfigured never renders as clean.** An unparseable value is reported as invalid and warns; + it is not silently equivalent to having no sink at all. + +Deliberately NOT here: a delivery record written back into `audit.log`. The sink lives inside +`audit.record`, so recording its own outcome would recurse, and an entry-count-changing side effect +is the bug J4's no-op check exists to prevent. Delivery outcome is reported by `vpcopilot +audit-sink` and `GET /api/audit-sink` instead. +""" +from __future__ import annotations + +import errno +import json +import os +import socket +import sys +import threading +import time +from urllib.parse import urlsplit + +from . import __version__, runmeta + +SINK_ENV = "VPCOPILOT_AUDIT_SINK" +TOKEN_ENV = "VPCOPILOT_AUDIT_SINK_TOKEN" + +TIMEOUT_S = 5 # an audit sink sits on the critical path of an apply, unlike reconcile's +COOLDOWN_S = 60.0 # escalation webhook (10s) which runs at the end of a pass +SYSLOG_PORT = 514 +SYSLOG_PRI = 134 # local0.info — facility 16 * 8 + severity 6 + +# Injected rather than called directly, so tests can drive the cooldown without sleeping (the +# house convention for clocks — cf. reconcile's `now`/`sleep` parameters). +_now = time.monotonic +_localtime = time.localtime + +_MONTHS = ("Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec") + +_state: dict = {"attempted": 0, "delivered": 0, "failed": 0, "suppressed": 0, + "last_error": "", "warned": False, "quiet_until": 0.0} +# The console runs each apply on its own daemon thread and they share this module, so the counters +# are read-modify-written concurrently. Only the bookkeeping is guarded — never the socket call, or +# one hung collector would serialize every other thread's audit write behind it. (`ledger._LOCK` is +# the in-repo precedent for a threading lock around shared state.) +_LOCK = threading.Lock() + + +def reset_state() -> None: + """Forget delivery counters and the one-warning latch. For tests only — a surface that wanted + a fresh answer used to call this, which silently destroyed a long-lived console's record of + earlier failures.""" + with _LOCK: + _state.update(attempted=0, delivered=0, failed=0, suppressed=0, + last_error="", warned=False, quiet_until=0.0) + + +def _claim_warning() -> bool: + """Check-and-set under the lock, so "log one warning" stays one warning when two console + threads fail at the same moment.""" + with _LOCK: + if _state["warned"]: + return False + _state["warned"] = True + return True + + +def _warn(msg: str) -> None: + """One line to stderr — never stdout. + + `audit.record` takes no log callable and adding one would land it in `**detail` and be + serialized into the entry. stderr is the only stream that is unconditionally safe: the MCP + server repoints `sys.stdout` at stderr for its lifetime, but a warning that *relies* on that + swap would corrupt the protocol stream anywhere the swap is not in force.""" + try: + sys.stderr.write(msg + "\n") + sys.stderr.flush() + except Exception: # noqa: BLE001 — a closed stderr must not break an audit write + pass + + +def redact(target: str, kind: str | None) -> str: + """What a surface may print. A webhook URL routinely carries the credential in its *path* + (Slack, Teams, a Splunk HEC token), so only the origin is ever shown — the same reason + `reconcile.notify` logs `type(e).__name__` and never `str(e)`. + + Rebuilt from `hostname`/`port`, deliberately NOT from `netloc`: `urlsplit` puts userinfo in + netloc, so `https://user:pass@host/x` would otherwise print the password on the CLI panel, the + Setup page and the API response.""" + if kind in ("http", "https"): + u = urlsplit(target) + host = u.hostname or "" + try: + port = f":{u.port}" if u.port else "" + except ValueError: + port = "" + auth = "…@" if (u.username or u.password) else "" + tail = "/…" if (u.path.strip("/") or u.query) else "" + return f"{u.scheme}://{auth}{host}{port}{tail}" + return target + + +def configure() -> dict: + """Read the sink configuration from the environment, every time. + + Deliberately not cached: `POST /api/config` writes `.env` and reloads it into `os.environ` in + the running console, so a module-level constant would leave the Setup page lying until relaunch. + + Returns `{configured, kind, target, redacted, reason}`. `kind` is None when nothing usable was + configured; `reason` says which of the three cases that is — unset, deliberately `off`, or a + value that could not be parsed.""" + raw = (os.environ.get(SINK_ENV) or "").strip() + if not raw: + return {"configured": False, "kind": None, "target": "", "redacted": "", + "reason": "no sink configured"} + if raw.lower() in ("off", "none", "disabled"): + # A value, not an empty box: the console's .env writer drops falsy updates, so clearing the + # field cannot unset a key. Without an explicit off, a sink turned on from the Setup page + # could never be turned off from it. + return {"configured": False, "kind": None, "target": "", "redacted": "", + "reason": "explicitly disabled"} + if raw.lower() in ("stdout", "stdout:", "stdout://"): + return {"configured": True, "kind": "stdout", "target": "stdout", "redacted": "stdout", + "reason": ""} + + try: + u = urlsplit(raw) + except ValueError as e: + # `urlsplit` RAISES on some malformed values — a dropped bracket in an IPv6 host + # (`https://[2001:db8::1/x`), a netloc that fails NFKC validation. Unguarded, that + # propagated out of `status()` into a CLI traceback and a console 500, while `emit`'s bare + # except swallowed it with no warning at all: the one input shape that produced *silence* + # on the apply path was the shape this module exists to make loud. + return {"configured": True, "kind": None, "target": raw, "redacted": "(unparseable)", + "reason": f"sink value could not be parsed as a URL ({e})"} + scheme = u.scheme.lower() + if scheme in ("http", "https"): + if not u.netloc: + return {"configured": True, "kind": None, "target": raw, "redacted": f"{scheme}://…", + "reason": f"{scheme} sink has no host"} + return {"configured": True, "kind": scheme, "target": raw, + "redacted": redact(raw, scheme), "reason": ""} + if scheme == "syslog": + if u.netloc: + host = u.hostname or "" + if not host: + return {"configured": True, "kind": None, "target": raw, "redacted": raw, + "reason": "syslog sink has no host"} + try: + port = u.port or SYSLOG_PORT + except ValueError: + return {"configured": True, "kind": None, "target": raw, "redacted": raw, + "reason": "syslog sink has an invalid port"} + return {"configured": True, "kind": "syslog-udp", "target": raw, + "redacted": f"syslog://{host}:{port}", "reason": "", + "host": host, "port": port} + if u.path: + return {"configured": True, "kind": "syslog-unix", "target": raw, + "redacted": f"syslog://{u.path}", "reason": "", "path": u.path} + return {"configured": True, "kind": None, "target": raw, "redacted": raw, + "reason": "syslog sink names neither a host nor a socket path"} + + got = scheme or "no scheme" + return {"configured": True, "kind": None, "target": raw, "redacted": raw.split("://")[0][:40], + "reason": f"unsupported sink ({got}) — expected https://, http://, syslog://, stdout or off"} + + +# ---------------------------------------------------------------- transports + + +def _syslog_frame(payload: str, *, pid: int) -> bytes: + """RFC 3164: `MMM DD HH:MM:SS HOSTNAME TAG[pid]: MSG`, day space-padded. + + The header timestamp is local time because the RFC says so; it is the *transport's* clock. The + authoritative one is `ts` inside the JSON, which is UTC — stated here because a reader + comparing the two will otherwise think one of them is wrong.""" + t = _localtime() + stamp = f"{_MONTHS[t.tm_mon - 1]} {t.tm_mday:2d} {t.tm_hour:02d}:{t.tm_min:02d}:{t.tm_sec:02d}" + # Whitespace of any kind would split the header into the wrong fields; a receiver would then + # read part of the hostname as the tag. `json.dumps` already escapes newlines inside `payload`, + # so the message itself is structurally single-line. + host = "".join(c for c in (runmeta.host() or "-").split(".")[0] if not c.isspace())[:255] or "-" + return f"<{SYSLOG_PRI}>{stamp} {host} vpcopilot[{pid}]: {payload}".encode("utf-8", "replace") + + +def _oversize_envelope(entry: dict, size: int) -> str: + """What goes out when the full entry will not fit in one datagram. + + Measured: a macOS `/var/run/syslog` unix socket has `SO_SNDBUF` 2048 and refuses anything + larger with EMSGSIZE; UDP walls at ~9216. The entries that overflow are the *interesting* ones + — a `drift_detected` carries a full field-level diff — so dropping them would lose exactly the + records worth shipping, and silence would be indistinguishable from a quiet day. This is a + valid JSON object that says an entry happened, identifies it, and says it did not fit.""" + def _short(v): + # The envelope exists because the entry did not fit, so it must not be able to overflow in + # turn: every field it borrows from the entry is bounded. + return v if not isinstance(v, str) else v[:120] + + return json.dumps({ + "ts": _short(entry.get("ts")), "action": _short(entry.get("action")), + "run_id": _short(entry.get("run_id")), "actor": _short(entry.get("actor")), + "host": _short(entry.get("host")), "tool_version": _short(entry.get("tool_version")), + "finding_id": _short(entry.get("finding_id")), "lb": _short(entry.get("lb")), + "audit_sink_oversize": True, "bytes": size, + "note": "entry too large for one syslog datagram — the full record is in the run's audit.log", + }) + + +def _send_syslog(cfg: dict, line: str, entry: dict) -> None: + """One datagram, with a second attempt carrying the reduced envelope on EMSGSIZE. + + The limit is never hardcoded: it differs by platform (2048 on a macOS unix socket, far larger + on a Linux `/dev/log`) and by socket buffer, so the kernel is asked rather than guessed — it + answers by refusing the write, which is the one signal that is right everywhere.""" + unix = cfg["kind"] == "syslog-unix" + if unix: + fam, dest = socket.AF_UNIX, cfg["path"] + else: + # Resolved rather than assumed AF_INET: a `syslog://[fe80::1]:514` collector is perfectly + # legal and an AF_INET socket cannot reach it at all — the sink would read as configured + # and deliver nothing, for ever. + fam, _, _, _, dest = socket.getaddrinfo( + cfg["host"], cfg["port"], type=socket.SOCK_DGRAM)[0] + pid = os.getpid() + s = socket.socket(fam, socket.SOCK_DGRAM) + try: + s.settimeout(TIMEOUT_S) + frame = _syslog_frame(line, pid=pid) + try: + s.sendto(frame, dest) + except OSError as e: + if e.errno != errno.EMSGSIZE: + raise + s.sendto(_syslog_frame(_oversize_envelope(entry, len(line.encode())), pid=pid), dest) + finally: + s.close() + + +def _send_http(cfg: dict, line: str) -> None: + # Function-local, matching `reconcile.notify` — the tests monkeypatch `httpx.Client`, which + # only works while the attribute is resolved at call time. + import httpx + headers = {"content-type": "application/json"} + token = (os.environ.get(TOKEN_ENV) or "").strip() + if token: + headers["authorization"] = f"Bearer {token}" + with httpx.Client(timeout=TIMEOUT_S) as c: + # `content=` sends the exact bytes that were appended to audit.log. Re-encoding the parsed + # dict would let key order or float formatting drift between the two copies. + r = c.post(cfg["target"], content=line.encode(), headers=headers) + if 300 <= r.status_code < 400: + # httpx does not follow redirects by default, so the body was never re-sent — treating a + # 3xx as success meant a moved ingest path (or a proxy bouncing to an SSO page) swallowed + # the entire audit stream while every surface read "delivered N/N". + # + # Refused rather than followed, deliberately: this request carries the bearer token, and + # chasing a Location to a host the operator did not configure is how a credential ends up + # somewhere it was never meant to go. Point the sink at the final URL instead. + raise OSError(f"HTTP {r.status_code} redirect — point the sink at the final URL") + if r.status_code >= 400: + raise OSError(f"HTTP {r.status_code}") + + +def _send_stdout(line: str) -> None: + """`sys.stdout` is resolved here, not bound at import, so the MCP server's stdout swap applies + and a scraped-stdout sink cannot corrupt the protocol stream.""" + out = sys.stdout + if out is None: + raise OSError("no stdout") + out.write(line + "\n") + out.flush() + + +# ---------------------------------------------------------------- the hook + + +def emit(line: str, entry: dict, *, force: bool = False) -> str: + """Deliver one already-written audit line. Returns a status; never raises. + + `skipped` no sink · `invalid` configured but unusable · `suppressed` inside the post-failure + cooldown · `sent` · `sent-unconfirmed` (UDP, which cannot answer) · `failed`. + + `force` bypasses the cooldown and is used only by `check(send=True)`: an operator asking + "is it working now" must not be answered with a suppression from a minute ago.""" + try: + cfg = configure() + except Exception: # noqa: BLE001 — configuration parsing must not break an audit write either + return "invalid" + if not cfg["configured"]: + return "skipped" + if cfg["kind"] is None: + with _LOCK: + _state["last_error"] = cfg["reason"] + if _claim_warning(): + _warn(f"⚠ audit sink not usable: {cfg['reason']} — nothing is being shipped off-box; " + f"the local audit.log is unaffected (check it with `vpcopilot audit-sink`)") + return "invalid" + + now = _now() + with _LOCK: + if not force and now < _state["quiet_until"]: + _state["suppressed"] += 1 + return "suppressed" + _state["attempted"] += 1 + + try: + if cfg["kind"] == "stdout": + _send_stdout(line) + elif cfg["kind"] in ("http", "https"): + _send_http(cfg, line) + else: + _send_syslog(cfg, line, entry) + except Exception as e: # noqa: BLE001 — every transport failure is the same answer + with _LOCK: + _state["failed"] += 1 + # Never `str(e)`: an httpx error message can carry the URL, and a URL can carry a token. + _state["last_error"] = type(e).__name__ + _state["quiet_until"] = now + COOLDOWN_S + if _claim_warning(): + _warn(f"⚠ audit sink delivery failed ({type(e).__name__}) — the entry is still in the " + f"run's audit.log, which is authoritative; further failures are counted silently " + f"(see `vpcopilot audit-sink`)") + return "failed" + with _LOCK: + _state["delivered"] += 1 + if force: + # A test event that landed is proof the collector is back, so stop suppressing. + _state["quiet_until"] = 0.0 + # UDP is fire-and-forget: a datagram to a port with nothing bound succeeds at the send call, so + # "sent" here means "handed to the kernel", not "delivered". Saying plainly that it could not be + # confirmed is the J2 `present-unverified` precedent — reporting "I cannot check this" the same + # way as "this worked" would destroy the distinction that matters most. The unix-socket and HTTP + # transports do get an answer from the other end, so they report `sent` unqualified. + return "sent-unconfirmed" if cfg["kind"] == "syslog-udp" else "sent" + + +def status() -> dict: + """Configuration plus what this process has actually delivered. The counters are per-process: + a one-shot CLI run reports its own apply, while a long-lived console or MCP session accumulates. + A CLI run that failed said so on stderr; there is nowhere else for it to have been recorded.""" + cfg = configure() + return { + "configured": cfg["configured"], "kind": cfg["kind"], "target": cfg["redacted"], + "usable": bool(cfg["configured"] and cfg["kind"]), "reason": cfg["reason"], + "token_set": bool((os.environ.get(TOKEN_ENV) or "").strip()), + "attempted": _state["attempted"], "delivered": _state["delivered"], + "failed": _state["failed"], "suppressed": _state["suppressed"], + "last_error": _state["last_error"], + } + + +def test_payload() -> str: + """A test event is deliberately NOT shaped like an audit entry: it carries `kind` where a real + entry carries `action`, so a collector alerting on actions cannot be tripped by a connectivity + check, and it says in-band that it is not evidence of anything.""" + return json.dumps({ + "kind": "vpcopilot-audit-sink-test", "ts": runmeta.utc_now(), + "actor": runmeta.actor(), "host": runmeta.host(), "tool_version": __version__, + "note": "connectivity test from `vpcopilot audit-sink --send` — not an audit record, " + "and not written to any audit.log", + }) + + +def check(*, send: bool = False) -> dict: + """Answer "is the sink configured, is it usable, and does a delivery land" without touching a + run directory. Read-only with respect to the audit trail: the test event is never written to + `audit.log`, so asking the question does not add to the evidence.""" + st = status() + if not send or not st["usable"]: + st["delivery"] = "not attempted" if st["usable"] else "not attempted (no usable sink)" + return st + payload = test_payload() + # `force`, deliberately not `reset_state()`: clearing the counters would destroy a long-lived + # console's record of earlier failures as a side effect of asking whether the sink works now. + outcome = emit(payload, json.loads(payload), force=True) + st = status() + st["delivery"] = outcome + return st diff --git a/src/vpcopilot/cli.py b/src/vpcopilot/cli.py index 4e0f8ad..1a32543 100644 --- a/src/vpcopilot/cli.py +++ b/src/vpcopilot/cli.py @@ -716,6 +716,50 @@ def audit_backfill( rprint(Panel.fit(body, title="audit-backfill")) +@app.command(name="audit-sink") +def audit_sink_cmd( + send: bool = typer.Option(False, "--send", help="deliver one test event to the configured sink"), +): + """J3: is the off-box audit sink configured, usable, and reachable? + + `VPCOPILOT_AUDIT_SINK` ships every audit entry to a collector as it is written. A sink that is + misconfigured, or configured against a collector that is not listening, would otherwise be + invisible: the run succeeds either way and the only signal is one warning on stderr. This + reports the configuration without touching the network, and `--send` proves delivery. + + The test event is not an audit record and is written to no `audit.log` — asking whether the + sink works must not add to the evidence it carries.""" + from .audit_sink import check + + st = check(send=send) + if not st["configured"]: + rprint(Panel.fit(f"[yellow]{st['reason']}[/yellow]\n" + "[dim]set VPCOPILOT_AUDIT_SINK to https://…, syslog://…, stdout or off[/dim]", + title="audit-sink")) + raise typer.Exit() + if not st["usable"]: + rprint(Panel.fit(f"[red]unusable[/red]: {st['reason']}", title="audit-sink")) + raise typer.Exit(1) + body = (f"[bold]kind[/bold]: {st['kind']}\n[bold]target[/bold]: {st['target']}\n" + f"[bold]auth token[/bold]: {'set' if st['token_set'] else 'not set'}") + if send: + if st["delivery"] == "sent": + body += "\n[bold]test delivery[/bold]: [green]sent[/green]" + elif st["delivery"] == "sent-unconfirmed": + # Not dressed up as success: a UDP datagram to a port with nothing bound succeeds at + # the send call, so this says what was actually established and what was not. + body += ("\n[bold]test delivery[/bold]: [yellow]sent, unconfirmed[/yellow]\n" + "[dim]UDP cannot acknowledge — this proves the datagram left this host, not " + "that anything received it. Check the collector.[/dim]") + else: + body += (f"\n[bold]test delivery[/bold]: [red]{st['delivery']}[/red] " + f"({st['last_error'] or 'no detail'})") + body += "\n[dim]the local audit.log stays authoritative — a sink is a copy, never the record[/dim]" + rprint(Panel.fit(body, title="audit-sink")) + if send and not st["delivery"].startswith("sent"): + raise typer.Exit(1) + + @app.command(name="patches-list") def patches_list_cmd( out: str = typer.Option("out", help="output directory"), diff --git a/src/vpcopilot/console/app.py b/src/vpcopilot/console/app.py index 5a857a1..3fc9ebf 100644 --- a/src/vpcopilot/console/app.py +++ b/src/vpcopilot/console/app.py @@ -60,13 +60,17 @@ def _active_tag() -> str: return "default" SECRET_KEYS = {"ANTHROPIC_API_KEY", "OPENAI_API_KEY", "GEMINI_API_KEY", "XC_API_TOKEN", "GITHUB_TOKEN", - "VPCOPILOT_PROBE_PASS", "VPCOPILOT_PROBE_TOKEN"} + "VPCOPILOT_PROBE_PASS", "VPCOPILOT_PROBE_TOKEN", "VPCOPILOT_AUDIT_SINK_TOKEN"} MANAGED_KEYS = [ "ANTHROPIC_API_KEY", "OPENAI_API_KEY", "GEMINI_API_KEY", "OLLAMA_API_BASE", "XC_API_URL", "XC_API_TOKEN", "XC_NAMESPACE", "GITHUB_TOKEN", # Validation auth for an auth-protected target — the probe logs in so it can demonstrate the # exploit (loaded before every apply via load_dotenv, so a change here takes effect next run). "VPCOPILOT_PROBE_USER", "VPCOPILOT_PROBE_PASS", "VPCOPILOT_PROBE_LOGIN_PATH", "VPCOPILOT_PROBE_TOKEN", + # J3 — off-box audit sink. The URL is NOT secret: it is echoed back so the page can show what + # is configured, and `audit_sink.redact` keeps the credential-bearing path out of every other + # surface. The bearer token is secret and lives in SECRET_KEYS above. + "VPCOPILOT_AUDIT_SINK", "VPCOPILOT_AUDIT_SINK_TOKEN", ] app = FastAPI(title="virtual-patch-copilot console") @@ -465,6 +469,33 @@ def set_config(body: ConfigUpdate): return get_config() +@app.get("/api/audit-sink") +def audit_sink_status(): + """J3: what the off-box audit sink is set to, and what this process has managed to deliver. + + Read-only and network-free, because the Setup page polls it — the CLI twin is + `vpcopilot audit-sink`. The target is reported through `audit_sink.redact`, so a webhook URL + carrying its credential in the path is never rendered into the page.""" + load_dotenv(ENV_PATH, override=True) + from ..audit_sink import status + return status() + + +class SinkCheckReq(BaseModel): + send: bool = False + + +@app.post("/api/audit-sink") +def audit_sink_check(body: SinkCheckReq): + """Same answer, plus one test event delivered to the collector when `send` is set. + + This is the only endpoint here that reaches the network, and it writes nothing: the test event + is not an audit record and never touches an `audit.log`.""" + load_dotenv(ENV_PATH, override=True) + from ..audit_sink import check + return check(send=body.send) + + @app.get("/api/lbs") def list_lbs(): """HTTP load balancers in the namespace + their domains, so the console can offer a picker diff --git a/src/vpcopilot/console/static/index.html b/src/vpcopilot/console/static/index.html index 822ac91..04ee35a 100644 --- a/src/vpcopilot/console/static/index.html +++ b/src/vpcopilot/console/static/index.html @@ -24,6 +24,11 @@ .sev-critical{background:#fde7ea;color:#a1001b}.sev-high{background:#fde7e0;color:#a1440b} .sev-medium{background:#fff4d6;color:#7a5a00}.sev-low{background:#e8f0fe;color:#1b4fa1} .band{color:var(--ok);font-weight:600}.nob{color:var(--amber);font-weight:700} + /* `.bad` was used in the H2 dependency preview (parse errors, "could not check …") and never + defined, so every one of those rendered as ordinary text — a failure looking exactly like a + clean result, which is the one thing that preview exists to prevent. Found while validating J3, + whose sink status uses the same class. */ + .bad{color:#a1001b;font-weight:600} .st-found{color:var(--grey)}.st-mitigated{color:var(--amber);font-weight:700}.st-remediated{color:var(--ok);font-weight:700}.st-retired{color:#1b4fa1;font-weight:700} label { display:block; font-weight:600; margin:10px 0 4px; } input[type=text],input[type=number],select { width:100%; padding:8px; border:1px solid var(--line); border-radius:8px; font:inherit; } @@ -283,6 +288,16 @@ +

Audit event sink — ships every audit entry off the box as it is written

+
…
+ + +

Set VPCOPILOT_AUDIT_SINK above to + https://…, syslog://host:514, + syslog:///var/run/syslog, stdout — or + off to disable (a blank field cannot clear a key). The local + audit.log stays authoritative: a sink is a copy, never the record.

+

Agents & models — model-independent, per-agent in config/agents.yaml

@@ -313,7 +328,7 @@ if(id==="cure"){ loadResults().then(renderCure); loadHero(); } if(id==="retire"){ loadLedger(); loadAudit(); loadHero(); } if(id==="benchmark"){ loadBenchmarks(); } - if(id==="setup"){ loadConfig(); loadAgents(); reportNote(); } + if(id==="setup"){ loadConfig(); loadAgents(); reportNote(); checkSink(false); } } // The report is rebuilt server-side on every open, so it always reflects the LATEST run — name the // run dir it comes from, since a scan can repoint it (out-claude-vampi, demo/out, …). @@ -877,6 +892,27 @@

Full matrix

${head}`).join(""); } async function saveConfig(){ const u={}; document.querySelectorAll("[id^=cfg-]").forEach(i=>{ if(i.value) u[i.id.slice(4)]=i.value; }); await jpost("/api/config",{updates:u}); await loadConfig(); alert("Saved to .env"); } +// J3 — a sink that is configured and silently not delivering is the failure this panel exists to +// make visible: the run succeeds either way, so "unset" and "set but unreachable" must never +// render the same. `send` is the only thing here that touches the network. +async function checkSink(send){ const box=document.getElementById("sinkStatus"); + box.textContent = send ? "sending test event…" : "checking…"; + let s; try { s = send ? await jpost("/api/audit-sink",{send:true}) : await jget("/api/audit-sink"); } + catch(e){ box.innerHTML=''+esc(e.message)+''; return; } + if(!s.configured){ box.innerHTML=''+esc(s.reason)+''; return; } + if(!s.usable){ box.innerHTML='unusable — '+esc(s.reason)+''; return; } + const bits=[`${esc(s.kind)} → ${esc(s.target)}`, + s.token_set?'auth token set':'no auth token', + `delivered ${s.delivered}/${s.attempted}`+(s.failed?`, ${s.failed} failed`:"")+ + (s.suppressed?`, ${s.suppressed} suppressed (cooldown)`:"")]; + if(s.last_error) bits.push('last error: '+esc(s.last_error)+''); + // Three states, not two. UDP cannot acknowledge, so "the datagram left this host" must not be + // painted the same green as "the collector answered" (the J2 present-unverified precedent). + if(s.delivery==="sent") bits.push('test delivery: sent'); + else if(s.delivery==="sent-unconfirmed") bits.push('test delivery: sent, unconfirmed(UDP cannot acknowledge — this proves it left this host, not that anything received it)'); + else if(s.delivery) bits.push('test delivery: '+esc(s.delivery)+''); + box.innerHTML=bits.join(" · "); +} async function loadAgents(){ let a; try { a=await jget("/api/agents"); } catch(e){ wfAgents.textContent="error: "+e.message; return; } wfAgents.innerHTML = a.agents.map(x=>`

${esc(x.name)}

${esc(x.model)}
${esc(x.role)}
`).join('
→
'); } async function xcStatus(){ const out=document.getElementById("xcOut"); out.style.display="block"; out.textContent="checking…"; diff --git a/src/vpcopilot/runmeta.py b/src/vpcopilot/runmeta.py index 8239c3b..ab41baf 100644 --- a/src/vpcopilot/runmeta.py +++ b/src/vpcopilot/runmeta.py @@ -14,6 +14,7 @@ import os import socket import subprocess +import threading import uuid from datetime import datetime, timezone from pathlib import Path @@ -21,6 +22,9 @@ from . import __version__ +_MINT_LOCK = threading.Lock() # guards the run_id read-modify-write (see `run_id`) + + def _path(out_dir) -> Path: return Path(out_dir) / "run.json" @@ -61,20 +65,34 @@ def load(out_dir) -> dict: def _save(out_dir, meta: dict) -> None: p = _path(out_dir) p.parent.mkdir(parents=True, exist_ok=True) - tmp = p.with_suffix(f".json.tmp.{os.getpid()}") + # PID *and* thread id. With the pid alone, two threads in one process share the temp path, so + # the second `os.replace` finds it already moved and raises FileNotFoundError — reproduced 12 + # times out of 12 with 8 threads against a fresh out dir. The console starts every apply on its + # own daemon thread, and this runs inside `audit.record`, so the loser of that race fails to + # record a change it already made to a load balancer. + tmp = p.with_suffix(f".json.tmp.{os.getpid()}.{threading.get_ident()}") tmp.write_text(json.dumps(meta, indent=2)) os.replace(tmp, p) # atomic — every export joins on this file def run_id(out_dir) -> str: """The stable id for this run dir. Minted on first use and persisted, so a scan and a later - apply/retire against the same dir all stamp their audit entries with the same run.""" - meta = load(out_dir) - if meta.get("run_id"): - return meta["run_id"] - rid = uuid.uuid4().hex[:12] - _save(out_dir, {**meta, "run_id": rid, "created": utc_now()}) - return rid + apply/retire against the same dir all stamp their audit entries with the same run. + + The mint is a read-modify-write, and `audit.record` calls it on the console's per-apply daemon + threads. Unguarded, each thread read an empty `run.json`, minted its own id and stamped its own + entry with it — eight concurrent first writes produced eight different run_ids, so seven audit + entries carried a join key that `run.json` did not contain. The lock makes the mint + thread-atomic; single-threaded behaviour is unchanged. It is deliberately NOT a cross-process + guarantee — two processes first-writing the same dir at the same instant can still diverge, and + a file lock is the wrong trade for a case a scan already serializes.""" + with _MINT_LOCK: + meta = load(out_dir) + if meta.get("run_id"): + return meta["run_id"] + rid = uuid.uuid4().hex[:12] + _save(out_dir, {**meta, "run_id": rid, "created": utc_now()}) + return rid def git_provenance(repo) -> dict: diff --git a/tests/conftest.py b/tests/conftest.py index 7f2046c..5cf8292 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -114,6 +114,22 @@ def warmup(self): pass # no registry to warm for the fake +@pytest.fixture(autouse=True) +def _no_audit_sink(monkeypatch): + """J3: the audit sink is configured by environment, and `audit.record` runs in most of this + suite. A developer or CI box with `VPCOPILOT_AUDIT_SINK` exported would therefore ship the + suite's ~270 *fabricated* audit records — apply_waf, retire, rollback_failed — to a real + collector, where they are indistinguishable from records of real changes to a load balancer. + Measured at 267 datagrams on one full run. + + Documenting "unset it before running the tests" would be a guard that depends on someone + remembering. This is the structural version, and it also keeps the "no network in tests" + invariant true regardless of the shell the suite is launched from. Tests that exercise the sink + set the variable themselves; a test-level `monkeypatch.setenv` runs after this and wins.""" + for var in ("VPCOPILOT_AUDIT_SINK", "VPCOPILOT_AUDIT_SINK_TOKEN"): + monkeypatch.delenv(var, raising=False) + + @pytest.fixture def fake_xc(): return FakeXC() diff --git a/tests/test_audit_provenance.py b/tests/test_audit_provenance.py index 0ea05f1..ebe2c7f 100644 --- a/tests/test_audit_provenance.py +++ b/tests/test_audit_provenance.py @@ -2,12 +2,38 @@ WHO ran it, and as part of which run. An LB change that can't name its justification is not an audit record.""" import json +import threading import pytest from vpcopilot import apply, audit, runmeta +def test_concurrent_first_writes_do_not_lose_an_audit_record(tmp_path): + """Found while building J3's concurrency test, and pre-existing: `runmeta._save` named its temp + file by PID only, so two threads minting a run_id for a fresh out dir raced on it and the loser + got FileNotFoundError out of `os.replace` — raised from inside `audit.record`, i.e. a change + already made to a load balancer that could not be recorded. The console starts every apply on + its own daemon thread. Reproduced 12 times out of 12 before the fix.""" + errs = [] + + def go(i): + try: + audit.record(str(tmp_path), "apply_waf", lb="lab", finding_id=f"f{i}") + except Exception as e: # noqa: BLE001 + errs.append(f"{type(e).__name__}: {e}") + + ts = [threading.Thread(target=go, args=(i,)) for i in range(8)] + for t in ts: + t.start() + for t in ts: + t.join() + assert errs == [] + assert len(audit.load(str(tmp_path))) == 8 + # …and all eight share one run identity, which is what run_id exists to guarantee. + assert len({e["run_id"] for e in audit.load(str(tmp_path))}) == 1 + + @pytest.fixture(autouse=True) def _fast(monkeypatch, noop_sleep): import vpcopilot.engine as engine diff --git a/tests/test_audit_sink.py b/tests/test_audit_sink.py new file mode 100644 index 0000000..147bee4 --- /dev/null +++ b/tests/test_audit_sink.py @@ -0,0 +1,707 @@ +"""J3 — the off-box audit event sink. + +Offline throughout: every transport is faked. The syslog tests speak to a real socket object but +one this file owns, so nothing leaves the process, and the HTTP tests monkeypatch `httpx.Client` +the way `test_reconcile.py` does — which works only because `audit_sink` imports httpx inside the +function that uses it. + +The class of bug this file exists to pin is one shape: a sink that is configured and silently not +delivering, rendering exactly like a box with no sink at all.""" +from __future__ import annotations + +import errno +import json +import socket + +import pytest + +from vpcopilot import audit, audit_sink + + +@pytest.fixture(autouse=True) +def _clean_env(monkeypatch): + """No global env isolation exists in conftest, so a developer with the variable exported in + their shell would otherwise run the whole suite against their own collector.""" + monkeypatch.delenv(audit_sink.SINK_ENV, raising=False) + monkeypatch.delenv(audit_sink.TOKEN_ENV, raising=False) + audit_sink.reset_state() + yield + audit_sink.reset_state() + + +def _fake_httpx(monkeypatch, *, status=200, boom=None): + sent = {} + + class C: + def __init__(self, **kw): + sent["timeout"] = kw.get("timeout") + + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def post(self, url, content=None, headers=None): + if boom: + raise boom + sent.update(url=url, content=content, headers=headers) + return type("R", (), {"status_code": status})() + + import httpx + if boom and not isinstance(boom, Exception): + monkeypatch.setattr(httpx, "Client", boom) + else: + monkeypatch.setattr(httpx, "Client", C) + return sent + + +# ---------------------------------------------------------------- configuration + + +@pytest.mark.parametrize("raw,kind,reason_bit", [ + ("https://c.test/ingest", "https", ""), + ("http://c.test/ingest", "http", ""), + ("syslog://10.0.0.9:514", "syslog-udp", ""), + ("syslog://10.0.0.9", "syslog-udp", ""), + ("syslog:///var/run/syslog", "syslog-unix", ""), + ("stdout", "stdout", ""), + ("STDOUT", "stdout", ""), + (" https://c.test/x ", "https", ""), +]) +def test_every_documented_sink_spelling_parses(monkeypatch, raw, kind, reason_bit): + monkeypatch.setenv(audit_sink.SINK_ENV, raw) + cfg = audit_sink.configure() + assert cfg["configured"] and cfg["kind"] == kind + + +@pytest.mark.parametrize("raw", ["ftp://c.test", "wat", "https://", "syslog://", "syslog://:notaport"]) +def test_an_unparseable_sink_is_invalid_not_absent(monkeypatch, raw): + """The distinction the whole feature turns on: "we are not shipping anything" must not read the + same as "nothing is configured". An invalid value is `configured` with no usable kind.""" + monkeypatch.setenv(audit_sink.SINK_ENV, raw) + cfg = audit_sink.configure() + assert cfg["configured"] is True and cfg["kind"] is None and cfg["reason"] + + +@pytest.mark.parametrize("raw", [ + "https://[2001:db8::5:8088/ingest", # dropped bracket — the typo the IPv6 syntax invites + "syslog://[fe80::1", + "https://ho℀st/x", # netloc that fails NFKC validation +]) +def test_a_url_that_urlsplit_itself_rejects_is_reported_not_raised(tmp_path, monkeypatch, capsys, raw): + """Found by adversarial review. `urlsplit` RAISES on these, and unguarded that propagated out of + status() into a CLI traceback and a console 500 — while emit's bare except swallowed it with no + warning at all. The one input shape that produced silence on the apply path was the shape this + module exists to make loud.""" + monkeypatch.setenv(audit_sink.SINK_ENV, raw) + cfg = audit_sink.configure() # must not raise + assert cfg["configured"] is True and cfg["kind"] is None and "could not be parsed" in cfg["reason"] + assert audit_sink.status()["usable"] is False # …and neither must the surfaces' entry point + audit.record(str(tmp_path), "apply_waf", lb="lab") + assert len(audit.load(str(tmp_path))) == 1 + assert "audit sink not usable" in capsys.readouterr().err # it is loud, like every other bad value + assert raw not in audit_sink.status()["target"] # and the unparseable value is not echoed + + +@pytest.mark.parametrize("raw", ["off", "OFF", "none", "disabled"]) +def test_off_is_a_value_because_a_blank_field_cannot_clear_a_key(monkeypatch, raw): + """`console._write_env` drops falsy updates and the UI skips empty inputs, so clearing the box + cannot unset the variable. Without an explicit off, a sink enabled from the Setup page could + never be disabled from it.""" + monkeypatch.setenv(audit_sink.SINK_ENV, raw) + cfg = audit_sink.configure() + assert cfg["configured"] is False and cfg["kind"] is None + assert cfg["reason"] == "explicitly disabled" # …and distinguishable from never-set + monkeypatch.delenv(audit_sink.SINK_ENV) + assert audit_sink.configure()["reason"] == "no sink configured" + + +def test_configuration_is_read_every_call_not_cached(monkeypatch): + """POST /api/config writes .env and reloads it into os.environ of the RUNNING console. A + module-level constant would leave the Setup page lying until relaunch.""" + assert audit_sink.configure()["kind"] is None + monkeypatch.setenv(audit_sink.SINK_ENV, "stdout") + assert audit_sink.configure()["kind"] == "stdout" + + +def test_a_webhook_url_never_renders_its_path(monkeypatch): + """Slack/Teams/Splunk-HEC style webhooks carry the credential in the path. Only the origin is + ever shown — the same reason `reconcile.notify` logs the exception type and never its message.""" + secret = "https://hooks.test/services/T000/B000/XXXXSECRETXXXX" + monkeypatch.setenv(audit_sink.SINK_ENV, secret) + for surface in (audit_sink.configure()["redacted"], audit_sink.status()["target"], + audit_sink.check()["target"]): + assert "XXXXSECRETXXXX" not in surface and surface == "https://hooks.test/…" + + +def test_a_bare_origin_is_shown_whole(monkeypatch): + monkeypatch.setenv(audit_sink.SINK_ENV, "https://collector.test") + assert audit_sink.configure()["redacted"] == "https://collector.test" + + +def test_a_password_in_the_url_is_never_rendered_on_any_surface(monkeypatch): + """`urlsplit` puts userinfo in `netloc`, so redacting from netloc would print the password on + the CLI panel, the Setup page and the API response. Rebuilt from hostname/port instead.""" + monkeypatch.setenv(audit_sink.SINK_ENV, "https://svc:hunter2@collector.test:8443/ingest") + red = audit_sink.configure()["redacted"] + assert "hunter2" not in red and "svc" not in red + assert red == "https://…@collector.test:8443/…" + assert "hunter2" not in json.dumps(audit_sink.status()) + + +def test_an_ipv6_collector_is_reachable(tmp_path, monkeypatch): + """A hardcoded AF_INET socket cannot reach `syslog://[::1]:514` at all — the sink would read as + configured and deliver nothing, for ever. The family is resolved, not assumed.""" + rx = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM) + try: + rx.bind(("::1", 0)) + except OSError: + pytest.skip("no IPv6 loopback on this box") + rx.settimeout(2.0) + try: + monkeypatch.setenv(audit_sink.SINK_ENV, f"syslog://[::1]:{rx.getsockname()[1]}") + audit.record(str(tmp_path), "apply_waf", lb="lab", finding_id="v6") + payload = rx.recv(65535).decode().partition(": ")[2] + finally: + rx.close() + assert json.loads(payload)["finding_id"] == "v6" + + +# ---------------------------------------------------------------- fail-soft + + +def test_a_dead_collector_never_fails_an_apply_and_warns_exactly_once(tmp_path, monkeypatch, capsys): + """The acceptance criterion. `audit.record` is called on the rollback-failed path, so a sink + that could raise would turn "the LB may be left changed" into an unrecorded event.""" + monkeypatch.setenv(audit_sink.SINK_ENV, "https://down.test/x") + _fake_httpx(monkeypatch, boom=lambda **k: (_ for _ in ()).throw(ConnectionError("refused"))) + out = str(tmp_path) + for i in range(4): + audit.record(out, "rollback_failed", finding_id=f"f{i}", lb="lab") + assert len(audit.load(out)) == 4 # every entry still on disk + err = capsys.readouterr().err + assert err.count("audit sink delivery failed") == 1 # one warning, not four + assert "ConnectionError" in err + + +def test_the_warning_goes_to_stderr_never_stdout(tmp_path, monkeypatch, capsys): + """`mcp.serve()` repoints sys.stdout at stderr, but a warning that RELIED on that swap would + corrupt the protocol stream anywhere the swap is not in force.""" + monkeypatch.setenv(audit_sink.SINK_ENV, "https://down.test/x") + _fake_httpx(monkeypatch, boom=lambda **k: (_ for _ in ()).throw(ConnectionError("refused"))) + audit.record(str(tmp_path), "apply_waf", lb="lab") + cap = capsys.readouterr() + assert cap.out == "" and "audit sink delivery failed" in cap.err + + +def test_a_failing_sink_never_leaks_the_url_or_the_error_message(tmp_path, monkeypatch, capsys): + monkeypatch.setenv(audit_sink.SINK_ENV, "https://hooks.test/services/SECRETPATH") + _fake_httpx(monkeypatch, boom=lambda **k: (_ for _ in ()).throw(ConnectionError("connecting to https://hooks.test/services/SECRETPATH"))) + audit.record(str(tmp_path), "apply_waf", lb="lab") + err = capsys.readouterr().err + assert "SECRETPATH" not in err and "hooks.test" not in err + assert audit_sink.status()["last_error"] == "ConnectionError" + + +def test_an_http_error_status_is_a_failure_not_a_delivery(tmp_path, monkeypatch): + monkeypatch.setenv(audit_sink.SINK_ENV, "https://c.test/x") + _fake_httpx(monkeypatch, status=503) + audit.record(str(tmp_path), "apply_waf", lb="lab") + st = audit_sink.status() + assert st["delivered"] == 0 and st["failed"] == 1 and st["last_error"] == "OSError" + + +@pytest.mark.parametrize("status", [301, 302, 307, 308]) +def test_a_redirect_is_a_failure_because_the_body_was_never_re_sent(tmp_path, monkeypatch, status): + """Found by adversarial review, reproduced against a real 302 server. httpx does not follow + redirects by default, so a moved ingest path — or a proxy bouncing to an SSO page — swallowed + the whole audit stream while every surface read `delivered N/N`. + + It is refused rather than followed on purpose: the request carries the bearer token, and + chasing a Location to a host the operator never configured is how a credential travels.""" + monkeypatch.setenv(audit_sink.SINK_ENV, "https://c.test/x") + _fake_httpx(monkeypatch, status=status) + audit.record(str(tmp_path), "apply_waf", lb="lab") + st = audit_sink.status() + assert st["delivered"] == 0 and st["failed"] == 1 + + +def test_udp_reports_sent_unconfirmed_because_it_cannot_be_acknowledged(tmp_path, monkeypatch): + """A datagram to a port with nothing bound succeeds at the send call, so "sent" over UDP means + "handed to the kernel". Painting that the same green as an answered POST would destroy the + distinction that matters most — the J2 `present-unverified` precedent.""" + monkeypatch.setenv(audit_sink.SINK_ENV, "syslog://127.0.0.1:9") # discard port, nothing bound + assert audit_sink.check(send=True)["delivery"] == "sent-unconfirmed" + # …while a transport that DOES get an answer from the other end is unqualified. + monkeypatch.setenv(audit_sink.SINK_ENV, "https://c.test/x") + _fake_httpx(monkeypatch) + assert audit_sink.check(send=True)["delivery"] == "sent" + + +def test_both_surfaces_distinguish_unconfirmed_from_delivered(): + """A guard that lives on one surface is not a guard — and this one is a wording guard.""" + from pathlib import Path + root = Path(__file__).resolve().parents[1] + html = (root / "src/vpcopilot/console/static/index.html").read_text() + cli = (root / "src/vpcopilot/cli.py").read_text() + assert "sent-unconfirmed" in html and "sent, unconfirmed" in html + assert "sent-unconfirmed" in cli and "UDP cannot acknowledge" in cli + + +def test_an_invalid_sink_warns_once_and_delivers_nothing(tmp_path, monkeypatch, capsys): + monkeypatch.setenv(audit_sink.SINK_ENV, "ftp://nope.test") + for _ in range(3): + audit.record(str(tmp_path), "apply_waf", lb="lab") + err = capsys.readouterr().err + assert err.count("audit sink not usable") == 1 and "unsupported sink" in err + assert audit_sink.status()["attempted"] == 0 # nothing was ever tried + + +def test_a_dead_collector_is_paid_for_once_not_once_per_entry(tmp_path, monkeypatch): + """A hung collector at a 5s timeout, times the several entries one apply writes, is a stall the + acceptance ("never fails an apply") does not name but would certainly be. After a failure the + sink goes quiet for a cooldown instead of paying the timeout again.""" + monkeypatch.setenv(audit_sink.SINK_ENV, "https://down.test/x") + calls = [] + _fake_httpx(monkeypatch, boom=lambda **k: (calls.append(1), (_ for _ in ()).throw(ConnectionError("x")))[1]) + clock = [1000.0] + monkeypatch.setattr(audit_sink, "_now", lambda: clock[0]) + out = str(tmp_path) + for _ in range(5): + audit.record(out, "apply_waf", lb="lab") + assert len(calls) == 1 and audit_sink.status()["suppressed"] == 4 + + clock[0] += audit_sink.COOLDOWN_S + 1 # cooldown expires -> it tries again + audit.record(out, "apply_waf", lb="lab") + assert len(calls) == 2 + + +def test_delivery_resumes_after_the_collector_comes_back(tmp_path, monkeypatch): + monkeypatch.setenv(audit_sink.SINK_ENV, "https://c.test/x") + state = {"up": False} + + class C: + def __enter__(self): return self + def __exit__(self, *a): return False + def post(self, url, content=None, headers=None): + if not state["up"]: + raise ConnectionError("refused") + return type("R", (), {"status_code": 200})() + + import httpx + monkeypatch.setattr(httpx, "Client", lambda **k: C()) + clock = [1000.0] + monkeypatch.setattr(audit_sink, "_now", lambda: clock[0]) + out = str(tmp_path) + audit.record(out, "apply_waf", lb="lab") + assert audit_sink.status()["failed"] == 1 + state["up"] = True + clock[0] += audit_sink.COOLDOWN_S + 1 + audit.record(out, "apply_waf", lb="lab") + assert audit_sink.status()["delivered"] == 1 + + +# ---------------------------------------------------------------- the local log stays authoritative + + +def test_the_shipped_copy_is_byte_identical_to_the_line_on_disk(tmp_path, monkeypatch): + """Serialized once and handed to both destinations. Re-encoding the parsed dict would let key + order or float formatting drift between the evidence and the copy of it.""" + monkeypatch.setenv(audit_sink.SINK_ENV, "https://c.test/x") + sent = _fake_httpx(monkeypatch) + out = str(tmp_path) + audit.record(out, "apply_service_policy", finding_id="neg-pay-001", lb="lab", passed=True) + on_disk = (tmp_path / "audit.log").read_text().splitlines()[0] + assert sent["content"] == on_disk.encode() + assert json.loads(sent["content"])["finding_id"] == "neg-pay-001" + + +def test_nothing_is_delivered_when_the_local_write_fails(tmp_path, monkeypatch): + """The local log is authoritative, so an off-box event with no local counterpart must not + exist — it could never be reconciled against the run it belongs to.""" + monkeypatch.setenv(audit_sink.SINK_ENV, "https://c.test/x") + sent = _fake_httpx(monkeypatch) + monkeypatch.setattr("builtins.open", lambda *a, **k: (_ for _ in ()).throw(OSError("disk full"))) + with pytest.raises(OSError): + audit.record(str(tmp_path), "apply_waf", lb="lab") + assert sent == {} + + +def test_a_sink_changes_nothing_about_the_entry_itself(tmp_path, monkeypatch): + """No new key, no reordering, no envelope: the exporter, the golden expectations and + `export --verify` all read this file.""" + out = str(tmp_path) + audit.record(out, "apply_waf", lb="lab", finding_id="f1") + without = (tmp_path / "audit.log").read_text() + (tmp_path / "audit.log").unlink() + monkeypatch.setenv(audit_sink.SINK_ENV, "stdout") + audit.record(out, "apply_waf", lb="lab", finding_id="f1") + a, b = json.loads(without), json.loads((tmp_path / "audit.log").read_text()) + assert list(a) == list(b) and {k: a[k] for k in a if k != "ts"} == {k: b[k] for k in b if k != "ts"} + + +def test_the_sink_never_writes_an_audit_record_of_its_own(tmp_path, monkeypatch): + """A delivery record inside `audit.record` would recurse, and an entry-count-changing side + effect is the bug J4's no-op check exists to prevent.""" + monkeypatch.setenv(audit_sink.SINK_ENV, "https://down.test/x") + _fake_httpx(monkeypatch, boom=lambda **k: (_ for _ in ()).throw(ConnectionError("x"))) + out = str(tmp_path) + audit.record(out, "apply_waf", lb="lab") + assert len(audit.load(out)) == 1 + assert audit.load(out)[0]["action"] == "apply_waf" + + +def test_identity_still_cannot_be_forged_with_a_sink_configured(tmp_path, monkeypatch): + monkeypatch.setenv(audit_sink.SINK_ENV, "https://c.test/x") + sent = _fake_httpx(monkeypatch) + audit.record(str(tmp_path), "apply_waf", lb="lab", actor="someone-else", run_id="deadbeef") + shipped = json.loads(sent["content"]) + assert shipped["actor"] != "someone-else" and shipped["run_id"] != "deadbeef" + + +# ---------------------------------------------------------------- transports + + +def test_the_http_sink_sends_json_with_the_bearer_token_when_set(tmp_path, monkeypatch): + monkeypatch.setenv(audit_sink.SINK_ENV, "https://c.test/ingest") + monkeypatch.setenv(audit_sink.TOKEN_ENV, "s3cr3t") + sent = _fake_httpx(monkeypatch) + audit.record(str(tmp_path), "apply_waf", lb="lab") + assert sent["url"] == "https://c.test/ingest" + assert sent["headers"]["content-type"] == "application/json" + assert sent["headers"]["authorization"] == "Bearer s3cr3t" + assert sent["timeout"] == audit_sink.TIMEOUT_S + + +def test_the_http_sink_sends_no_auth_header_when_no_token_is_set(tmp_path, monkeypatch): + monkeypatch.setenv(audit_sink.SINK_ENV, "https://c.test/ingest") + sent = _fake_httpx(monkeypatch) + audit.record(str(tmp_path), "apply_waf", lb="lab") + assert "authorization" not in sent["headers"] + + +def test_the_stdout_sink_writes_one_json_line_resolved_at_call_time(tmp_path, monkeypatch, capsys): + monkeypatch.setenv(audit_sink.SINK_ENV, "stdout") + audit.record(str(tmp_path), "apply_waf", lb="lab", finding_id="f1") + out = capsys.readouterr().out.strip().splitlines() + assert len(out) == 1 and json.loads(out[0])["finding_id"] == "f1" + + +def test_a_stdout_sink_cannot_corrupt_the_mcp_protocol_stream(tmp_path, monkeypatch): + """K1 points `sys.stdout` at stderr for the server's lifetime precisely so a stray write beneath + it cannot land on the message stream. That guarantee has to cover this new writer, which is + only true because the sink resolves `sys.stdout` when it writes rather than at import.""" + import io + import sys as _sys + + from vpcopilot import mcp + monkeypatch.setenv(audit_sink.SINK_ENV, "stdout") + frames, err = io.StringIO(), io.StringIO() + monkeypatch.setattr(_sys, "stderr", err) + req = json.dumps({"jsonrpc": "2.0", "id": 1, "method": "ping"}) + "\n" + mcp.serve(io.StringIO(req), frames, enable_writes=False, swap_stdout=True) + audit_sink.reset_state() + + # Now emit while the swap is in force, exactly as a write tool would. + saved = _sys.stdout + _sys.stdout = _sys.stderr + try: + audit.record(str(tmp_path), "apply_waf", lb="lab") + finally: + _sys.stdout = saved + assert "apply_waf" in err.getvalue() # the event landed on stderr… + assert "apply_waf" not in frames.getvalue() # …and never on the protocol stream + assert json.loads(frames.getvalue().strip())["id"] == 1 + + +def _udp_listener(): + rx = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + rx.bind(("127.0.0.1", 0)) + rx.settimeout(2.0) + return rx, rx.getsockname()[1] + + +def test_the_syslog_sink_emits_a_wellformed_rfc3164_frame(tmp_path, monkeypatch): + rx, port = _udp_listener() + try: + monkeypatch.setenv(audit_sink.SINK_ENV, f"syslog://127.0.0.1:{port}") + audit.record(str(tmp_path), "apply_service_policy", lb="lab", finding_id="f1") + frame = rx.recvfrom(65535)[0].decode() + finally: + rx.close() + assert frame.startswith(f"<{audit_sink.SYSLOG_PRI}>") + head, _, payload = frame.partition(": ") + assert "vpcopilot[" in head + assert json.loads(payload)["finding_id"] == "f1" # the JSON survives the framing intact + + +def test_an_oversize_entry_degrades_to_an_envelope_that_says_so(tmp_path, monkeypatch): + """Measured on macOS: a `/var/run/syslog` unix socket has SO_SNDBUF 2048 and refuses anything + larger with EMSGSIZE; UDP walls around 9216. The entries that overflow are the interesting ones + — `drift_detected` carries a whole field-level diff — so dropping them would lose exactly the + records worth shipping. The kernel sets the limit; nothing here hardcodes one.""" + rx, port = _udp_listener() + try: + monkeypatch.setenv(audit_sink.SINK_ENV, f"syslog://127.0.0.1:{port}") + real = socket.socket + + class Refusing: + def __init__(self, *a, **k): + self._s = real(*a, **k) + self._first = True + + def sendto(self, data, dest): + if self._first: # the kernel's answer to a too-large datagram + self._first = False + raise OSError(errno.EMSGSIZE, "Message too long") + return self._s.sendto(data, dest) + + def __getattr__(self, n): + return getattr(self._s, n) + + monkeypatch.setattr(audit_sink.socket, "socket", Refusing) + audit.record(str(tmp_path), "drift_detected", lb="lab", finding_id="f1", + changes={"k%d" % i: "v" * 80 for i in range(40)}) + payload = rx.recvfrom(65535)[0].decode().partition(": ")[2] + finally: + rx.close() + env = json.loads(payload) # valid JSON, not a truncated fragment + assert env["audit_sink_oversize"] is True and env["action"] == "drift_detected" + assert env["finding_id"] == "f1" and env["bytes"] > 1024 + assert "audit.log" in env["note"] + assert audit_sink.status()["delivered"] == 1 # …and it counts as delivered, not as a failure + + +def test_an_oversize_entry_that_still_will_not_fit_is_a_failure(tmp_path, monkeypatch, capsys): + monkeypatch.setenv(audit_sink.SINK_ENV, "syslog://127.0.0.1:9") + + class AlwaysRefusing: + def __init__(self, *a, **k): pass + def settimeout(self, t): pass + def close(self): pass + def sendto(self, data, dest): raise OSError(errno.EMSGSIZE, "Message too long") + + monkeypatch.setattr(audit_sink.socket, "socket", AlwaysRefusing) + audit.record(str(tmp_path), "apply_waf", lb="lab") + assert audit_sink.status()["failed"] == 1 + assert "audit sink delivery failed" in capsys.readouterr().err + + +def test_the_unix_syslog_transport_reaches_a_real_socket(tmp_path, monkeypatch): + """The `syslog:///var/run/syslog` shape, against a socket this test owns rather than the box's + syslogd — same code path, nothing leaves the process.""" + import shutil + import tempfile + # NOT tmp_path: sun_path is capped at ~104 bytes and pytest's per-test directory blows it. + short = tempfile.mkdtemp(prefix="vpcs") + sock_path = f"{short}/s" + rx = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) + rx.bind(sock_path) + rx.settimeout(2.0) + try: + monkeypatch.setenv(audit_sink.SINK_ENV, f"syslog://{sock_path}") + assert audit_sink.configure()["kind"] == "syslog-unix" + audit.record(str(tmp_path), "retire", lb="lab", finding_id="f1") + payload = rx.recv(65535).decode().partition(": ")[2] + finally: + rx.close() + shutil.rmtree(short, ignore_errors=True) + assert json.loads(payload)["action"] == "retire" + + +# ---------------------------------------------------------------- the check surface + + +def test_check_reports_configuration_without_touching_the_network(monkeypatch): + monkeypatch.setenv(audit_sink.SINK_ENV, "https://c.test/x") + + def boom(**k): + raise AssertionError("check() must not deliver without --send") + import httpx + monkeypatch.setattr(httpx, "Client", boom) + st = audit_sink.check() + assert st["usable"] and st["kind"] == "https" and st["delivery"] == "not attempted" + + +def test_check_send_delivers_a_test_event_that_is_not_an_audit_record(tmp_path, monkeypatch): + monkeypatch.setenv(audit_sink.SINK_ENV, "https://c.test/x") + sent = _fake_httpx(monkeypatch) + st = audit_sink.check(send=True) + assert st["delivery"] == "sent" + body = json.loads(sent["content"]) + assert body["kind"] == "vpcopilot-audit-sink-test" and "action" not in body + assert "not an audit record" in body["note"] + assert not (tmp_path / "audit.log").exists() # asking the question adds no evidence + + +def test_check_send_reports_a_dead_collector_rather_than_a_clean_bill(monkeypatch): + monkeypatch.setenv(audit_sink.SINK_ENV, "https://down.test/x") + _fake_httpx(monkeypatch, boom=lambda **k: (_ for _ in ()).throw(ConnectionError("x"))) + st = audit_sink.check(send=True) + assert st["delivery"] == "failed" and st["last_error"] == "ConnectionError" + + +def test_check_send_is_not_silenced_by_an_earlier_failures_cooldown(monkeypatch): + """The check exists to answer "is it working NOW". Reporting `suppressed` because an unrelated + entry failed a minute ago would be the tool declining to look.""" + monkeypatch.setenv(audit_sink.SINK_ENV, "https://c.test/x") + audit_sink._state.update(quiet_until=audit_sink._now() + 999, failed=1, warned=True) + _fake_httpx(monkeypatch) + assert audit_sink.check(send=True)["delivery"] == "sent" + + +def test_check_send_does_not_erase_the_record_of_earlier_failures(monkeypatch): + """A long-lived console accumulates delivery history. Answering "does it work now" by wiping + the counters would destroy the evidence that it had been failing — the operator would click a + button to diagnose a problem and delete the symptom.""" + monkeypatch.setenv(audit_sink.SINK_ENV, "https://c.test/x") + audit_sink._state.update(failed=7, attempted=9, delivered=2, last_error="ConnectError", + quiet_until=audit_sink._now() + 999, warned=True) + _fake_httpx(monkeypatch) + st = audit_sink.check(send=True) + assert st["delivery"] == "sent" + assert st["failed"] == 7 and st["delivered"] == 3 and st["attempted"] == 10 + + +def test_a_successful_test_event_re_arms_a_suppressed_sink(monkeypatch): + """Proving the collector is back is a reason to stop suppressing: otherwise the operator sees + "sent" and the next real audit entry is still silently dropped for the rest of the cooldown.""" + monkeypatch.setenv(audit_sink.SINK_ENV, "https://c.test/x") + audit_sink._state.update(quiet_until=audit_sink._now() + 999) + _fake_httpx(monkeypatch) + audit_sink.check(send=True) + assert audit_sink._state["quiet_until"] == 0.0 + + +def test_one_warning_survives_concurrent_failures_from_console_threads(tmp_path, monkeypatch, capsys): + """The console runs each apply on its own daemon thread against this shared module. Without a + check-and-set the "log one warning" guarantee degrades to "log one warning per racing thread".""" + import threading + monkeypatch.setenv(audit_sink.SINK_ENV, "https://down.test/x") + gate = threading.Barrier(8) + + class C: + def __enter__(self): return self + def __exit__(self, *a): return False + def post(self, *a, **k): + gate.wait(timeout=5) # every thread fails at the same instant + raise ConnectionError("refused") + + import httpx + monkeypatch.setattr(httpx, "Client", lambda **k: C()) + monkeypatch.setattr(audit_sink, "_now", lambda: 1000.0) # nobody is suppressed by a cooldown + ts = [threading.Thread(target=audit.record, args=(str(tmp_path), "apply_waf"), kwargs={"lb": "lab"}) + for _ in range(8)] + for t in ts: + t.start() + for t in ts: + t.join() + assert capsys.readouterr().err.count("audit sink delivery failed") == 1 + assert len(audit.load(str(tmp_path))) == 8 # …and every entry still reached disk + + +def test_check_on_an_unusable_sink_says_so_and_sends_nothing(monkeypatch): + monkeypatch.setenv(audit_sink.SINK_ENV, "ftp://nope.test") + st = audit_sink.check(send=True) + assert st["configured"] and not st["usable"] and "no usable sink" in st["delivery"] + + +# ---------------------------------------------------------------- surfaces + + +def test_the_sink_keys_are_on_the_setup_page_and_the_token_is_secret(): + """The acceptance names the ⚙ Setup page, which renders exactly MANAGED_KEYS.""" + from vpcopilot.console import app + assert audit_sink.SINK_ENV in app.MANAGED_KEYS + assert audit_sink.TOKEN_ENV in app.MANAGED_KEYS + assert audit_sink.TOKEN_ENV in app.SECRET_KEYS # a bearer token is never echoed back + assert audit_sink.SINK_ENV not in app.SECRET_KEYS # …the URL is, so the page can show it + + +def test_the_console_endpoint_and_the_cli_call_the_same_module_function(monkeypatch): + """One module function, two surfaces — a guard or a readout that lives on one is neither.""" + from fastapi.testclient import TestClient + + from vpcopilot.console import app as A + monkeypatch.setenv(audit_sink.SINK_ENV, "stdout") + monkeypatch.setattr(A, "ENV_PATH", A.Path("/nonexistent-vpcopilot-env")) + c = TestClient(A.app) + got = c.get("/api/audit-sink").json() + assert got["kind"] == "stdout" and got["usable"] is True + assert got == {k: v for k, v in audit_sink.status().items()} + + +def test_the_console_status_endpoint_sends_nothing(monkeypatch): + from fastapi.testclient import TestClient + + from vpcopilot.console import app as A + monkeypatch.setenv(audit_sink.SINK_ENV, "https://c.test/x") + monkeypatch.setattr(A, "ENV_PATH", A.Path("/nonexistent-vpcopilot-env")) + + def boom(**k): + raise AssertionError("GET /api/audit-sink must not reach the network — the page polls it") + import httpx + monkeypatch.setattr(httpx, "Client", boom) + assert TestClient(A.app).get("/api/audit-sink").status_code == 200 + + +def test_the_console_post_can_deliver_a_test_event(monkeypatch): + from fastapi.testclient import TestClient + + from vpcopilot.console import app as A + monkeypatch.setenv(audit_sink.SINK_ENV, "https://c.test/x") + monkeypatch.setattr(A, "ENV_PATH", A.Path("/nonexistent-vpcopilot-env")) + sent = _fake_httpx(monkeypatch) + r = TestClient(A.app).post("/api/audit-sink", json={"send": True}).json() + assert r["delivery"] == "sent" and json.loads(sent["content"])["kind"] == "vpcopilot-audit-sink-test" + + +def test_the_setup_page_actually_calls_the_endpoint(): + """A surface nothing calls is not a surface (the I1 precedent).""" + from pathlib import Path + html = (Path(__file__).resolve().parents[1] + / "src/vpcopilot/console/static/index.html").read_text() + assert "/api/audit-sink" in html and "sinkStatus" in html and "checkSink(" in html + + +def test_every_css_class_the_console_uses_to_mark_a_failure_is_actually_defined(): + """Found live while validating J3, and it was already broken: `class="bad"` marks a failure in + the H2 dependency preview (a manifest that would not parse, a package OSV could not be asked + about) and the class was defined NOWHERE, so all of it rendered as ordinary text. A warning + styled like body copy is the same defect that preview exists to prevent.""" + import re + from pathlib import Path + html = (Path(__file__).resolve().parents[1] + / "src/vpcopilot/console/static/index.html").read_text() + used = set(re.findall(r'class="([a-z-]+)"', html)) & {"bad", "band", "nob", "muted", "sub", "mono"} + for cls in sorted(used): + assert re.search(rf"\.{cls}\s*{{", html), f'.{cls} is used in markup but never defined in CSS' + + +def test_the_suite_cannot_ship_fabricated_audit_records_to_a_real_collector(): + """Found by adversarial review: with `VPCOPILOT_AUDIT_SINK` exported in the shell, a full run + shipped **267 datagrams** of invented records — apply_waf, retire, rollback_failed — to the + developer's real collector, where they are indistinguishable from records of real changes. + Documenting "unset it first" is a guard that depends on remembering; conftest clears it.""" + from pathlib import Path + conftest = (Path(__file__).resolve().parents[1] / "tests/conftest.py").read_text() + assert "autouse=True" in conftest and audit_sink.SINK_ENV in conftest + # The fixture is autouse, so by the time this test body runs the variable is already gone + # whatever the launching shell had in it. + assert audit_sink.configure()["configured"] is False + + +def test_the_cli_command_exists_and_is_kebab_case(): + from vpcopilot import cli + names = {c.name for c in cli.app.registered_commands} + assert "audit-sink" in names + + +def test_the_documented_limit_is_actually_written_down(): + """This project states its limits rather than leaving them to be discovered (the J1/J4 + precedent). A sink is a copy, not a second source of truth, and the doc has to say so.""" + from pathlib import Path + txt = (Path(__file__).resolve().parents[1] / "docs/AUDIT.md").read_text() + assert "VPCOPILOT_AUDIT_SINK" in txt + assert "authoritative" in txt and "best-effort" in txt