diff --git a/.agents/notes/implemented/2026-10-27-recovery-must-not-claim-certainty.md b/.agents/notes/implemented/2026-10-27-recovery-must-not-claim-certainty.md new file mode 100644 index 0000000..e416736 --- /dev/null +++ b/.agents/notes/implemented/2026-10-27-recovery-must-not-claim-certainty.md @@ -0,0 +1,40 @@ +--- +title: Recovery must not claim certainty +status: implemented +date: 2026-10-27 +scope: stack +--- + +## Decision + +Record remediation step starts and results durably, and classify interrupted +steps as completed, unknown, or not started. Redact a step's whole output, then +keep its last 10,000 characters (omit it past a megabyte). Reserve signed +inbound requests with renewable SQLite leases, recording completion only after +handling. Treat incomplete posture measurements as unverifiable, and ASK the +instance metadata service, as the AWS CLI does, when nothing names an AWS +credential: a role it offers is measured; no role, another vendor's service or +nothing there is no credential. A writing node with a declared radius refuses +an unverified measurement under enforce. + +Amended in review before merge: the first version assumed metadata credentials +whenever nothing named one, which read every laptop and every probe on an +internal network as `unverifiable` and stopped a declared writing node from +starting; it omitted any output past 10,000 bytes, tail included; and it cut +the pipe's explanatory comments to fit its line ceiling. The pipe's two unused +signature methods went instead. + +## Why + +A crash between a side effect and its record cannot establish that it did not +happen. Concurrent requests must not both enter the pipeline while the first is +still processing, and a failed permission check cannot prove a declared radius. + +## Consequences + +- An unknown remediation step requires reconciliation before another attempt. +- A lease can expire after a crash, allowing retries but not exactly-once external effects. +- Patterns do not catch every secret, and proposals retain executable commands. +- Declared writing nodes may need corrected tooling or an explicit metadata setting before starting. +- The pipe gains a `pending_signatures` table. A rollback to an earlier pipe + ignores it: nothing fails to open, and a lease left in it simply expires. diff --git a/docs/containment.md b/docs/containment.md index 976fa3e..f12f972 100644 --- a/docs/containment.md +++ b/docs/containment.md @@ -19,9 +19,17 @@ a real model cannot be steered into emitting a harmful one. > defaults to `enforce` in the code, so the deploy composes now spell it out rather than leaving the reader to find it; > a `danger-only` runner records its blast radius; an identity that cannot be asked is reported *unverifiable*, never confirmed. > A credential whose measuring tool is not in the image is *unmeasurable* (2026-10-12): a readonly runner refuses to start -> under `enforce`, a writing runner starts and pages, and the hourly selftest fails until the tool is back — ten days of +> under `enforce`, as does a writing runner with a declared radius; an undeclared writing runner starts and pages, and the hourly selftest fails until the tool is back — ten days of > `kubectl: not found` had passed as `unverifiable` and `recorded` before this. The measurement is served on `GET /v1/posture` > and cited in every run's audit record. +> An unverifiable writing runner with a declared radius also refuses under `enforce`. With no explicit AWS source, the +> instance metadata service is asked directly whether it offers a role: one it offers is measured like any other source. + +> **Recovery is not exactly-once (2026-10-27).** Concurrent signed hook requests use a durable, renewable processing +> lease; an in-flight duplicate receives 503 and can retry, while completed signatures are deduplicated. A crash after +> an external side effect but before the completion record can still cause that effect to repeat. Remediation records +> each step's start and result; a started step without a durable result is unknown and needs operator reconciliation. +> Output is redacted whole before it is cut to its last 10,000 characters (omitted past a megabyte); redaction is by pattern and can miss secrets, and original proposal commands remain stored. ## What holds, and what only helps diff --git a/hookprobe/README.md b/hookprobe/README.md index 744bded..ade8f9f 100644 --- a/hookprobe/README.md +++ b/hookprobe/README.md @@ -178,11 +178,18 @@ always reach. Two things about the word *default*, before the layers: over the dangerous actions) and compares the answer with the declaration: `readonly-confirmed`, `wider-than-declared` (refuses to start under the default `HOOKPROBE_POSTURE_CHECK=enforce`), `unverifiable`, `unmeasurable` - (the measuring tool is not in the image: a readonly node refuses to start - under `enforce`, a writing node starts and pages), `recorded`, + (the measuring tool is not in the image: a readonly node or a writing node + with a declared radius refuses to start under `enforce`; an undeclared + writing node starts and pages), `recorded`, `no-credentials`. The verdict is at `GET /v1/posture` and at the top of every run's audit, so "read-only" is a measurement with a date rather than a sentence in this file. + An unverifiable writing node with a declared radius also refuses under + `enforce`. When nothing names an AWS credential, the instance metadata + service is asked directly, the way the AWS CLI asks it (no proxy, a second + per question): a role it offers is measured like any other source; no role, + another vendor's service at the address, or nothing there is no credential; + `AWS_EC2_METADATA_DISABLED=true` skips the question. - **Widening is a decision, and it is signed.** Under `danger-only` the guard keeps only the rules no credential scope can undo — `rm -rf`, `mkfs`/`dd` onto a device, fork bombs, container runtimes, `terraform destroy`, @@ -198,6 +205,13 @@ always reach. Two things about the word *default*, before the layers: (`HOOKPROBE_REMEDIATION_COOLDOWN_SECONDS`, 15 minutes, 0 disables) is held rather than changed twice with nobody looking in between. Both are recorded in the run's audit with the person's action beside them. + Execution checkpoints distinguish completed steps, a started step with an + unknown outcome, and steps not started after an interruption; reconcile an + unknown outcome before retrying. A step's output is redacted whole and then + cut to its last 10,000 characters, so a secret the cut would split was seen + whole by the redactor; past a megabyte it is omitted instead. Redaction is by + pattern, not a security boundary, and the stored proposal retains its + original commands. 1. **Credentials** — mount query-only credentials (read-only kubeconfig, Prometheus/Loki endpoints, viewer tokens). This is the real boundary; diff --git a/hookprobe/hookprobe/posture.py b/hookprobe/hookprobe/posture.py index e0ae484..176cb8c 100644 --- a/hookprobe/hookprobe/posture.py +++ b/hookprobe/hookprobe/posture.py @@ -48,6 +48,8 @@ import logging import os import time +import urllib.error +import urllib.request from collections.abc import Awaitable, Callable from pathlib import Path from typing import Any @@ -155,7 +157,19 @@ def kubeconfig_path() -> Path | None: def aws_configured() -> bool: - if os.environ.get("AWS_ACCESS_KEY_ID"): + # A profile NAME alone is not a source: it names a section of a file, and + # the files are what the loop below looks for. An instance metadata service + # is not configured at all — it is asked (see metadata_offers_credentials). + if any( + os.environ.get(name) + for name in ( + "AWS_ACCESS_KEY_ID", + "AWS_ROLE_ARN", + "AWS_WEB_IDENTITY_TOKEN_FILE", + "AWS_CONTAINER_CREDENTIALS_RELATIVE_URI", + "AWS_CONTAINER_CREDENTIALS_FULL_URI", + ) + ): return True for var, default in (("AWS_SHARED_CREDENTIALS_FILE", "~/.aws/credentials"), ("AWS_CONFIG_FILE", "~/.aws/config")): if Path(os.environ.get(var) or default).expanduser().is_file(): @@ -190,11 +204,60 @@ async def check_kube(run: Runner) -> dict[str, Any]: return record +def metadata_offers_credentials(timeout: float = 1.0) -> bool: + """Whether an AWS instance metadata service here hands out a role's credentials. + + When nothing names a credential this is the source left, and the AWS CLI + asks it the same two questions: a token (IMDSv2), then the role listing. + A role name in the answer is a credential; a 404 is an instance with no role; + anything else is no AWS service at all. A cloud VM of another vendor answers + at the same address, and an answer of any kind is not a credential — GitHub's + runners are Azure VMs, and counting any answer made the image refuse to start + there. No proxy, on purpose: the probes' egress proxy answers for addresses + it refuses, and the question is what this container reaches by itself. + """ + base = ( + os.environ.get("AWS_EC2_METADATA_SERVICE_ENDPOINT") + or ( + "http://[fd00:ec2::254]" + if os.environ.get("AWS_EC2_METADATA_SERVICE_ENDPOINT_MODE", "").lower() == "ipv6" + else "http://169.254.169.254" + ) + ).rstrip("/") + opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) + token: dict[str, str] = {} + put = urllib.request.Request( + base + "/latest/api/token", method="PUT", headers={"X-aws-ec2-metadata-token-ttl-seconds": "60"} + ) + try: + with opener.open(put, timeout=timeout) as answer: # nosec B310 — a fixed link-local address + token = {"X-aws-ec2-metadata-token": answer.read(256).decode("ascii", "replace").strip()} + except (OSError, ValueError): + pass # IMDSv1 only, another vendor's service, or nothing: the listing decides + listing = urllib.request.Request(base + "/latest/meta-data/iam/security-credentials/", headers=token) + try: + with opener.open(listing, timeout=timeout) as answer: # nosec B310 + return bool(answer.read(1024).strip()) + except (OSError, ValueError): # HTTPError included: 404 is no role, anything else no AWS + return False + + async def check_aws(run: Runner) -> dict[str, Any]: if not aws_configured(): - return {"present": False} + # Asked, not assumed. Assuming made every laptop and every probe on an + # internal network `unverifiable`, and a writing node with a declared + # radius refused to start over a service that cannot exist there. + if os.environ.get("AWS_EC2_METADATA_DISABLED", "").lower() == "true": + return {"present": False} + if not await asyncio.to_thread(metadata_offers_credentials): + return {"present": False} + # A role is offered: its credentials are measured like any other source. record: dict[str, Any] = {"present": True, "identity": None, "allowed": [], "errors": [], "unverifiable": False} code, out, err = await run(["aws", "sts", "get-caller-identity", "--output", "json"]) + if code != 0 and "Unable to locate credentials" in (err or out): + # The CLI looked everywhere it looks and found nothing: an instance with + # no role, say. That is an absence, not a measurement that failed. + return {"present": False} if code != 0: record["errors"].append(f"get-caller-identity: {(err or out).strip()[:200]}") record["unverifiable"] = True @@ -228,7 +291,29 @@ async def check_aws(run: Runner) -> dict[str, Any]: record["errors"].append(f"simulate-principal-policy: {(err or out).strip()[:200]}") record["unverifiable"] = True else: - record["allowed"] = aws_allowed(out) + try: + results = json.loads(out)["EvaluationResults"] + if not isinstance(results, list): + raise ValueError("invalid results") + record["allowed"] = sorted( + { + str(item.get("EvalActionName")) + for item in results + if isinstance(item, dict) and item.get("EvalDecision") == "allowed" and item.get("EvalActionName") + } + ) + if any( + not isinstance(item, dict) + or item.get("EvalDecision") not in {"allowed", "explicitDeny", "implicitDeny"} + for item in results + ): + raise ValueError("invalid decision") + decisions = {str(item["EvalActionName"]) for item in results if item.get("EvalDecision")} + if not set(DANGEROUS_AWS_ACTIONS).issubset(decisions): + raise ValueError("incomplete simulation") + except (ValueError, KeyError, TypeError, AttributeError): + record["errors"].append("simulate-principal-policy: invalid or incomplete results") + record["unverifiable"] = True return record @@ -281,19 +366,21 @@ def verdict(bash_guard: str, kube: dict[str, Any], aws: dict[str, Any], declared # nothing to answer that with and this stays an observation, which is # what it was for a year. Declared, `enforce` means the same thing on # both postures: the credential may not exceed the declaration. - if declared is None: - return "unmeasurable" if missing else "recorded" if beyond(kube, aws, declared): return "wider-than-declared" - return "unmeasurable" if missing else "within-declared-radius" - if not kube.get("present") and not aws.get("present"): - return "no-credentials" + if missing: + return "unmeasurable" + if kube.get("errors") or aws.get("unverifiable"): + return "unverifiable" + return "recorded" if declared is None else "within-declared-radius" if kube.get("mutating") or aws.get("allowed"): return "wider-than-declared" if missing: return "unmeasurable" if kube.get("errors") or aws.get("unverifiable"): return "unverifiable" + if not kube.get("present") and not aws.get("present"): + return "no-credentials" return "readonly-confirmed" @@ -395,13 +482,18 @@ async def on_startup( if alarm is not None: with contextlib.suppress(Exception): # the page is a courtesy; the verdict below is the decision await alarm(f"posture UNMEASURABLE — {text}") - if bash_guard == "readonly" and mode == "enforce": + if (bash_guard == "readonly" or declared is not None) and mode == "enforce": logger.error("REFUSING TO START — %s", text) raise PostureViolation( f"posture cannot be measured as declared {bash_guard!r}: {missing} not in this image" ) logger.error("%s", text) elif record["verdict"] == "unverifiable": + if alarm is not None: + with contextlib.suppress(Exception): + await alarm(f"posture UNVERIFIABLE — {line}") + if bash_guard != "readonly" and declared is not None and mode == "enforce": + raise PostureViolation(f"declared writing posture could not be verified: {line}") logger.warning( "posture could not be fully verified: %s · %s", line, diff --git a/hookprobe/hookprobe/remediation.py b/hookprobe/hookprobe/remediation.py index e4d4c65..c6ee78f 100644 --- a/hookprobe/hookprobe/remediation.py +++ b/hookprobe/hookprobe/remediation.py @@ -74,6 +74,7 @@ from hookprobe import audit, automation from hookprobe.files import atomic_write +from hookprobe.redact import redact logger = logging.getLogger("hookprobe.remediation") @@ -276,6 +277,7 @@ def pending_duplicate(workdir: Path, steps: list[dict[str, Any]], now: float | N def save(workdir: Path, row: dict[str, Any]) -> None: + (workdir / DIRNAME).mkdir(parents=True, exist_ok=True) _write(workdir / DIRNAME / f"{row['id']}.json", row) @@ -284,10 +286,8 @@ def settle_interrupted(workdir: Path) -> list[dict[str, Any]]: A procedure is written as 1-2-3 and runs stop-on-first-failure, so the fact an operator needs before touching the target again is which commands landed. - The results list already holds one entry per command that ran, in order — - everything past it never started. `failed` is the terminal state for a - sequence that did not complete, and unlike `running` it is a state the row - can be read in. + A started step without a durable result has an unknown outcome. Older + rows without a start marker cannot prove whether any remaining step ran. """ settled: list[dict[str, Any]] = [] for row in list_all(workdir, limit=1000): @@ -297,7 +297,13 @@ def settle_interrupted(workdir: Path) -> list[dict[str, Any]]: ran = len(row.get("results") or []) row["status"] = FAILED row["executed_at"] = round(time.time(), 3) - row["interrupted"] = {"ran": commands[:ran], "not_run": commands[ran:]} + if "in_flight" not in row: + unknown, not_run = commands[ran:], [] + elif row["in_flight"] is None: + unknown, not_run = [], commands[ran:] + else: + unknown, not_run = commands[ran : ran + 1], commands[ran + 1 :] + row["interrupted"] = {"ran": commands[:ran], "unknown": unknown, "not_run": not_run} try: save(workdir, row) except OSError: @@ -784,6 +790,7 @@ def approve( "target needs now." ) row["status"] = "running" + row["in_flight"] = None row["approved_at"] = round(time.time(), 3) row["approved_note"] = note[:300] row["approved_by"] = actor[:120] @@ -886,7 +893,7 @@ async def execute( """ timeout = max(30.0, (bash_timeout_ms or 120000) / 1000.0) failed = False - for step in row.get("steps", []): + for index, step in enumerate(row.get("steps", [])): command = str(step.get("command") or "") started = time.monotonic() # Second gate. Deny-by-default holds: a file that has since been emptied @@ -902,13 +909,20 @@ async def execute( if argv is None: row.setdefault("results", []).append( { - "command": command, + "command": redact(command), "exit": -1, "ms": 0, - "output": f"refused at execution: {refusal}", + "output": redact(f"refused at execution: {refusal}"), } ) - logger.warning("remediation step refused at execution: %s — %s", command, refusal) + logger.warning("remediation step refused at execution: %s — %s", redact(command), redact(str(refusal))) + failed = True + break + row["in_flight"] = index + try: + save(workdir, row) + except OSError: + logger.exception("remediation %s cannot record step start; nothing further will run", row["id"]) failed = True break try: @@ -920,23 +934,35 @@ async def execute( env=execution_env(), ) try: - output, _ = await asyncio.wait_for(process.communicate(), timeout=timeout) + output, truncated = await asyncio.wait_for(_read_output(process), timeout=timeout) except TimeoutError: process.kill() await process.wait() - output, returncode = b"(timed out)", -1 + output, returncode, truncated = "(timed out)", -1, False + except asyncio.CancelledError: + if process.returncode is None: + process.kill() + await process.wait() + raise else: returncode = int(process.returncode or 0) except OSError as exc: - output, returncode = str(exc).encode(), -1 + output, returncode, truncated = redact(str(exc)), -1, False result = { - "command": command, + "command": redact(command), "exit": returncode, "ms": int((time.monotonic() - started) * 1000), - "output": output.decode("utf-8", "replace")[-10000:], + "output": output, + "output_truncated": truncated, } row.setdefault("results", []).append(result) _audit(workdir, row["id"], command, returncode != 0) + row["in_flight"] = None + try: + save(workdir, row) + except OSError: + logger.exception("remediation %s cannot record step result; nothing further will run", row["id"]) + return if returncode != 0: failed = True break @@ -1038,6 +1064,36 @@ def evidence( return stamped +OUTPUT_KEPT = 10_000 # characters of a step's output the record keeps, from the end +OUTPUT_HELD = 1_000_000 # bytes held so the whole output is redacted before it is cut + + +async def _read_output(process: asyncio.subprocess.Process) -> tuple[str, bool]: + """A step's output as the record keeps it, and whether any of it was dropped. + + The whole output is redacted BEFORE it is cut to its last 10,000 characters, + so a secret the cut would have split was seen whole by the redactor — and the + tail, where a failing command says why, is kept. That needs the whole output + in hand: past a megabyte nothing is held, and the output is omitted rather + than cut at a point nobody redacted. + """ + if process.stdout is None: + raise RuntimeError("remediation process has no output pipe") + held = bytearray() + total = 0 + while chunk := await process.stdout.read(65536): + total += len(chunk) + if total <= OUTPUT_HELD: + held.extend(chunk) + elif held: + held.clear() + await process.wait() + if total > OUTPUT_HELD: + return f"(output omitted: {total} bytes, more than can be redacted whole)", True + text = redact(bytes(held).decode("utf-8", "replace")) + return text[-OUTPUT_KEPT:], len(text) > OUTPUT_KEPT + + def _audit(workdir: Path, proposal_id: str, command: str, error: bool, *, tool: str = "Exec") -> None: """Same flight recorder the agent's tools write to — one account. @@ -1054,7 +1110,7 @@ def _audit(workdir: Path, proposal_id: str, command: str, error: bool, *, tool: "ts": round(time.time(), 3), "session": f"remediation:{proposal_id}", "tool": tool, - "detail": command[:300], + "detail": redact(command)[:300], "error": error, }, ) diff --git a/hookprobe/hookprobe/selftest.py b/hookprobe/hookprobe/selftest.py index b1a1f05..8555d9e 100644 --- a/hookprobe/hookprobe/selftest.py +++ b/hookprobe/hookprobe/selftest.py @@ -221,7 +221,9 @@ async def posture_still_holds(settings: Settings) -> dict[str, Any]: detail = f"{detail} · cannot measure: {missing} not in this image"[:280] return _check( "credentials are no wider than the declared posture", - verdict not in ("wider-than-declared", "unmeasurable"), + (False if settings.bash_guard != "readonly" and settings.blast_radius is not None else None) + if verdict == "unverifiable" + else verdict not in ("wider-than-declared", "unmeasurable"), detail, "a credential whose far side cannot be asked — `unverifiable` is a verdict, not a pass", ) diff --git a/hookprobe/hookprobe/service.py b/hookprobe/hookprobe/service.py index c2f0591..45c4886 100644 --- a/hookprobe/hookprobe/service.py +++ b/hookprobe/hookprobe/service.py @@ -927,9 +927,10 @@ def sweep_interrupted_remediations(self) -> int: for row in settled: interrupted = row.get("interrupted") or {} logger.warning( - "remediation interrupted by a restart id=%s ran=%s not_run=%s", + "remediation interrupted by a restart id=%s ran=%s unknown=%s not_run=%s", row.get("id"), len(interrupted.get("ran") or []), + len(interrupted.get("unknown") or []), len(interrupted.get("not_run") or []), ) if settled: diff --git a/hookprobe/hookprobe/ui.html b/hookprobe/hookprobe/ui.html index 2ffd835..1232322 100644 --- a/hookprobe/hookprobe/ui.html +++ b/hookprobe/hookprobe/ui.html @@ -1217,6 +1217,7 @@

Recipes

"sys.kube_v": "present · {m} mutating verbs allowed · {e} errors", "sys.aws_v": "present · {n} write actions allowed", "sys.not_present": "not present", + "sys.unverifiable": "unverifiable", "act.title": "Remediation proposals", "act.hint": "An investigation proposes a step only when it is confident, and nothing runs without an approval.", "act.none": "No remediation proposals.", @@ -1228,6 +1229,9 @@

Recipes

"act.status.rejected": "Rejected", "act.status.expired": "Expired", "act.undo": "undo: {cmd}", + "act.unknown": "Outcome unknown — check the target before retrying", + "act.not_run": "Not started", + "act.truncated": "Output truncated", "act.approved_by": "approved by {who}", "audit.title": "Audit", "audit.hint": "Every tool call in every run, subagents included · {dir}/audit/*.jsonl", @@ -1568,6 +1572,7 @@

Recipes

"sys.kube_v": "已挂载 · 允许 {m} 个变更动作 · {e} 个错误", "sys.aws_v": "已挂载 · 允许 {n} 个写操作", "sys.not_present": "未挂载", + "sys.unverifiable": "无法验证", "act.title": "处置建议", "act.hint": "调查只在有把握时才提出处置步骤，而且没有批准什么都不会执行。", "act.none": "没有处置建议。", @@ -1579,6 +1584,9 @@

Recipes

"act.status.rejected": "已拒绝", "act.status.expired": "已过期", "act.undo": "撤销：{cmd}", + "act.unknown": "结果未知——重试前请检查目标状态", + "act.not_run": "未开始", + "act.truncated": "输出已截断", "act.approved_by": "批准人 {who}", "audit.title": "审计", "audit.hint": "每次运行里的每一次工具调用，包括子智能体的 · {dir}/audit/*.jsonl", @@ -3133,7 +3141,7 @@

Recipes

(posture ? `

${icon("shield")}${esc(t("sys.posture"))}

` + kv([ [t("sys.verdict"), posture.verdict], [t("sys.guard"), posture.bash_guard], [t("sys.checked"), fmtTime(posture.checked_at)], ["kube", (posture.kube || {}).present ? t("sys.kube_v", { m: ((posture.kube || {}).mutating || []).length, e: ((posture.kube || {}).errors || []).length }) : t("sys.not_present")], - ["aws", (posture.aws || {}).present ? t("sys.aws_v", { n: ((posture.aws || {}).allowed || []).length }) : t("sys.not_present")], + ["aws", (posture.aws || {}).unverifiable ? t("sys.unverifiable") : (posture.aws || {}).present ? t("sys.aws_v", { n: ((posture.aws || {}).allowed || []).length }) : t("sys.not_present")], ]) + "
" : ""); } catch (e) { box.innerHTML = errorHtml(e); } } @@ -3151,9 +3159,11 @@

Recipes

box.innerHTML = head + rows.map((r) => { const steps = (r.steps || []).map((s, i) => { const res = (r.results || [])[i]; - const tag = res ? (res.exit === 0 ? `exit 0 · ${fmtMs(res.ms)}` : `exit ${esc(res.exit)}`) : ""; + const interrupted = r.interrupted; + const unknown = interrupted && i >= (interrupted.ran || []).length && (!Array.isArray(interrupted.unknown) || i < (interrupted.ran || []).length + interrupted.unknown.length); + const tag = res ? (res.exit === 0 ? `exit 0 · ${fmtMs(res.ms)}` : `exit ${esc(res.exit)}`) : interrupted ? `${esc(t(unknown ? "act.unknown" : "act.not_run"))}` : ""; return `
${esc(s.risk)}${esc(s.action)} ${tag}
$ ${esc(s.command)}
` + - (s.rollback ? `
${esc(t("act.undo", { cmd: s.rollback }))}
` : "") + (res && res.output ? `
${esc(res.output)}
` : "") + "
"; + (s.rollback ? `
${esc(t("act.undo", { cmd: s.rollback }))}
` : "") + (res && res.output ? `
${esc(res.output)}
` : "") + (res && res.output_truncated ? `
${esc(t("act.truncated"))}
` : "") + ""; }).join(""); // Past its approval window a proposal is inert: approving it answers 409, // so it offers nothing but says what it is. A status this page has no diff --git a/hookprobe/tests/conftest.py b/hookprobe/tests/conftest.py new file mode 100644 index 0000000..d7aef80 --- /dev/null +++ b/hookprobe/tests/conftest.py @@ -0,0 +1,22 @@ +"""Fixtures every investigator test shares.""" + +from __future__ import annotations + +import pytest + +from hookprobe import posture + +_REAL_METADATA_PROBE = posture.metadata_offers_credentials + + +@pytest.fixture(autouse=True) +def _no_instance_metadata(monkeypatch): + """No test asks the network for an instance metadata service. A CI runner on + a cloud VM has a metadata service at the same address, and a verdict must not depend on the + machine that ran the tests; a test that wants a service says so.""" + monkeypatch.setattr(posture, "metadata_offers_credentials", lambda timeout=1.0: False) + + +@pytest.fixture +def real_metadata_probe(): + return _REAL_METADATA_PROBE diff --git a/hookprobe/tests/test_posture.py b/hookprobe/tests/test_posture.py index 9ae3185..52519bb 100644 --- a/hookprobe/tests/test_posture.py +++ b/hookprobe/tests/test_posture.py @@ -10,6 +10,7 @@ import asyncio import json +import os import pytest @@ -79,6 +80,7 @@ def _with_kube(tmp_path, monkeypatch) -> None: cfg.write_text("apiVersion: v1\n") monkeypatch.setenv("KUBECONFIG", str(cfg)) monkeypatch.delenv("AWS_ACCESS_KEY_ID", raising=False) + monkeypatch.setenv("AWS_EC2_METADATA_DISABLED", "true") monkeypatch.setenv("AWS_SHARED_CREDENTIALS_FILE", str(tmp_path / "absent")) monkeypatch.setenv("AWS_CONFIG_FILE", str(tmp_path / "absent2")) @@ -123,6 +125,7 @@ def test_unverifiable_is_a_verdict_not_a_confirmation(tmp_path, monkeypatch) -> def test_no_credentials_and_off(tmp_path, monkeypatch) -> None: monkeypatch.setenv("KUBECONFIG", str(tmp_path / "nope")) monkeypatch.delenv("AWS_ACCESS_KEY_ID", raising=False) + monkeypatch.setenv("AWS_EC2_METADATA_DISABLED", "true") monkeypatch.setenv("AWS_SHARED_CREDENTIALS_FILE", str(tmp_path / "absent")) monkeypatch.setenv("AWS_CONFIG_FILE", str(tmp_path / "absent2")) run = _runner({}) @@ -131,6 +134,194 @@ def test_no_credentials_and_off(tmp_path, monkeypatch) -> None: assert asyncio.run(posture.on_startup(tmp_path, "readonly", "off", run))["verdict"] == "skipped" +@pytest.mark.parametrize( + "variables", + [ + {"AWS_ROLE_ARN": "arn:aws:iam::123456789012:role/example", "AWS_WEB_IDENTITY_TOKEN_FILE": "/tmp/token"}, + {"AWS_CONTAINER_CREDENTIALS_RELATIVE_URI": "/v2/credentials/example"}, + {"AWS_CONTAINER_CREDENTIALS_FULL_URI": "http://localhost/credentials"}, + ], +) +def test_aws_credential_sources_are_detected(tmp_path, monkeypatch, variables) -> None: + for name in list(os.environ): + if name.startswith("AWS_"): + monkeypatch.delenv(name) + monkeypatch.setenv("HOME", str(tmp_path)) + for name, value in variables.items(): + monkeypatch.setenv(name, value) + assert posture.aws_configured() + + +def _no_aws_named(tmp_path, monkeypatch) -> None: + for name in list(os.environ): + if name.startswith("AWS_"): + monkeypatch.delenv(name) + monkeypatch.setenv("HOME", str(tmp_path)) + + +def test_a_profile_name_with_no_file_behind_it_is_not_a_credential(tmp_path, monkeypatch) -> None: + _no_aws_named(tmp_path, monkeypatch) + monkeypatch.setenv("AWS_PROFILE", "example") + assert not posture.aws_configured() + + +def test_no_metadata_service_answering_is_no_credential(tmp_path, monkeypatch) -> None: + """The laptop and the internal network: nothing at the address, nothing to measure.""" + _no_aws_named(tmp_path, monkeypatch) + run = _runner({}) + aws = asyncio.run(posture.check_aws(run)) + assert aws == {"present": False} and run.calls == [] + assert posture.verdict("readonly", {"present": False}, aws) == "no-credentials" + + +def test_a_metadata_service_that_answers_is_measured_like_any_source(tmp_path, monkeypatch) -> None: + _no_aws_named(tmp_path, monkeypatch) + monkeypatch.setattr(posture, "metadata_offers_credentials", lambda timeout=1.0: True) + denied = [{"EvalActionName": name, "EvalDecision": "implicitDeny"} for name in posture.DANGEROUS_AWS_ACTIONS] + run = _runner( + { + "aws sts": (0, '{"Arn":"arn:aws:sts::1:assumed-role/instance/i-0"}', ""), + "aws iam": (0, json.dumps({"EvaluationResults": denied}), ""), + } + ) + aws = asyncio.run(posture.check_aws(run)) + assert aws["present"] is True and aws["unverifiable"] is False and aws["allowed"] == [] + assert posture.verdict("readonly", {"present": False}, aws) == "readonly-confirmed" + + +def test_an_instance_with_no_role_has_no_credential(tmp_path, monkeypatch) -> None: + _no_aws_named(tmp_path, monkeypatch) + monkeypatch.setattr(posture, "metadata_offers_credentials", lambda timeout=1.0: True) + run = _runner({"aws sts": (253, "", "Unable to locate credentials. You can configure credentials by running")}) + assert asyncio.run(posture.check_aws(run)) == {"present": False} + + +def test_metadata_disabled_is_not_asked(tmp_path, monkeypatch) -> None: + _no_aws_named(tmp_path, monkeypatch) + monkeypatch.setenv("AWS_EC2_METADATA_DISABLED", "true") + + def asked(timeout=1.0): + raise AssertionError("the disabled service was asked") + + monkeypatch.setattr(posture, "metadata_offers_credentials", asked) + assert asyncio.run(posture.check_aws(_runner({}))) == {"present": False} + + +def _metadata_service(put_status: int, listing_status: int, listing: bytes = b""): + """A stand-in for the address, answering the CLI's two questions as told.""" + import threading + from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + + class Service(BaseHTTPRequestHandler): + def do_PUT(self) -> None: # noqa: N802 + self.send_response(put_status) + self.end_headers() + if put_status == 200: + self.wfile.write(b"a-token") + + def do_GET(self) -> None: # noqa: N802 + asked_with_token = self.headers.get("X-aws-ec2-metadata-token") == "a-token" or put_status != 200 + ok = listing_status == 200 and asked_with_token + self.send_response(200 if ok else listing_status if listing_status != 200 else 401) + self.end_headers() + if ok: + self.wfile.write(listing) + + def log_message(self, *args) -> None: + pass + + server = ThreadingHTTPServer(("127.0.0.1", 0), Service) + threading.Thread(target=server.serve_forever, daemon=True).start() + return server + + +@pytest.mark.parametrize( + ("put_status", "listing_status", "listing", "offered"), + [ + (200, 200, b"instance-role", True), # IMDSv2 with a role + (403, 200, b"instance-role", True), # IMDSv1 only, a role all the same + (200, 404, b"", False), # an instance with no role + (400, 400, b"", False), # another vendor's service at the address + ], +) +def test_the_metadata_question_is_the_clis_and_goes_direct( + monkeypatch, real_metadata_probe, put_status, listing_status, listing, offered +) -> None: + """Only a role name is a credential, and the question skips the proxy: one + that refuses the address would otherwise read as a service being there.""" + server = _metadata_service(put_status, listing_status, listing) + monkeypatch.setenv("HTTP_PROXY", "http://127.0.0.1:9") + monkeypatch.setenv("http_proxy", "http://127.0.0.1:9") + monkeypatch.setenv("AWS_EC2_METADATA_SERVICE_ENDPOINT", f"http://127.0.0.1:{server.server_port}") + try: + assert real_metadata_probe(timeout=2.0) is offered + finally: + server.shutdown() + server.server_close() + + +def test_nothing_at_the_address_is_no_credential(monkeypatch, real_metadata_probe) -> None: + import socket + + with socket.socket() as closed: + closed.bind(("127.0.0.1", 0)) + port = closed.getsockname()[1] + monkeypatch.setenv("AWS_EC2_METADATA_SERVICE_ENDPOINT", f"http://127.0.0.1:{port}") + assert real_metadata_probe(timeout=2.0) is False + + +def test_writing_node_refuses_unverifiable_declared_radius(tmp_path, monkeypatch) -> None: + _with_kube(tmp_path, monkeypatch) + radius = tmp_path / "radius.txt" + radius.write_text("", encoding="utf-8") + _with_aws_too(tmp_path, monkeypatch) + run = _runner( + { + "kubectl auth can-i --list": (0, CAN_I_LIST_READONLY, ""), + "aws sts": (0, '{"Arn":"arn:aws:iam::1:user/example"}', ""), + "aws iam": (1, "", "AccessDenied"), + } + ) + with pytest.raises(posture.PostureViolation, match="could not be verified"): + asyncio.run(posture.on_startup(tmp_path, "danger-only", "enforce", run, radius=radius)) + assert posture.read(tmp_path)["verdict"] == "unverifiable" + + +@pytest.mark.parametrize( + "response", + [ + "not-json", + "{}", + '{"EvaluationResults": []}', + '{"EvaluationResults": [{"EvalActionName": "x", "EvalDecision": "bad"}]}', + ], +) +def test_incomplete_or_invalid_aws_simulation_is_unverifiable(tmp_path, monkeypatch, response) -> None: + _with_aws_too(tmp_path, monkeypatch) + run = _runner( + { + "aws sts": (0, '{"Arn":"arn:aws:iam::1:user/example"}', ""), + "aws iam": (0, response, ""), + } + ) + record = asyncio.run(posture.check_aws(run)) + assert record["unverifiable"] and record["errors"] + assert posture.verdict("readonly", {"present": False}, record) == "unverifiable" + + +@pytest.mark.parametrize("malformed", [False, True]) +def test_partial_simulation_still_reports_known_allowed_actions(tmp_path, monkeypatch, malformed) -> None: + _with_aws_too(tmp_path, monkeypatch) + results = [{"EvalActionName": "s3:PutObject", "EvalDecision": "allowed"}] + if malformed: + results.append(None) + response = json.dumps({"EvaluationResults": results}) + run = _runner({"aws sts": (0, '{"Arn":"arn:aws:iam::1:user/example"}', ""), "aws iam": (0, response, "")}) + record = asyncio.run(posture.check_aws(run)) + assert record["unverifiable"] and record["allowed"] == ["s3:PutObject"] + assert posture.verdict("readonly", {"present": False}, record) == "wider-than-declared" + + def test_a_typo_in_the_mode_fails_closed(tmp_path, monkeypatch) -> None: _with_kube(tmp_path, monkeypatch) run = _runner({"kubectl auth can-i --list": (0, CAN_I_LIST_ADMIN, "")}) @@ -264,6 +455,16 @@ def test_a_writing_node_without_kubectl_starts_and_pages(tmp_path, monkeypatch) assert pages and "kubectl" in pages[0] +def test_writing_node_with_declared_radius_refuses_missing_tool(tmp_path, monkeypatch) -> None: + _with_kube(tmp_path, monkeypatch) + radius = tmp_path / "radius.txt" + radius.write_text("", encoding="utf-8") + run = _runner({"kubectl": (127, "", "kubectl: not found")}) + with pytest.raises(posture.PostureViolation, match="kubectl"): + asyncio.run(posture.on_startup(tmp_path, "danger-only", "enforce", run, radius=radius)) + assert posture.read(tmp_path)["verdict"] == "unmeasurable" + + def test_warn_starts_a_readonly_node_without_its_tool_and_pages(tmp_path, monkeypatch) -> None: _with_kube(tmp_path, monkeypatch) run = _runner({"kubectl": (127, "", "kubectl: not found")}) diff --git a/hookprobe/tests/test_remediation.py b/hookprobe/tests/test_remediation.py index 99ca1e8..b16a52f 100644 --- a/hookprobe/tests/test_remediation.py +++ b/hookprobe/tests/test_remediation.py @@ -280,6 +280,7 @@ def test_a_restart_settles_a_procedure_it_died_in_the_middle_of(tmp_path): "session_key": "probe:inbound:7", "created_at": 1.0, "status": "running", + "in_flight": None, "steps": [ {"action": "drain", "command": "kubectl drain node-3", "risk": "high", "rollback": "uncordon"}, {"action": "restart", "command": "kubectl rollout restart deploy/api", "risk": "medium"}, @@ -311,6 +312,76 @@ def test_a_restart_settles_a_procedure_it_died_in_the_middle_of(tmp_path): ] +def test_interruption_records_completed_and_unknown_steps(tmp_path, monkeypatch): + async def scenario(): + allow = tmp_path / "allow" + allow.write_text(".*\n", encoding="utf-8") + steps = [ + {"action": "first", "command": "echo completed", "risk": "low"}, + {"action": "second", "command": "sleep 2", "risk": "low"}, + {"action": "third", "command": "echo later", "risk": "low"}, + ] + proposal_id = remediation.propose(tmp_path, "run", steps) + row = remediation.load(tmp_path, proposal_id) + row.update(status="running", in_flight=None) + remediation.save(tmp_path, row) + first_done = asyncio.Event() + original = remediation._audit + + def record(*args, **kwargs): + original(*args, **kwargs) + first_done.set() + + monkeypatch.setattr(remediation, "_audit", record) + task = asyncio.create_task(remediation.execute(tmp_path, row, bash_timeout_ms=30000, allowlist=allow)) + await asyncio.wait_for(first_done.wait(), 5) + for _ in range(100): + if remediation.load(tmp_path, proposal_id).get("in_flight") == 1: + break + await asyncio.sleep(0.01) + else: + pytest.fail("second step never started") + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + return remediation.settle_interrupted(tmp_path)[0]["interrupted"] + + interrupted = asyncio.run(scenario()) + assert interrupted == {"ran": ["echo completed"], "unknown": ["sleep 2"], "not_run": ["echo later"]} + + +def test_remediation_output_is_redacted_and_bounded(tmp_path): + import sys + + async def scenario(command): + allow = tmp_path / "allow" + allow.write_text(".*\n", encoding="utf-8") + row = {"id": "0123456789", "status": "running", "steps": [{"command": command, "risk": "low"}]} + await remediation.execute(tmp_path, row, bash_timeout_ms=30000, allowlist=allow) + return row["results"][0] + + secret = asyncio.run(scenario("echo password=not-a-real-secret")) + assert "not-a-real-secret" not in secret["output"] + assert "[redacted]" in secret["output"] + # A secret straddling the 10,000-character cut: redacted whole before the cut, + # so no piece of it survives, and the tail a failing command writes is kept. + script = tmp_path / "straddle.py" + script.write_text( + 'print("A" * 20000 + " sk-ant-" + "Q" * 24 + " " + "B" * 9985)\n', + encoding="utf-8", + ) + large = asyncio.run(scenario(f"{sys.executable} {script}")) + assert large["output_truncated"] is True + assert len(large["output"]) <= 10000 + assert "QQQQ" not in large["output"] + assert large["output"].rstrip().endswith("B" * 9985) + flood = tmp_path / "flood.py" + flood.write_text('import sys\nsys.stdout.write("x" * 1_100_000)\n', encoding="utf-8") + omitted = asyncio.run(scenario(f"{sys.executable} {flood}")) + assert omitted["output_truncated"] is True + assert omitted["output"].startswith("(output omitted: 1100000 bytes") + + def test_a_row_that_cannot_be_written_after_execution_is_loud(tmp_path, monkeypatch, caplog): """The commands have already run by then, so a lost write is not a detail: without a line in the log, the only account of it would be the audit file diff --git a/hookprobe/tests/test_selftest.py b/hookprobe/tests/test_selftest.py index c0982e6..231d46d 100644 --- a/hookprobe/tests/test_selftest.py +++ b/hookprobe/tests/test_selftest.py @@ -50,6 +50,17 @@ def test_a_check_that_could_not_run_is_never_a_pass() -> None: assert len(worse["checks"]) == 3 +def test_unverifiable_posture_is_unproven(tmp_path, monkeypatch) -> None: + async def unverifiable(*args, **kwargs): + return {"verdict": "unverifiable", "kube": {}, "aws": {}} + + monkeypatch.setattr(selftest.posture, "check", unverifiable) + result = asyncio.run(selftest.posture_still_holds(make_settings(tmp_path, workdir=tmp_path))) + assert result["held"] is None + settings = make_settings(tmp_path, workdir=tmp_path, bash_guard="danger-only", blast_radius=tmp_path / "radius") + assert asyncio.run(selftest.posture_still_holds(settings))["held"] is False + + def test_the_gate_check_actually_asks_the_gate(tmp_path) -> None: """In-process and through the subprocess path. The second is the one that was once simply absent — the hook command could not import hookprobe, the diff --git a/hookrelay/hookrelay/app.py b/hookrelay/hookrelay/app.py index 41a02b0..e2a7e84 100644 --- a/hookrelay/hookrelay/app.py +++ b/hookrelay/hookrelay/app.py @@ -19,6 +19,7 @@ import os import re import tempfile +import uuid from collections.abc import AsyncIterator from pathlib import Path from typing import Any @@ -451,70 +452,107 @@ async def hook(source_name: str, request: Request) -> JSONResponse: raise HTTPException(status_code=401, detail="bad signature") # A captured request replayed inside the skew is the same signed BYTES, # and the content fingerprint cannot see that: it folds a restatement of - # the same alert however long ago the content window closed. So ask here - # and answer the second copy as a skip — but PEEK, do not claim: the row - # is written only after the pipeline has answered (below), because a - # claim made before it would turn a sender's identical retry after a 5xx, - # a timeout or the fuse's 429 into a replay, losing the event while both - # sides believed it was handled. Off when the window is 0, and a scheme + # the same alert however long ago the content window closed. So the + # signature is held while this request runs and marked spent only once + # the pipeline has answered (below): spent before it, a sender's identical + # retry after a 5xx, a timeout or the fuse's 429 would read as a replay, + # losing the event while both sides believed it was handled. The hold is + # a renewable lease, so an identical copy arriving meanwhile is told to + # retry (503) instead of entering the pipeline beside the first, and a + # crashed holder's lease runs out. Off when the window is 0, and a scheme # this cannot name (a plugin adapter's own header) yields no token, which # is a refusal to guess rather than a miss. replay: str | None = None + owner = uuid.uuid4().hex + lease_seconds = 120 + heartbeat: asyncio.Task[None] | None = None if app_settings.signature_replay_seconds > 0: replay = replay_token(source.name, headers.get("x-hook-signature")) - if replay is not None and await app.state.store.signature_seen( - replay, app_settings.signature_replay_seconds, now_ts() - ): - return JSONResponse({"outcome": "skipped", "skip_code": "replayed_signature"}) + if replay is not None: + state = await app.state.store.acquire_signature( + replay, owner, app_settings.signature_replay_seconds, now_ts(), lease_seconds + ) + if state == "completed": + return JSONResponse({"outcome": "skipped", "skip_code": "replayed_signature"}) + if state == "busy": + return JSONResponse({"detail": "identical request is still processing"}, 503, {"Retry-After": "1"}) + + processing = asyncio.current_task() + + async def renew() -> None: + while True: + await asyncio.sleep(lease_seconds / 3) + try: + renewed = await app.state.store.renew_signature(replay, owner, now_ts(), lease_seconds) + except Exception: + logger.exception("signature processing lease renewal failed") + renewed = False + if not renewed: + logger.error("signature processing lease was lost") + if processing is not None: + processing.cancel() + return + + heartbeat = asyncio.create_task(renew()) try: - payload = json.loads(body or b"{}") - except ValueError: - raise HTTPException(status_code=400, detail="body is not JSON") from None - # ── storm fuse (volume, not content) ────────────────────────────── - # After verification (an unsigned flood is already 401 and free) and - # before the pipeline. Two stages: suppress keeps the account, reject - # protects the account itself. - verdict = app.state.fuse.check(source.name, source.storm_threshold, source.storm_window_seconds, now_ts()) - if verdict == "reject": - raise HTTPException(status_code=429, detail="storm fuse open (hard)") - # The adapter reads the payload for BOTH paths from here on. A suppressed - # event is still recorded, so it must be recorded the way this door - # actually reads its senders: the storm path used to re-extract with the - # door's templates alone, and a reshaping adapter (SNS hides the alert - # inside a JSON-string `Message`) mis-titled every row the storm made. - # It sits after the hard reject, which stores nothing and so reads nothing. - extracted = adapter.parse(source, payload) - if verdict == "suppress": - event_id = await record_storm_suppressed( - app.state.store, - source, - payload, - now_ts(), - app.state.fuse.window_count(source.name), - source.storm_threshold, - extracted=extracted, - ) + try: + payload = json.loads(body or b"{}") + except ValueError: + raise HTTPException(status_code=400, detail="body is not JSON") from None + # ── storm fuse (volume, not content) ────────────────────────── + # After verification (an unsigned flood is already 401 and free) and + # before the pipeline. Two stages: suppress keeps the account, reject + # protects the account itself. + verdict = app.state.fuse.check(source.name, source.storm_threshold, source.storm_window_seconds, now_ts()) + if verdict == "reject": + raise HTTPException(status_code=429, detail="storm fuse open (hard)") + # The adapter reads the payload for BOTH paths from here on. A suppressed + # event is still recorded, so it must be recorded the way this door + # actually reads its senders: the storm path used to re-extract with the + # door's templates alone, and a reshaping adapter (SNS hides the alert + # inside a JSON-string `Message`) mis-titled every row the storm made. + # It sits after the hard reject, which stores nothing and so reads nothing. + extracted = adapter.parse(source, payload) + if verdict == "suppress": + event_id = await record_storm_suppressed( + app.state.store, + source, + payload, + now_ts(), + app.state.fuse.window_count(source.name), + source.storm_threshold, + extracted=extracted, + ) + result = {"event_id": event_id, "outcome": "skipped", "skip_code": "storm_suppressed"} + else: + result = await handle_hook( + app.state.store, + app.state.config, + source, + payload, + now_ts(), + settings=app_settings, + client=app.state.http_client, + extracted=extracted, + ) + # The pipeline answered, so the event was handled once: the signature is + # spent now. Anything that raised above (the 400 for a body that is not + # JSON, the 429 from the hard fuse, a fault inside handle_hook) reaches + # `finally` having spent nothing and is released there, so a retry is processed. + if replay is not None and not await app.state.store.complete_signature( + replay, owner, app_settings.signature_replay_seconds, now_ts() + ): + logger.error("signature processing lease was lost before completion") + raise HTTPException(status_code=503, detail="signature lease lost; outcome needs reconciliation") + return JSONResponse(result) + finally: + if heartbeat is not None: + heartbeat.cancel() + with contextlib.suppress(asyncio.CancelledError): + await heartbeat if replay is not None: - await app.state.store.claim_signature(replay, app_settings.signature_replay_seconds, now_ts()) - return JSONResponse({"event_id": event_id, "outcome": "skipped", "skip_code": "storm_suppressed"}) - - result = await handle_hook( - app.state.store, - app.state.config, - source, - payload, - now_ts(), - settings=app_settings, - client=app.state.http_client, - extracted=extracted, - ) - # The pipeline answered, so the event was handled once: remember the - # signature now. Anything that raised above — the 400 for a body that is - # not JSON, the 429 from the hard fuse, a fault inside handle_hook — - # reaches here having claimed nothing, so the sender's retry is processed. - if replay is not None: - await app.state.store.claim_signature(replay, app_settings.signature_replay_seconds, now_ts()) - return JSONResponse(result) + with contextlib.suppress(Exception): + await app.state.store.release_signature(replay, owner) @app.post("/explain/{source_name}") async def explain( diff --git a/hookrelay/hookrelay/security.py b/hookrelay/hookrelay/security.py index ba7cb2f..612e8b5 100644 --- a/hookrelay/hookrelay/security.py +++ b/hookrelay/hookrelay/security.py @@ -4,7 +4,7 @@ ("sha256=" prefix tolerated). With X-Hook-Timestamp the signature covers "{timestamp}.{body}" and must be FRESH, which BOUNDS how long a captured request stays useful rather than preventing replay outright; the door pairs it -with `replay_token` + `Store.claim_signature` to close that window (see +with `replay_token` + the store's signature lease to close that window (see verify_signature). Without a timestamp the legacy body-only form is accepted unless the door sets require_timestamp. A source configured without a secret is accepted unsigned — a deliberate, documented decision for trusted @@ -15,6 +15,7 @@ import hashlib import hmac +import math import time @@ -63,7 +64,7 @@ def verify_signature( CLOSED BY THE DOOR, not here (2026-10-12). A pure function has no store to write to, and that is deliberate: `verify_signature` is the one thing both the pipe's doors and any second implementation must agree on, so it stays - pure. The door pairs it with `replay_token` and `Store.claim_signature`, + pure. The door pairs it with `replay_token` and the store's signature lease, which together answer the second copy as a duplicate before the pipeline and keep it from spending a second verdict. The old reasoning — "the window is narrower than the dedup window, so a replay lands as a duplicate anyway" — @@ -89,6 +90,8 @@ def verify_signature( sent_at = float(str(timestamp_value).strip()) except ValueError: return False + if not math.isfinite(sent_at): + return False current = time.time() if now is None else now if abs(current - sent_at) > max(1, max_skew_seconds): return False diff --git a/hookrelay/hookrelay/store.py b/hookrelay/hookrelay/store.py index b975dda..8e5dcbc 100644 --- a/hookrelay/hookrelay/store.py +++ b/hookrelay/hookrelay/store.py @@ -146,6 +146,11 @@ seen_at REAL NOT NULL ); CREATE INDEX IF NOT EXISTS ix_spent_signatures_seen ON spent_signatures (seen_at); +CREATE TABLE IF NOT EXISTS pending_signatures ( + token TEXT NOT NULL PRIMARY KEY, + owner TEXT NOT NULL, + expires_at REAL NOT NULL +); """ @@ -973,42 +978,52 @@ async def record_action_outcome(self, jti: str, outcome: str) -> None: # ── inbound signatures ──────────────────────────────────────────────── - async def signature_seen(self, token: str, window_seconds: int, now: float) -> bool: - """Whether this signature was CLAIMED inside the window. A read, no claim. - - Paired with `claim_signature` so the door can ask before the pipeline and - record only once the pipeline has answered. Claiming up front looked - tidier and lost events: a row written before a 5xx, a client timeout or - the fuse's 429 makes the sender's identical RETRY answer as a replay, and - both sides then believe the event was handled. The cost of the split is - the race between two concurrent identical requests, which degrades to the - content dedup — the behaviour before this cache existed, and acceptable. - - The prune lives in `claim_signature`, which runs on every HANDLED - request, so this stays a read and the table stays bounded by the window. + async def acquire_signature(self, token: str, owner: str, window: int, now: float, lease: int) -> str: + """`acquired`, `busy` (another request holds the lease) or `completed` + (spent inside the window). One write, so two identical requests cannot + both acquire; a crashed owner's lease simply expires. Spent rows are + pruned in `complete_signature`, so the table is the size of the window. """ - cursor = await self.read.execute( - "SELECT 1 FROM spent_signatures WHERE token = ? AND seen_at >= ? LIMIT 1", - (token, now - window_seconds), - ) - return await cursor.fetchone() is not None - - async def claim_signature(self, token: str, window_seconds: int, now: float) -> bool: - """True the first time this signature is seen inside the window, else False. + async with self._write(): + await self.db.execute("DELETE FROM pending_signatures WHERE expires_at <= ?", (now,)) + cursor = await self.db.execute( + "INSERT OR IGNORE INTO pending_signatures (token, owner, expires_at) " + "SELECT ?, ?, ? WHERE NOT EXISTS " + "(SELECT 1 FROM spent_signatures WHERE token = ? AND seen_at >= ?)", + (token, owner, now + lease, token, now - window), + ) + if cursor.rowcount: + return "acquired" + cursor = await self.db.execute( + "SELECT 1 FROM spent_signatures WHERE token = ? AND seen_at >= ?", (token, now - window) + ) + return "completed" if await cursor.fetchone() else "busy" - Same shape as spend_action: the PRIMARY KEY is the decision rather than a - read-then-write with a gap. The prune runs inside the same write lock, so - the table is the size of the window and not of uptime. - `window_seconds <= 0` is the caller's business, not this method's — a zero - window is a refusal to cache, not a claim with nothing to compare against. - """ + async def renew_signature(self, token: str, owner: str, now: float, lease: int) -> bool: async with self._write(): - await self.db.execute("DELETE FROM spent_signatures WHERE seen_at < ?", (now - window_seconds,)) cursor = await self.db.execute( - "INSERT OR IGNORE INTO spent_signatures (token, seen_at) VALUES (?, ?)", (token, now) + "UPDATE pending_signatures SET expires_at = ? WHERE token = ? AND owner = ? AND expires_at > ?", + (now + lease, token, owner, now), ) return cursor.rowcount == 1 + async def release_signature(self, token: str, owner: str) -> None: + async with self._write(): + await self.db.execute("DELETE FROM pending_signatures WHERE token = ? AND owner = ?", (token, owner)) + + async def complete_signature(self, token: str, owner: str, window: int, now: float) -> bool: + async with self._write(): + cursor = await self.db.execute( + "SELECT 1 FROM pending_signatures WHERE token = ? AND owner = ? AND expires_at > ?", + (token, owner, now), + ) + if not await cursor.fetchone(): + return False + await self.db.execute("DELETE FROM spent_signatures WHERE seen_at < ?", (now - window,)) + await self.db.execute("INSERT OR REPLACE INTO spent_signatures(token,seen_at) VALUES (?,?)", (token, now)) + await self.db.execute("DELETE FROM pending_signatures WHERE token = ? AND owner = ?", (token, owner)) + return True + async def recent_actions(self, limit: int = 50) -> list[dict[str, Any]]: cursor = await self.read.execute( "SELECT kind, event_id, correlation_id, actor, outcome, pressed_at " diff --git a/hookrelay/tests/test_signature_replay.py b/hookrelay/tests/test_signature_replay.py index 913674b..c6ee992 100644 --- a/hookrelay/tests/test_signature_replay.py +++ b/hookrelay/tests/test_signature_replay.py @@ -2,8 +2,8 @@ The timestamp on a signed door bounds how long a capture stays useful; on its own it does not close the window, because the same signature inside it is -accepted again. `replay_token` + the store's `signature_seen`/`claim_signature` -pair close it at the door. These tests pin three things: the door's observable +accepted again. `replay_token` + the store's signature lease +(`acquire_signature` / `complete_signature`) close it at the door. These tests pin three things: the door's observable answer, the `None` cases where there is nothing a guard can track without guessing, and — the one a first cut got wrong — that the signature is claimed only once the pipeline has ANSWERED, so a retry after a 5xx or a 429 is @@ -12,6 +12,7 @@ from __future__ import annotations +import asyncio import contextlib import dataclasses import hashlib @@ -19,10 +20,12 @@ import json import httpx +import pytest import hookrelay.app as app_module from hookrelay.app import create_app -from hookrelay.security import replay_token +from hookrelay.security import replay_token, verify_signature +from hookrelay.store import Store PAYLOAD = {"title": "db down", "message": "x", "state": "alerting"} @@ -163,6 +166,105 @@ async def test_a_second_request_after_success_is_still_folded(client): assert again.json() == {"outcome": "skipped", "skip_code": "replayed_signature"} +async def test_simultaneous_replay_waits_for_the_first_request(settings, cfg, monkeypatch): + entered = asyncio.Event() + release = asyncio.Event() + real = app_module.handle_hook + calls = 0 + + async def delayed(*args, **kwargs): + nonlocal calls + calls += 1 + entered.set() + await asyncio.wait_for(release.wait(), 2) + return await real(*args, **kwargs) + + monkeypatch.setattr(app_module, "handle_hook", delayed) + body = json.dumps(PAYLOAD).encode() + headers = {"X-Hook-Signature": _sig("s3cret", body)} + app = create_app(settings=settings, cfg=cfg) + async with _serve(app) as client: + first = asyncio.create_task(client.post("/hook/grafana", content=body, headers=headers)) + await asyncio.wait_for(entered.wait(), 2) + concurrent = await client.post("/hook/grafana", content=body, headers=headers) + assert concurrent.status_code == 503 and concurrent.headers["retry-after"] == "1" + release.set() + assert (await first).json()["outcome"] == "routed" + retry = await client.post("/hook/grafana", content=body, headers=headers) + assert retry.json()["skip_code"] == "replayed_signature" + assert calls == 1 + + +async def test_expired_processing_lease_can_be_recovered(store): + assert await store.acquire_signature("token", "first", 300, 1000, 120) == "acquired" + assert await store.acquire_signature("token", "second", 300, 1001, 120) == "busy" + assert await store.acquire_signature("token", "second", 300, 1121, 120) == "acquired" + assert not await store.complete_signature("token", "first", 300, 1122) + assert await store.complete_signature("token", "second", 300, 1122) + assert await store.acquire_signature("token", "third", 300, 1123, 120) == "completed" + + +async def test_separate_stores_serialize_signature_acquisition(tmp_path): + first = Store(str(tmp_path / "shared.db")) + second = Store(str(tmp_path / "shared.db")) + await first.open() + await second.open() + try: + outcomes = await asyncio.gather( + first.acquire_signature("token", "first", 300, 1000, 120), + second.acquire_signature("token", "second", 300, 1000, 120), + ) + assert sorted(outcomes) == ["acquired", "busy"] + finally: + await first.close() + await second.close() + + +async def test_failed_lease_renewal_cancels_processing_and_releases_lease(settings, cfg, monkeypatch): + entered = asyncio.Event() + real_handle = app_module.handle_hook + real_sleep = asyncio.sleep + calls = 0 + + async def fast_heartbeat(delay): + if delay == 40: + await entered.wait() + else: + await real_sleep(delay) + + async def handle(*args, **kwargs): + nonlocal calls + calls += 1 + if calls == 1: + entered.set() + await asyncio.Event().wait() + return await real_handle(*args, **kwargs) + + async def broken_renew(*args, **kwargs): + raise OSError("database unavailable") + + monkeypatch.setattr(app_module.asyncio, "sleep", fast_heartbeat) + monkeypatch.setattr(app_module, "handle_hook", handle) + body = json.dumps(PAYLOAD).encode() + headers = {"X-Hook-Signature": _sig("s3cret", body)} + app = create_app(settings=settings, cfg=cfg) + async with _serve(app) as client: + monkeypatch.setattr(app.state.store, "renew_signature", broken_renew) + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(client.post("/hook/grafana", content=body, headers=headers), 2) + monkeypatch.setattr(app_module.asyncio, "sleep", real_sleep) + response = await client.post("/hook/grafana", content=body, headers=headers) + assert response.status_code == 200 and response.json()["outcome"] == "routed" + assert calls == 2 + + +def test_nonfinite_signed_timestamps_are_refused(): + body = b"{}" + for timestamp in ("NaN", "inf", "-inf"): + signature = hmac.new(b"s3cret", timestamp.encode() + b"." + body, hashlib.sha256).hexdigest() + assert not verify_signature("s3cret", body, signature, timestamp_value=timestamp, require_timestamp=True) + + # ── the token ───────────────────────────────────────────────────────────── @@ -182,23 +284,25 @@ def test_replay_token_is_none_when_there_is_nothing_to_track(): assert replay_token("", "sig") is None -async def test_claim_signature_is_true_once_then_false(store): +async def test_a_completed_signature_is_spent_once(store): now = 1_000.0 - assert await store.claim_signature("grafana:abc", 300, now) is True - assert await store.claim_signature("grafana:abc", 300, now + 1) is False + assert await store.acquire_signature("grafana:abc", "a", 300, now, 120) == "acquired" + assert await store.complete_signature("grafana:abc", "a", 300, now) is True + assert await store.acquire_signature("grafana:abc", "b", 300, now + 1, 120) == "completed" -async def test_signature_seen_asks_without_claiming(store): - """The peek must not have the side effect of the claim, or the door's - peek-then-claim split would be a claim wearing two names.""" - assert await store.signature_seen("grafana:abc", 300, 1_000.0) is False - assert await store.signature_seen("grafana:abc", 300, 1_000.0) is False - assert await store.claim_signature("grafana:abc", 300, 1_000.0) is True - assert await store.signature_seen("grafana:abc", 300, 1_000.0) is True +async def test_a_held_signature_is_busy_until_its_holder_lets_go(store): + """Released after a failure, the next identical request is processed — + the property the spent-after-answer order exists for.""" + assert await store.acquire_signature("grafana:abc", "a", 300, 1_000.0, 120) == "acquired" + assert await store.acquire_signature("grafana:abc", "b", 300, 1_000.0, 120) == "busy" + await store.release_signature("grafana:abc", "a") + assert await store.acquire_signature("grafana:abc", "b", 300, 1_000.0, 120) == "acquired" -async def test_claim_signature_forgets_outside_the_window(store): +async def test_a_spent_signature_is_forgotten_outside_the_window(store): """The prune is what keeps the table the size of the window, not of uptime — and what lets a genuinely new request with recycled bytes through later.""" - assert await store.claim_signature("grafana:abc", 300, 1_000.0) is True - assert await store.claim_signature("grafana:abc", 300, 1_301.0) is True + assert await store.acquire_signature("grafana:abc", "a", 300, 1_000.0, 120) == "acquired" + assert await store.complete_signature("grafana:abc", "a", 300, 1_000.0) is True + assert await store.acquire_signature("grafana:abc", "b", 300, 1_301.0, 120) == "acquired" diff --git a/scripts/assert_guards_are_tested.py b/scripts/assert_guards_are_tested.py index 47357ef..35ef066 100755 --- a/scripts/assert_guards_are_tested.py +++ b/scripts/assert_guards_are_tested.py @@ -163,8 +163,8 @@ ( "a credential nobody could measure passes as recorded", "hookprobe/hookprobe/posture.py", - ' return "unmeasurable" if missing else "recorded"', - ' return "recorded"', + ' if missing:\n return "unmeasurable"\n if kube.get("errors")', + ' if missing:\n return "recorded"\n if kube.get("errors")', ("tests/test_posture.py",), ), (