diff --git a/README.md b/README.md index 79933df..d438149 100644 --- a/README.md +++ b/README.md @@ -65,9 +65,9 @@ Sometimes a customer, auditor, or partner needs evidence that a human approved a This is an **experimental cryptographic preview**, not a production compliance claim. It uses an embedded BN254/Groth16 circuit and a development single-party setup; the circuit has not received an independent audit. Use it to evaluate the disclosure model, then replace the setup through a ceremony before relying on it in production. See [zero-knowledge approval proofs](docs/zk-approval-proofs.md) for the trust model, exact statement, and limitations. -> **New in v0.6.0:** the gate holds under load. The [tool-call gate](docs/tool-gate.md) answers asynchronously (`mode: async`, `wait:`) so a harness with a short HTTP timeout never loses a decision, and a tool call waiting on a human is durable across a restart. Rules constrain arguments (`args:` — glob, regex, `one_of`, `min`/`max`) and never widen on a mismatch. A repeat guard stops an agent that loops on one call from paging you, the operator hears about denials the gate made on its own, `/pending` and `draftcat pending` list every open gate, `/status` shows spend against caps, cost caps enforce the provider's real charge, rate limits back off instead of failing the run — and one Telegram update pump fixes taps that were silently lost while two gates were open at once. +> **New in v0.7.0:** execution decisions now carry their proof. Every tool-gate route is authenticated, each request has a stable action identity and exact policy binding, and an allowed decision becomes an atomic consume-once permit before the side effect runs. Webhook acceptance is durable before HTTP 202 and can be polled after handoff. Versioned receipts bind action, payload, policy, and expiry, with `draftcat receipts list|show|export` for verification-ready JSONL. Ordered `model_policy` rules can deny or send matching model input/output to a human, while `/healthz` and `/readyz` give orchestrators a safe listener contract. > -> **In v0.5.0:** approvals reach any operator surface via the [`hitl/v0` protocol](docs/hitl-protocol.md) — Microsoft Teams through a Power Automate flow in your own tenant, with no bot and no Azure app registration. Plus a tool-call gate for an agent's MCP/SDK calls (`POST /gate/tool-call`), risk tiers with pre-declared `approval_policy` exemptions, run-correlated audit rows, spend shown at the moment of decision, and `escalate_after` reminders before a gate times out. +> **In v0.6.0:** the gate holds under load. The [tool-call gate](docs/tool-gate.md) answers asynchronously (`mode: async`, `wait:`) so a harness with a short HTTP timeout never loses a decision, and a tool call waiting on a human is durable across a restart. Rules constrain arguments (`args:` - glob, regex, `one_of`, `min`/`max`) and never widen on a mismatch. A repeat guard stops an agent that loops on one call from paging you, the operator hears about denials the gate made on its own, `/pending` and `draftcat pending` list every open gate, `/status` shows spend against caps, cost caps enforce the provider's real charge, rate limits back off instead of failing the run - and one Telegram update pump fixes taps that were silently lost while two gates were open at once. ![Demo](demo.gif) @@ -106,7 +106,7 @@ However your agent runs, draftcat sits between it and your customer systems as a - **Cost budgets** — `per_day_cost` / `per_pipeline_cost` cap spend in money. On OpenRouter the caps are enforced on the charge the provider reports for each call (cached and reasoning tokens included); elsewhere on your configured per-1k rates. The approval prompt shows what the run has spent, and `/status` shows the day against every cap. - **Human-in-the-loop** — every outbound action requires an explicit operator decision, made live or declared in advance. - **Any operator channel** — the [`hitl/v0` protocol](docs/hitl-protocol.md) keeps draftcat as the gate and lets an untrusted relay own presentation. Teams runs through a Power Automate flow in your own tenant: no bot, no Azure app registration, no admin consent. Check yours with `draftcat hitl verify `. -- **Tool-call gate** — `POST /gate/tool-call` puts an agent's MCP or SDK calls through the same gate as a pipeline step. Denies by default; the approval binds to a hash of the exact arguments. Rules can constrain the arguments themselves (`args:`) and a mismatch only ever tightens — ask a human, or refuse. A decision that needs a human can be collected asynchronously (`mode: async`, `wait:`, `GET /gate/tool-call/`), and the open gate is durable across a restart. See [`docs/tool-gate.md`](docs/tool-gate.md). +- **Consume-once tool permits** - `POST /gate/tool-call` puts an agent's MCP or SDK call through the same gate as a pipeline step. Bearer authentication covers ask, poll, and consume. A stable `action_id` makes retries idempotent; the binding covers the exact arguments, policy, and expiry; only the first successful `POST /gate/tool-call//consume` carries `permit: execute`. See [`docs/tool-gate.md`](docs/tool-gate.md). - **Repeat guard** — inside `repeat_window` an identical tool call (same agent, tool, arguments) gets the gate's remembered answer instead of a new prompt: a denied call stays denied, an in-flight call joins the open prompt, and `max_repeats` stops a looping agent from paging you. - **Denial notices** — a refusal the gate makes on its own (unlisted tool, argument outside a rule, repeat guard) is reported to the operator channel, one notice per agent, tool and reason per window, so nothing is refused silently. - **Open gates** — `/pending` on the channel and `draftcat pending` on the host list every approval waiting on a human, pipeline steps and tool calls alike, with how long each has waited and how long it has left. @@ -114,9 +114,12 @@ However your agent runs, draftcat sits between it and your customer systems as a - **Escalation** — `escalate_after` re-notifies before a gate times out; `escalate_to` widens who is told, never who may decide. - **Durable, run-correlated gates** — every gate is written to SQLite before the draft goes out, so an approval in flight survives a restart, and each decision records the run it released. - **Approver scoping** — `approvers:` on a step narrows who may decide it to a subset of `allowed_users`. Quorum says *how many*; this says *which ones*. It can only narrow, never widen. +- **Model I/O policy** - ordered `model_policy` regex rules check exact input before it reaches the provider and output before it leaves Draftcat. A match can deny or enter the existing human approval gate, and the decision is written as a versioned receipt. - **Input sanitization** — operator input is scrubbed for prompt-injection patterns before the LLM. - **Output validation** — AI output is checked against the skill's `output_schema` (field types, numeric `min`/`max`, `enum` membership) and rejected if it doesn't conform. -- **Checked action receipts** — approval decisions can be tied to a payload hash and verified later; see [`docs/action-receipts.md`](docs/action-receipts.md). +- **Checked action receipts** - v2 receipts bind immutable action ID, payload hash, policy digest, validity window, run, and decision. List, inspect, or stream JSONL from SQLite with `draftcat receipts`; see [`docs/action-receipts.md`](docs/action-receipts.md). +- **Durable webhook admission** - Draftcat writes a body-hash-only admission row before returning HTTP 202. The response includes `admission_id` and an authenticated poll URL; unfinished admissions become `interrupted` after restart. +- **Health contract** - `GET /healthz` reports process liveness and `GET /readyz` succeeds only while the SQLite decision store is available. - **Private approval proofs** — share proof that a direct human approval met quorum without sharing the action, approver, or counts; see [`docs/zk-approval-proofs.md`](docs/zk-approval-proofs.md). - **Encrypted approval tally** — combine three or more encrypted votes without letting the collector read any individual vote; see [`docs/fhe-vote-tally.md`](docs/fhe-vote-tally.md). - **Rate limiting** — per-user, per-minute caps on operator interactions. @@ -192,6 +195,14 @@ curl -X POST https://draftcat.yourco.eu/hooks/ \ ``` The POST only **starts** a gated pipeline — the approval step still runs, so inbound can never make the LLM fire a customer-facing action. +Draftcat writes the admission to SQLite before returning `202`: + +```json +{"admission_id":"wh_...","status":"accepted","poll":"/hooks/status/wh_..."} +``` + +Poll that path with the same bearer token. `GET /healthz` is a liveness check; +`GET /readyz` verifies that the decision store is reachable. ## How it works @@ -283,6 +294,41 @@ tool_gate: on_mismatch: deny # outside it: refuse without asking ``` +Every gate request uses the webhook bearer token. Send a stable `action_id`, +then consume an allowed binding exactly once before running the side effect: + +```bash +curl -X POST http://127.0.0.1:8088/gate/tool-call \ + -H "Authorization: Bearer $DRAFTCAT_WEBHOOK_SECRET" \ + -H 'Content-Type: application/json' \ + -d '{"action_id":"send-invoice-4821","tool":"send_email","args":{"to":"billing@example.com"}}' + +curl -X POST http://127.0.0.1:8088/gate/tool-call/send-invoice-4821/consume \ + -H "Authorization: Bearer $DRAFTCAT_WEBHOOK_SECRET" \ + -H 'Content-Type: application/json' \ + -d '{"binding_hash":"sha256:..."}' +``` + +Model input and output policy is ordered and deterministic. `deny` fails the +LLM call closed; `review` pauses at the configured operator channel: + +```yaml +model_policy: + max_preview_chars: 800 + rules: + - id: credentials-in-input + phase: input + pattern: '(?i)(api[_ -]?key|password)' + action: review + reason: Credentials require an explicit operator decision. + - id: unsupported-claim + phase: output + roles: [drafter] + pattern: '(?i)guaranteed results' + action: deny + reason: Do not send unsupported guarantees. +``` + Skills are YAML prompt templates in `skills/` with an `output_schema` the engine enforces. With `-tags voice`, a `voice:` block configures the webhook receivers, Dograh endpoints, and pre-call lookup — see [docs/voice.md](docs/voice.md). ## Commands @@ -293,6 +339,9 @@ draftcat validate [--strict] # lint config + skills draftcat test # dry-run against fixtures// (never touches real APIs) draftcat runs [pipeline] # recent runs + the approval decisions in each (--json to archive) draftcat pending # approval gates waiting on a human right now (--json) +draftcat receipts list # approval receipts and verification status (--json) +draftcat receipts show # one versioned receipt +draftcat receipts export # JSONL to stdout (--out path writes mode 0600) draftcat audit-verify # verify signed approval receipts draftcat hitl verify # run the hitl/v0 conformance suite against a relay ``` @@ -328,7 +377,7 @@ curl -X POST http://127.0.0.1:8088/hooks/invoice-due-diligence \ -H "Authorization: Bearer $DRAFTCAT_WEBHOOK_SECRET" -d '{"path": "/inbox/invoice.pdf"}' ``` -The body reaches the pipeline as `{{webhook_body}}` / `{{input}}`; bearer auth is constant-time, and a second trigger while the pipeline is running gets `409`. A webhook only *starts* a pipeline — the approval gate still runs, so an inbound request can never make the LLM fire an outbound action. +The body reaches the pipeline as `{{webhook_body}}` / `{{input}}`; bearer auth is constant-time, and a second trigger while the pipeline is running gets `409`. Before `202`, Draftcat stores an admission ID, pipeline, body hash, and status in SQLite. `GET /hooks/status/` returns the authenticated status without retaining the request body. A webhook only *starts* a pipeline - the approval gate still runs, so an inbound request can never make the LLM fire an outbound action. **Signed requests.** Bind each trigger to its exact body and a timestamp with an HMAC receipt, on top of the bearer token: diff --git a/config.yaml b/config.yaml index 397c1b4..7bec70b 100644 --- a/config.yaml +++ b/config.yaml @@ -45,6 +45,18 @@ roles: classifier: gpt-4o-mini drafter: gpt-4o-mini +# Ordered checks applied before model input is sent and after output returns. +# A matching rule can fail closed or enter the normal operator approval gate. +# model_policy: +# max_preview_chars: 800 +# rules: +# - id: credentials-in-input +# phase: input # input | output | both +# roles: [drafter] # optional; empty means every role +# pattern: '(?i)(api[_ -]?key|password)' +# action: review # deny | review +# reason: Credentials require an explicit operator decision. + budgets: per_step_tokens: 2048 per_pipeline_tokens: 10000 @@ -67,9 +79,14 @@ observability: # addr: 127.0.0.1:8088 # secret_env: DRAFTCAT_WEBHOOK_SECRET # required when enabled (bearer token) # max_body_bytes: 65536 - -# Tool-call gate for an agent's MCP/SDK calls — POST /gate/tool-call on the -# webhook listener (requires webhook.enabled). Anything not listed is denied. +# require_signature: true # also signs gate POST bodies when enabled +# GET /healthz reports liveness. GET /readyz checks SQLite. Accepted triggers +# return an admission_id for authenticated GET /hooks/status/. + +# Tool-call gate for an agent's MCP/SDK calls - POST /gate/tool-call on the +# webhook listener (requires webhook.enabled). Every route requires the webhook +# bearer token. Send a stable action_id and consume an allowed binding once at +# POST /gate/tool-call//consume before executing the side effect. # See docs/tool-gate.md. # tool_gate: # enabled: true @@ -82,8 +99,8 @@ observability: # risk: high # require_approval: true # args: -# to: {glob: "*@example.com"} # arguments inside the rule → ask as usual -# on_mismatch: deny # outside it → refuse without asking +# to: {glob: "*@example.com"} # arguments inside the rule: ask as usual +# on_mismatch: deny # outside it: refuse without asking pipelines: # Uses skill reference instead of inline prompt diff --git a/docs/action-receipts.md b/docs/action-receipts.md index e648c31..2216c70 100644 --- a/docs/action-receipts.md +++ b/docs/action-receipts.md @@ -16,10 +16,12 @@ a persisted fact that can be inspected later. 5. The approved action executes. 6. A receipt can be exported for audit or incident review. -## Current integrity layer +## Integrity layer -The approval package signs immutable decision fields with HMAC-SHA256: +New decisions use receipt schema v2. The approval package signs immutable +decision fields with HMAC-SHA256: +- receipt, run, and action IDs - pipeline - step - decision time @@ -28,10 +30,14 @@ The approval package signs immutable decision fields with HMAC-SHA256: - payload hash - quorum requirement - quorum result +- policy and policy digest +- binding digest +- permit expiry and lifecycle - nonce See [`internal/approval/receipt.go`](../internal/approval/receipt.go). If any -covered field changes after signing, verification fails. +covered field changes after signing, verification fails. Existing v1 receipts +continue to verify with their original canonical field set. ## Receipt shape @@ -48,40 +54,59 @@ Example: ```json { - "receipt_id": "act_20260704_001", + "version": 2, + "receipt_id": "rcpt_9d4d...", "run_id": "run_abc123", + "action_id": "run_abc123:lead_reply:send_email", "pipeline": "lead_reply", "step": "send_email", - "action_type": "outbound_message", - "status": "executed", - "proposed_by": "agent", - "approved_by": "operator:12345", - "approved_at": "2026-07-04T09:30:00Z", - "executed_at": "2026-07-04T09:31:00Z", - "schema_version": 1, + "decision": "approve", + "operator_id": 12345, + "decided_at": "2026-09-17T09:30:00Z", "payload_hash": "sha256:...", - "signature_status": "valid", - "policy_checks": [ - {"id": "recipient_allowlist", "status": "pass"}, - {"id": "budget_limit", "status": "pass"} - ], - "human_decision": { - "decision": "edit_then_approve", - "notes": "Tightened the CTA and removed an unsupported claim." - } + "policy": "human-approval", + "policy_hash": "sha256:...", + "binding_hash": "sha256:...", + "expires_at": "2026-09-17T13:30:00Z", + "lifecycle": "decided", + "quorum_n": 1, + "quorum_got": 1, + "nonce": "...", + "signature": "...", + "verification": "ok" } ``` -## CLI direction +The SQLite row stores hashes and identifiers, not the customer payload. -A small receipt surface should be enough for operators and auditors: +## CLI + +List recent receipts across all pipelines, or narrow to one pipeline: + +```bash +draftcat receipts list --limit 100 +draftcat receipts list --pipeline lead_reply --json +``` + +Inspect one receipt by its stable ID (legacy rows also accept their numeric row +ID): ```bash -draftcat receipts list --run run_abc123 -draftcat receipts show act_20260704_001 -draftcat receipts export --format jsonl --out receipts.jsonl +draftcat receipts show rcpt_9d4d... ``` +Export newline-delimited JSON in chronological order. Stdout makes it easy to +pipe into an auditor or log shipper; `--out` creates a mode `0600` file: + +```bash +draftcat receipts export --pipeline lead_reply > receipts.jsonl +draftcat receipts export --out receipts.jsonl +``` + +Set `DRAFTCAT_APPROVAL_SECRET` while reading to receive `verification: ok` or +`tampered`. Signed rows without the key report `unverified`; unsigned rows +report `unsigned` explicitly. + ## Design rule Do not let the model decide whether the approval boundary was satisfied. The diff --git a/docs/tool-gate.md b/docs/tool-gate.md index abedb1f..28f4274 100644 --- a/docs/tool-gate.md +++ b/docs/tool-gate.md @@ -3,7 +3,9 @@ `POST /gate/tool-call` puts one tool call an agent is about to make through the same gate a pipeline step goes through: allowlist, risk tier, human approval, audit row. It is served on the webhook listener, so `tool_gate.enabled` requires -`webhook.enabled`. +`webhook.enabled`. Every ask, poll, and consume request requires the webhook +bearer token. If `webhook.require_signature` is enabled, gate POST bodies also +require `X-Draftcat-Signature`. Two properties matter more than convenience, and everything below preserves them: @@ -13,14 +15,19 @@ them: - **The arguments are part of the approval.** The gate hashes the exact arguments proposed and the human approves *those*. A harness that then calls the tool with different arguments leaves an audit trail that does not match. +- **One decision releases one execution.** An allowed decision is not itself an + execution permit. The caller atomically consumes its exact binding, and only + that first response carries `"permit":"execute"`. ## Asking ``` POST /gate/tool-call Content-Type: application/json +Authorization: Bearer { + "action_id": "send-invoice-2026-114", "tool": "send_email", "args": {"to": "anna@example.com", "subject": "Invoice 2026-114"}, "agent": "harness-1", @@ -28,26 +35,31 @@ Content-Type: application/json } ``` -| Field | Meaning | -|----------|---------| +| Field | Meaning | +|-------------|---------| +| `action_id` | Stable caller identity for this intended side effect. Matching retries return the existing state; changed data under the same ID gets `409`. Omitted clients receive a generated ID, but cannot safely retry across a restart without persisting it. | | `tool` | Required. Looked up in `tool_gate.tools`. | | `args` | The exact arguments the harness intends to call with. Hashed into the decision, never stored. | | `agent` | Free-form caller name, shown to the operator and used to key the repeat guard. | | `run_id` | Optional. Ties the decision to a pipeline run in the audit trail. | | `mode` | `sync` (default) or `async` — see *Collecting a decision*. | | `wait` | A duration such as `30s`. Bounds a sync hold; see below. | +| `expires_at` | Optional RFC3339 deadline that can narrow, never extend, the approval window. | Every answer has the same shape: ```json -{"decision": "allow", "reason": "operator approved", "args_hash": "sha256:9f2b...", - "decided_by": "operator", "approval_id": "tc_5c1e..."} +{"action_id":"send-invoice-2026-114","decision":"allow","state":"allowed", + "args_hash":"sha256:9f2b...","policy_hash":"sha256:41a0...", + "binding_hash":"sha256:acd1...","decided_by":"operator", + "consume":"/gate/tool-call/send-invoice-2026-114/consume","expires_at":"2026-09-17T18:00:00Z"} ``` `decision` is `allow`, `deny` or `pending`. `decided_by` says who settled it: `allowlist` (the rule itself), `policy` (an argument rule), `operator` (a human), or `repeat-guard`. `rule` names the config that produced a decision the gate made on its own, so a denial can be traced without opening the audit log. +`state` is `pending`, `allowed`, `denied`, `expired`, or `consumed`. ## What the rule decides @@ -118,17 +130,38 @@ server-side. **Async.** `"mode": "async"` answers `202 pending` immediately. -Either way the harness collects the decision from +Either way the harness collects the decision from (using the bearer token): ``` GET /gate/tool-call/ # 200 pending | allow | deny GET /gate/tool-call/?wait=30s # long-poll, capped at 100s ``` -An unknown id answers `404` with `decision: deny`. That is the fail-closed -answer for "the gate restarted, or the id is stale": there is no decision to -hand over, so the harness must ask again. Decided tickets stay collectable for -an hour. +An unknown id answers `404` with `decision: deny`. Durable action state remains +queryable after restart. Work that was still pending when the process stopped +becomes `expired`; it is never resumed from an unprovable in-memory point. + +## Consuming an allowed decision + +Do not execute on `state: allowed`. Atomically consume the exact binding first: + +``` +POST /gate/tool-call/send-invoice-2026-114/consume +Authorization: Bearer +Content-Type: application/json + +{"binding_hash":"sha256:acd1..."} +``` + +The first matching request before expiry returns `200` with +`"state":"consumed","permit":"execute"`. That response authorizes exactly one +execution of the bound action. A second consume, the wrong binding, a denial, +or an expired permit returns `409` without `permit: execute`. + +The action ledger is in SQLite. A retry using the same `action_id`, tool, +arguments, caller, run, policy, and expiry returns its current state without a +second prompt. Reusing the ID after any of those bound facts changes is a `409` +conflict and requires a new action ID and a fresh decision. A tool call waiting on a human is written to `pending_approvals` before the prompt goes out, exactly like a pipeline gate. It shows up in `/pending` and @@ -176,11 +209,11 @@ the notices off. ## Audit -Every decision writes an `action_approvals` row with `pipeline = tool-gate`, +Every decision writes a v2 `action_approvals` receipt with `pipeline = tool-gate`, `step = `, the argument hash as the payload hash, and the reason or rule -in the `policy` column: `unlisted`, `args mismatch: …`, `allowlisted risk=…`, -`repeat-guard: …`, `operator approved`. `draftcat runs --json` and -`draftcat audit-verify` see them like any other gate decision. +in the `policy` column. The signed envelope also carries the stable action ID, +policy digest, exact binding, run ID, and expiry. `draftcat receipts list`, +`show`, and `export` read the rows directly from SQLite. ## Configuration reference diff --git a/internal/approval/receipt.go b/internal/approval/receipt.go index c7c7627..289a090 100644 --- a/internal/approval/receipt.go +++ b/internal/approval/receipt.go @@ -41,6 +41,28 @@ type Fields struct { QuorumGot int } +// FieldsV2 is the canonical receipt envelope for decisions written by v0.7.0 +// and later. It keeps the v1 fields and binds the decision to an immutable +// action identity, the policy/configuration that evaluated it, and its +// validity window. Versioned signing lets old receipts continue to verify. +type FieldsV2 struct { + ReceiptID string + RunID string + ActionID string + Pipeline string + Step string + DecidedAt int64 + Decision string + OperatorID int64 + PayloadHash string + Policy string + PolicyHash string + BindingHash string + ExpiresAt int64 + QuorumN int + QuorumGot int +} + // canonical serializes (fields, nonce) into an unambiguous byte string. Each // value is length-prefixed so no combination of field values can collide with a // different set of fields — e.g. pipeline="a", step="b|c" must not sign the same @@ -68,6 +90,36 @@ func canonical(f Fields, nonce string) []byte { return []byte(b.String()) } +func canonicalV2(f FieldsV2, nonce string) []byte { + parts := []string{ + "2", + f.ReceiptID, + f.RunID, + f.ActionID, + f.Pipeline, + f.Step, + strconv.FormatInt(f.DecidedAt, 10), + f.Decision, + strconv.FormatInt(f.OperatorID, 10), + f.PayloadHash, + f.Policy, + f.PolicyHash, + f.BindingHash, + strconv.FormatInt(f.ExpiresAt, 10), + strconv.Itoa(f.QuorumN), + strconv.Itoa(f.QuorumGot), + nonce, + } + var b strings.Builder + for _, p := range parts { + b.WriteString(strconv.Itoa(len(p))) + b.WriteByte(':') + b.WriteString(p) + b.WriteByte('|') + } + return []byte(b.String()) +} + // NewNonce returns a random 128-bit nonce (hex). A per-row nonce means two // otherwise-identical decisions still produce distinct receipts, and it makes a // signed row impossible to replay onto a different row. @@ -94,3 +146,16 @@ func Verify(secret []byte, f Fields, nonce, sig string) bool { want := Sign(secret, f, nonce) return subtle.ConstantTimeCompare([]byte(want), []byte(sig)) == 1 } + +// SignV2 returns the HMAC-SHA256 signature for a v2 receipt envelope. +func SignV2(secret []byte, f FieldsV2, nonce string) string { + mac := hmac.New(sha256.New, secret) + mac.Write(canonicalV2(f, nonce)) + return hex.EncodeToString(mac.Sum(nil)) +} + +// VerifyV2 verifies a v2 receipt in constant time. +func VerifyV2(secret []byte, f FieldsV2, nonce, sig string) bool { + want := SignV2(secret, f, nonce) + return subtle.ConstantTimeCompare([]byte(want), []byte(sig)) == 1 +} diff --git a/internal/approval/receipt_test.go b/internal/approval/receipt_test.go index dc31424..5df6b31 100644 --- a/internal/approval/receipt_test.go +++ b/internal/approval/receipt_test.go @@ -111,3 +111,32 @@ func TestEmptySignatureFails(t *testing.T) { t.Fatal("empty signature verified") } } + +func TestV2BindsActionPolicyAndExpiry(t *testing.T) { + secret := []byte("test-secret") + f := FieldsV2{ + ReceiptID: "rcpt_1", RunID: "run_1", ActionID: "send_1", + Pipeline: "sales", Step: "send", DecidedAt: 1_750_000_000, + Decision: "approve", OperatorID: 42, PayloadHash: "sha256:payload", + Policy: "human-approval", PolicyHash: "sha256:policy", + BindingHash: "sha256:binding", ExpiresAt: 1_750_003_600, + QuorumN: 1, QuorumGot: 1, + } + nonce := "fixed" + sig := SignV2(secret, f, nonce) + if !VerifyV2(secret, f, nonce, sig) { + t.Fatal("valid v2 receipt did not verify") + } + for name, mutate := range map[string]func(*FieldsV2){ + "action": func(x *FieldsV2) { x.ActionID = "send_2" }, + "policy": func(x *FieldsV2) { x.PolicyHash = "sha256:changed" }, + "binding": func(x *FieldsV2) { x.BindingHash = "sha256:changed" }, + "expiry": func(x *FieldsV2) { x.ExpiresAt++ }, + } { + changed := f + mutate(&changed) + if VerifyV2(secret, changed, nonce, sig) { + t.Errorf("v2 receipt still verified after %s drift", name) + } + } +} diff --git a/internal/config/config.go b/internal/config/config.go index 24d0c24..a3e6744 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -22,21 +22,22 @@ import ( ) type Config struct { - Telegram TelegramConfig `yaml:"telegram"` - Relay RelayConfig `yaml:"relay"` - Gmail gmailapi.GmailConfig `yaml:"gmail"` - GHL ghlapi.GHLConfig `yaml:"gohighlevel"` - State StateConfig `yaml:"state"` - Provider ProviderConfig `yaml:"provider"` - Models map[string]ModelConfig `yaml:"models"` - Roles map[string]string `yaml:"roles"` - Budgets BudgetConfig `yaml:"budgets"` - Timeouts TimeoutConfig `yaml:"timeouts"` - Policy ApprovalPolicy `yaml:"approval_policy"` - ToolGate ToolGateConfig `yaml:"tool_gate"` - Webhook WebhookConfig `yaml:"webhook"` - Observ ObservabilityConfig `yaml:"observability"` - Pipelines []PipelineConfig `yaml:"pipelines"` + Telegram TelegramConfig `yaml:"telegram"` + Relay RelayConfig `yaml:"relay"` + Gmail gmailapi.GmailConfig `yaml:"gmail"` + GHL ghlapi.GHLConfig `yaml:"gohighlevel"` + State StateConfig `yaml:"state"` + Provider ProviderConfig `yaml:"provider"` + Models map[string]ModelConfig `yaml:"models"` + Roles map[string]string `yaml:"roles"` + Budgets BudgetConfig `yaml:"budgets"` + Timeouts TimeoutConfig `yaml:"timeouts"` + Policy ApprovalPolicy `yaml:"approval_policy"` + ModelPolicy ModelPolicyConfig `yaml:"model_policy"` + ToolGate ToolGateConfig `yaml:"tool_gate"` + Webhook WebhookConfig `yaml:"webhook"` + Observ ObservabilityConfig `yaml:"observability"` + Pipelines []PipelineConfig `yaml:"pipelines"` // Voice is parsed unconditionally as raw YAML. Decoded into voice.Config // only when draftcat is built with -tags voice. Lean builds ignore it. Voice yaml.Node `yaml:"voice"` @@ -194,6 +195,61 @@ type ModelConfig struct { CostOut float64 `yaml:"cost_per_1k_output"` } +// ModelPolicyConfig applies deterministic checks immediately before model +// input is sent and immediately after model output is received. +type ModelPolicyConfig struct { + Rules []ModelPolicyRule `yaml:"rules"` + MaxPreviewChars int `yaml:"max_preview_chars"` +} + +// ModelPolicyRule is an ordered regex rule. Action is deny or review. +type ModelPolicyRule struct { + ID string `yaml:"id"` + Phase string `yaml:"phase"` // input | output | both + Roles []string `yaml:"roles"` + Pattern string `yaml:"pattern"` + Action string `yaml:"action"` // deny | review + Reason string `yaml:"reason"` +} + +func (p ModelPolicyConfig) PreviewLimit() int { + if p.MaxPreviewChars > 0 { + return p.MaxPreviewChars + } + return 800 +} + +// Match returns the first applicable rule, keeping evaluation deterministic. +func (p ModelPolicyConfig) Match(role, phase, text string) (*ModelPolicyRule, error) { + for i := range p.Rules { + r := &p.Rules[i] + rPhase := strings.ToLower(strings.TrimSpace(r.Phase)) + if rPhase != "both" && rPhase != phase { + continue + } + if len(r.Roles) > 0 { + matchedRole := false + for _, candidate := range r.Roles { + if candidate == role { + matchedRole = true + break + } + } + if !matchedRole { + continue + } + } + re, err := regexp.Compile(r.Pattern) + if err != nil { + return nil, fmt.Errorf("model policy rule %q: %w", r.ID, err) + } + if re.MatchString(text) { + return r, nil + } + } + return nil, nil +} + type BudgetConfig struct { PerStepTokens int `yaml:"per_step_tokens"` PerPipelineTokens int `yaml:"per_pipeline_tokens"` diff --git a/internal/config/model_policy_test.go b/internal/config/model_policy_test.go new file mode 100644 index 0000000..1982896 --- /dev/null +++ b/internal/config/model_policy_test.go @@ -0,0 +1,27 @@ +package config + +import "testing" + +func TestModelPolicyFirstMatchByPhaseAndRole(t *testing.T) { + p := ModelPolicyConfig{Rules: []ModelPolicyRule{ + {ID: "input-secret", Phase: "input", Roles: []string{"drafter"}, Pattern: `(?i)secret`, Action: "review"}, + {ID: "output-id", Phase: "output", Pattern: `ID-[0-9]+`, Action: "deny"}, + }} + r, err := p.Match("drafter", "input", "contains SECRET") + if err != nil || r == nil || r.ID != "input-secret" { + t.Fatalf("match=%+v err=%v", r, err) + } + if r, err = p.Match("classifier", "input", "contains secret"); err != nil || r != nil { + t.Fatalf("role filter=%+v err=%v", r, err) + } + if r, err = p.Match("classifier", "output", "ID-42"); err != nil || r == nil || r.ID != "output-id" { + t.Fatalf("output=%+v err=%v", r, err) + } +} + +func TestModelPolicyRejectsBadRegexAtRuntime(t *testing.T) { + p := ModelPolicyConfig{Rules: []ModelPolicyRule{{ID: "bad", Phase: "both", Pattern: `[`, Action: "deny"}}} + if _, err := p.Match("drafter", "input", "x"); err == nil { + t.Fatal("bad regex was accepted") + } +} diff --git a/internal/state/execution_ledger_test.go b/internal/state/execution_ledger_test.go new file mode 100644 index 0000000..d2c68ac --- /dev/null +++ b/internal/state/execution_ledger_test.go @@ -0,0 +1,73 @@ +package state + +import ( + "testing" + "time" +) + +func TestToolActionIdempotencyAndConsumeOnce(t *testing.T) { + s := openTestStore(t) + now := time.Unix(1_800_000_000, 0) + a := ToolAction{ + ActionID: "send-1", Tool: "send_email", Agent: "agent", ArgsHash: "sha256:args", + PolicyHash: "sha256:policy", BindingHash: "sha256:binding", + CreatedAt: now, UpdatedAt: now, ExpiresAt: now.Add(time.Hour), + } + got, created, err := s.ReserveToolAction(a) + if err != nil || !created || got.Status != "pending" { + t.Fatalf("reserve = %+v created=%v err=%v", got, created, err) + } + if _, created, err = s.ReserveToolAction(a); err != nil || created { + t.Fatalf("matching retry created=%v err=%v", created, err) + } + drift := a + drift.BindingHash = "sha256:drift" + if _, _, err = s.ReserveToolAction(drift); err == nil { + t.Fatal("binding drift under one action id was accepted") + } + if err = s.DecideToolAction(a.ActionID, "allowed", "allow", "approved", "operator", now); err != nil { + t.Fatal(err) + } + if _, ok, err := s.ConsumeToolAction(a.ActionID, a.BindingHash, now.Add(time.Minute)); err != nil || !ok { + t.Fatalf("first consume ok=%v err=%v", ok, err) + } + if final, ok, err := s.ConsumeToolAction(a.ActionID, a.BindingHash, now.Add(2*time.Minute)); err != nil || ok || final.Status != "consumed" { + t.Fatalf("second consume = %+v ok=%v err=%v", final, ok, err) + } +} + +func TestWebhookAdmissionSurvivesAndInterrupts(t *testing.T) { + s := openTestStore(t) + now := time.Unix(1_800_000_000, 0) + a := WebhookAdmission{ID: "wh_1", Pipeline: "p", BodyHash: "sha256:body", CreatedAt: now, UpdatedAt: now} + if err := s.BeginWebhookAdmission(a); err != nil { + t.Fatal(err) + } + if err := s.InterruptWebhookAdmissions(now.Add(time.Minute)); err != nil { + t.Fatal(err) + } + got, err := s.WebhookAdmission(a.ID) + if err != nil || got.Status != "interrupted" || got.BodyHash != a.BodyHash { + t.Fatalf("admission = %+v err=%v", got, err) + } +} + +func TestReconcilePreservesUnexpiredAllowedPermit(t *testing.T) { + s := openTestStore(t) + now := time.Unix(1_800_000_000, 0) + a := ToolAction{ActionID: "a", Tool: "send", ArgsHash: "h", PolicyHash: "p", BindingHash: "b", + CreatedAt: now, UpdatedAt: now, ExpiresAt: now.Add(time.Hour)} + if _, _, err := s.ReserveToolAction(a); err != nil { + t.Fatal(err) + } + if err := s.DecideToolAction("a", "allowed", "allow", "approved", "operator", now); err != nil { + t.Fatal(err) + } + if err := s.ExpireToolActions(now.Add(time.Minute)); err != nil { + t.Fatal(err) + } + got, err := s.ToolAction("a") + if err != nil || got.Status != "allowed" { + t.Fatalf("permit=%+v err=%v", got, err) + } +} diff --git a/internal/state/migration_v2_test.go b/internal/state/migration_v2_test.go new file mode 100644 index 0000000..58975b6 --- /dev/null +++ b/internal/state/migration_v2_test.go @@ -0,0 +1,47 @@ +package state + +import ( + "context" + "database/sql" + "path/filepath" + "testing" + "time" +) + +func TestV2SchemaMigratesExistingApprovalRows(t *testing.T) { + path := filepath.Join(t.TempDir(), "state.db") + db, err := sql.Open("sqlite", path) + if err != nil { + t.Fatal(err) + } + _, err = db.ExecContext(context.Background(), `CREATE TABLE action_approvals ( + id INTEGER PRIMARY KEY AUTOINCREMENT, pipeline TEXT NOT NULL, step TEXT NOT NULL, + decided_at INTEGER NOT NULL, decision TEXT NOT NULL, operator_id INTEGER NOT NULL, + payload_hash TEXT NOT NULL, quorum_n INTEGER NOT NULL DEFAULT 1, + quorum_got INTEGER NOT NULL DEFAULT 1, nonce TEXT NOT NULL DEFAULT '', + signature TEXT NOT NULL DEFAULT '', run_id TEXT NOT NULL DEFAULT '', policy TEXT NOT NULL DEFAULT ''); + INSERT INTO action_approvals (pipeline, step, decided_at, decision, operator_id, payload_hash) + VALUES ('p', 'gate', 1800000000, 'approve', 7, 'sha256:old')`) + if err != nil { + t.Fatal(err) + } + if err := db.Close(); err != nil { + t.Fatal(err) + } + + st, err := OpenStateStore(path) + if err != nil { + t.Fatalf("migrate: %v", err) + } + defer func() { _ = st.Close() }() + rows, err := st.AllApprovals(10) + if err != nil || len(rows) != 1 { + t.Fatalf("rows=%+v err=%v", rows, err) + } + if rows[0].Version != 1 || rows[0].Lifecycle != "decided" || rows[0].DecidedAt != time.Unix(1_800_000_000, 0) { + t.Fatalf("legacy row changed during migration: %+v", rows[0]) + } + if _, _, err := st.ReserveToolAction(ToolAction{ActionID: "a", Tool: "t", ArgsHash: "h", PolicyHash: "p", BindingHash: "b", CreatedAt: time.Now(), UpdatedAt: time.Now(), ExpiresAt: time.Now().Add(time.Hour)}); err != nil { + t.Fatalf("new action ledger unavailable after migration: %v", err) + } +} diff --git a/internal/state/state.go b/internal/state/state.go index d417336..8a3fa34 100644 --- a/internal/state/state.go +++ b/internal/state/state.go @@ -72,7 +72,14 @@ CREATE TABLE IF NOT EXISTS action_approvals ( nonce TEXT NOT NULL DEFAULT '', -- per-row random; anti-replay signature TEXT NOT NULL DEFAULT '', -- HMAC receipt over the row's fields (empty = unsigned) run_id TEXT NOT NULL DEFAULT '', -- the pipeline run this decision released - policy TEXT NOT NULL DEFAULT '' -- rule that released it when decision='policy_approve' + policy TEXT NOT NULL DEFAULT '', -- rule that released it when decision='policy_approve' + receipt_version INTEGER NOT NULL DEFAULT 1, + receipt_id TEXT NOT NULL DEFAULT '', + action_id TEXT NOT NULL DEFAULT '', + policy_hash TEXT NOT NULL DEFAULT '', + binding_hash TEXT NOT NULL DEFAULT '', + expires_at INTEGER NOT NULL DEFAULT 0, + lifecycle TEXT NOT NULL DEFAULT 'decided' ); CREATE INDEX IF NOT EXISTS idx_approvals_pipeline ON action_approvals(pipeline, decided_at DESC); CREATE INDEX IF NOT EXISTS idx_approvals_runid ON action_approvals(run_id); @@ -87,6 +94,34 @@ CREATE TABLE IF NOT EXISTS pending_approvals ( status TEXT NOT NULL -- pending|resolved|interrupted ); CREATE INDEX IF NOT EXISTS idx_pending_status ON pending_approvals(status, opened_at); +CREATE TABLE IF NOT EXISTS tool_actions ( + action_id TEXT PRIMARY KEY, + tool TEXT NOT NULL, + agent TEXT NOT NULL DEFAULT '', + run_id TEXT NOT NULL DEFAULT '', + args_hash TEXT NOT NULL, + policy_hash TEXT NOT NULL, + binding_hash TEXT NOT NULL, + status TEXT NOT NULL, + decision TEXT NOT NULL DEFAULT '', + reason TEXT NOT NULL DEFAULT '', + decided_by TEXT NOT NULL DEFAULT '', + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + expires_at INTEGER NOT NULL, + consumed_at INTEGER NOT NULL DEFAULT 0 +); +CREATE INDEX IF NOT EXISTS idx_tool_actions_status ON tool_actions(status, updated_at); +CREATE TABLE IF NOT EXISTS webhook_admissions ( + id TEXT PRIMARY KEY, + pipeline TEXT NOT NULL, + body_hash TEXT NOT NULL, + status TEXT NOT NULL, + error_text TEXT NOT NULL DEFAULT '', + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_webhook_admissions_status ON webhook_admissions(status, created_at); ` if _, err := db.ExecContext(context.Background(), schema); err != nil { return err @@ -105,6 +140,13 @@ CREATE INDEX IF NOT EXISTS idx_pending_status ON pending_approvals(status, opene `ALTER TABLE action_approvals ADD COLUMN run_id TEXT NOT NULL DEFAULT ''`, `ALTER TABLE pipeline_runs ADD COLUMN run_id TEXT NOT NULL DEFAULT ''`, `ALTER TABLE action_approvals ADD COLUMN policy TEXT NOT NULL DEFAULT ''`, + `ALTER TABLE action_approvals ADD COLUMN receipt_version INTEGER NOT NULL DEFAULT 1`, + `ALTER TABLE action_approvals ADD COLUMN receipt_id TEXT NOT NULL DEFAULT ''`, + `ALTER TABLE action_approvals ADD COLUMN action_id TEXT NOT NULL DEFAULT ''`, + `ALTER TABLE action_approvals ADD COLUMN policy_hash TEXT NOT NULL DEFAULT ''`, + `ALTER TABLE action_approvals ADD COLUMN binding_hash TEXT NOT NULL DEFAULT ''`, + `ALTER TABLE action_approvals ADD COLUMN expires_at INTEGER NOT NULL DEFAULT 0`, + `ALTER TABLE action_approvals ADD COLUMN lifecycle TEXT NOT NULL DEFAULT 'decided'`, } { if _, err := db.ExecContext(context.Background(), alter); err != nil && !strings.Contains(err.Error(), "duplicate column name") { return err @@ -266,6 +308,11 @@ func (s *StateStore) AllRecentRuns(n int) ([]RunRecord, error) { } type ApprovalRecord struct { + ID int64 + Version int + ReceiptID string + RunID string + ActionID string Pipeline string Step string DecidedAt time.Time @@ -276,6 +323,34 @@ type ApprovalRecord struct { QuorumGot int Nonce string Signature string + Policy string + PolicyHash string + BindingHash string + ExpiresAt time.Time + Lifecycle string +} + +// ApprovalEnvelope contains every immutable field written into a v2 receipt. +// Payloads remain represented only by hashes. +type ApprovalEnvelope struct { + ReceiptID string + RunID string + ActionID string + Pipeline string + Step string + DecidedAt time.Time + Decision string + OperatorID int64 + PayloadHash string + QuorumN int + QuorumGot int + Nonce string + Signature string + Policy string + PolicyHash string + BindingHash string + ExpiresAt time.Time + Lifecycle string } // RecordApproval appends one approval-decision row. Called on every terminal @@ -313,11 +388,52 @@ func (s *StateStore) RecordApprovalRow(runID, pipeline, step string, decidedAt t return err } +// RecordApprovalV2 appends a versioned, action-bound receipt. Legacy writers +// continue through RecordApprovalRow and retain v1 verification semantics. +func (s *StateStore) RecordApprovalV2(e ApprovalEnvelope) error { + lifecycle := e.Lifecycle + if lifecycle == "" { + lifecycle = "decided" + } + _, err := s.db.ExecContext(context.Background(), + `INSERT INTO action_approvals + (pipeline, step, decided_at, decision, operator_id, payload_hash, quorum_n, quorum_got, + nonce, signature, run_id, policy, receipt_version, receipt_id, action_id, policy_hash, + binding_hash, expires_at, lifecycle) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 2, ?, ?, ?, ?, ?, ?)`, + e.Pipeline, e.Step, e.DecidedAt.Unix(), e.Decision, e.OperatorID, e.PayloadHash, + e.QuorumN, e.QuorumGot, e.Nonce, e.Signature, e.RunID, e.Policy, e.ReceiptID, + e.ActionID, e.PolicyHash, e.BindingHash, e.ExpiresAt.Unix(), lifecycle, + ) + return err +} + +const approvalColumns = `id, receipt_version, receipt_id, run_id, action_id, + pipeline, step, decided_at, decision, operator_id, payload_hash, quorum_n, quorum_got, + nonce, signature, policy, policy_hash, binding_hash, expires_at, lifecycle` + +func scanApproval(scan func(...interface{}) error) (ApprovalRecord, error) { + var r ApprovalRecord + var decided, expires int64 + err := scan(&r.ID, &r.Version, &r.ReceiptID, &r.RunID, &r.ActionID, + &r.Pipeline, &r.Step, &decided, &r.Decision, &r.OperatorID, &r.PayloadHash, + &r.QuorumN, &r.QuorumGot, &r.Nonce, &r.Signature, &r.Policy, &r.PolicyHash, + &r.BindingHash, &expires, &r.Lifecycle) + if err != nil { + return r, err + } + r.DecidedAt = time.Unix(decided, 0) + if expires > 0 { + r.ExpiresAt = time.Unix(expires, 0) + } + return r, nil +} + // ApprovalsForRun returns every approval decision recorded against one run, in // decision order. This is the join the audit trail previously could not make. func (s *StateStore) ApprovalsForRun(runID string) ([]ApprovalRecord, error) { rows, err := s.db.QueryContext(context.Background(), - `SELECT pipeline, step, decided_at, decision, operator_id, payload_hash, quorum_n, quorum_got, nonce, signature + `SELECT `+approvalColumns+` FROM action_approvals WHERE run_id=? ORDER BY decided_at ASC, id ASC`, runID, ) @@ -327,12 +443,10 @@ func (s *StateStore) ApprovalsForRun(runID string) ([]ApprovalRecord, error) { defer func() { _ = rows.Close() }() var out []ApprovalRecord for rows.Next() { - var r ApprovalRecord - var ts int64 - if err := rows.Scan(&r.Pipeline, &r.Step, &ts, &r.Decision, &r.OperatorID, &r.PayloadHash, &r.QuorumN, &r.QuorumGot, &r.Nonce, &r.Signature); err != nil { + r, err := scanApproval(rows.Scan) + if err != nil { return nil, err } - r.DecidedAt = time.Unix(ts, 0) out = append(out, r) } return out, rows.Err() @@ -342,7 +456,7 @@ func (s *StateStore) ApprovalsForRun(runID string) ([]ApprovalRecord, error) { // newest-first. func (s *StateStore) ApprovalsForPipeline(pipeline string, n int) ([]ApprovalRecord, error) { rows, err := s.db.QueryContext(context.Background(), - `SELECT pipeline, step, decided_at, decision, operator_id, payload_hash, quorum_n, quorum_got, nonce, signature + `SELECT `+approvalColumns+` FROM action_approvals WHERE pipeline=? ORDER BY decided_at DESC, id DESC LIMIT ?`, pipeline, n, ) @@ -352,23 +466,79 @@ func (s *StateStore) ApprovalsForPipeline(pipeline string, n int) ([]ApprovalRec defer func() { _ = rows.Close() }() var out []ApprovalRecord for rows.Next() { - var r ApprovalRecord - var ts int64 - if err := rows.Scan(&r.Pipeline, &r.Step, &ts, &r.Decision, &r.OperatorID, &r.PayloadHash, &r.QuorumN, &r.QuorumGot, &r.Nonce, &r.Signature); err != nil { + r, err := scanApproval(rows.Scan) + if err != nil { + return nil, err + } + out = append(out, r) + } + return out, rows.Err() +} + +// AllApprovals returns the newest receipt rows across every pipeline. +func (s *StateStore) AllApprovals(n int) ([]ApprovalRecord, error) { + rows, err := s.db.QueryContext(context.Background(), + `SELECT `+approvalColumns+` FROM action_approvals ORDER BY decided_at DESC, id DESC LIMIT ?`, n) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + var out []ApprovalRecord + for rows.Next() { + r, err := scanApproval(rows.Scan) + if err != nil { return nil, err } - r.DecidedAt = time.Unix(ts, 0) out = append(out, r) } return out, rows.Err() } +// ApprovalByReceiptID looks up one receipt by its stable public id. +func (s *StateStore) ApprovalByReceiptID(id string) (ApprovalRecord, error) { + row := s.db.QueryRowContext(context.Background(), + `SELECT `+approvalColumns+` FROM action_approvals WHERE receipt_id=? OR (receipt_id='' AND CAST(id AS TEXT)=?) ORDER BY id DESC LIMIT 1`, id, id) + return scanApproval(row.Scan) +} + // ApprovalVerification pairs an audit row with the result of checking its receipt. type ApprovalVerification struct { Record ApprovalRecord Status string // "ok" | "tampered" | "unsigned" } +// VerifyApprovalRecord checks one receipt and preserves the explicit unsigned +// state for rows written without a signing secret. +func VerifyApprovalRecord(secret []byte, r ApprovalRecord) string { + if r.Signature == "" { + return "unsigned" + } + ok := false + if r.Version >= 2 { + expires := int64(0) + if !r.ExpiresAt.IsZero() { + expires = r.ExpiresAt.Unix() + } + ok = approval.VerifyV2(secret, approval.FieldsV2{ + ReceiptID: r.ReceiptID, RunID: r.RunID, ActionID: r.ActionID, + Pipeline: r.Pipeline, Step: r.Step, DecidedAt: r.DecidedAt.Unix(), + Decision: r.Decision, OperatorID: r.OperatorID, PayloadHash: r.PayloadHash, + Policy: r.Policy, PolicyHash: r.PolicyHash, BindingHash: r.BindingHash, + ExpiresAt: expires, QuorumN: r.QuorumN, QuorumGot: r.QuorumGot, + }, r.Nonce, r.Signature) + } else { + ok = approval.Verify(secret, approval.Fields{ + Pipeline: r.Pipeline, Step: r.Step, DecidedAt: r.DecidedAt.Unix(), + Decision: r.Decision, OperatorID: r.OperatorID, PayloadHash: r.PayloadHash, + QuorumN: r.QuorumN, QuorumGot: r.QuorumGot, + }, r.Nonce, r.Signature) + } + if ok { + return "ok" + } + return "tampered" +} + // VerifyApprovals re-checks the receipts on the last n approval rows for a // pipeline under secret. "tampered" means the row's fields no longer match its // signature — someone altered the audit trail after the decision was recorded. @@ -382,19 +552,7 @@ func (s *StateStore) VerifyApprovals(secret []byte, pipeline string, n int) ([]A } out := make([]ApprovalVerification, 0, len(recs)) for _, r := range recs { - status := "unsigned" - if r.Signature != "" { - f := approval.Fields{ - Pipeline: r.Pipeline, Step: r.Step, DecidedAt: r.DecidedAt.Unix(), - Decision: r.Decision, OperatorID: r.OperatorID, PayloadHash: r.PayloadHash, - QuorumN: r.QuorumN, QuorumGot: r.QuorumGot, - } - if approval.Verify(secret, f, r.Nonce, r.Signature) { - status = "ok" - } else { - status = "tampered" - } - } + status := VerifyApprovalRecord(secret, r) out = append(out, ApprovalVerification{Record: r, Status: status}) } return out, nil @@ -537,6 +695,217 @@ func (s *StateStore) MarkInterrupted(id int64) error { return err } +// ToolAction is the durable state machine behind an execution permit. +type ToolAction struct { + ActionID string + Tool string + Agent string + RunID string + ArgsHash string + PolicyHash string + BindingHash string + Status string + Decision string + Reason string + DecidedBy string + CreatedAt time.Time + UpdatedAt time.Time + ExpiresAt time.Time + ConsumedAt time.Time +} + +func scanToolAction(scan func(...interface{}) error) (ToolAction, error) { + var a ToolAction + var created, updated, expires, consumed int64 + err := scan(&a.ActionID, &a.Tool, &a.Agent, &a.RunID, &a.ArgsHash, &a.PolicyHash, + &a.BindingHash, &a.Status, &a.Decision, &a.Reason, &a.DecidedBy, + &created, &updated, &expires, &consumed) + if err != nil { + return a, err + } + a.CreatedAt, a.UpdatedAt, a.ExpiresAt = time.Unix(created, 0), time.Unix(updated, 0), time.Unix(expires, 0) + if consumed > 0 { + a.ConsumedAt = time.Unix(consumed, 0) + } + return a, nil +} + +const toolActionColumns = `action_id, tool, agent, run_id, args_hash, policy_hash, + binding_hash, status, decision, reason, decided_by, created_at, updated_at, expires_at, consumed_at` + +// ReserveToolAction atomically creates an action identity or returns the +// existing record for an idempotent retry. A mismatched binding is an error. +func (s *StateStore) ReserveToolAction(a ToolAction) (ToolAction, bool, error) { + if s == nil || s.db == nil { + return a, true, nil + } + res, err := s.db.ExecContext(context.Background(), + `INSERT OR IGNORE INTO tool_actions + (action_id, tool, agent, run_id, args_hash, policy_hash, binding_hash, status, + decision, reason, decided_by, created_at, updated_at, expires_at) + VALUES (?, ?, ?, ?, ?, ?, ?, 'pending', '', '', '', ?, ?, ?)`, + a.ActionID, a.Tool, a.Agent, a.RunID, a.ArgsHash, a.PolicyHash, a.BindingHash, + a.CreatedAt.Unix(), a.UpdatedAt.Unix(), a.ExpiresAt.Unix()) + if err != nil { + return ToolAction{}, false, err + } + n, _ := res.RowsAffected() + existing, err := s.ToolAction(a.ActionID) + if err != nil { + return ToolAction{}, false, err + } + if existing.BindingHash != a.BindingHash { + return existing, false, fmt.Errorf("action_id %q is already bound to different action data", a.ActionID) + } + return existing, n == 1, nil +} + +// ToolAction returns one durable permit state. +func (s *StateStore) ToolAction(id string) (ToolAction, error) { + if s == nil || s.db == nil { + return ToolAction{}, sql.ErrNoRows + } + row := s.db.QueryRowContext(context.Background(), + `SELECT `+toolActionColumns+` FROM tool_actions WHERE action_id=?`, id) + return scanToolAction(row.Scan) +} + +// DecideToolAction transitions a pending action to allowed or denied. +func (s *StateStore) DecideToolAction(id, status, decision, reason, decidedBy string, at time.Time) error { + if s == nil || s.db == nil { + return nil + } + _, err := s.db.ExecContext(context.Background(), + `UPDATE tool_actions SET status=?, decision=?, reason=?, decided_by=?, updated_at=? + WHERE action_id=? AND status='pending'`, status, decision, reason, decidedBy, at.Unix(), id) + return err +} + +// ConsumeToolAction atomically spends one allowed permit. Only the first +// matching caller before expiry succeeds. +func (s *StateStore) ConsumeToolAction(id, bindingHash string, at time.Time) (ToolAction, bool, error) { + if s == nil || s.db == nil { + return ToolAction{}, false, nil + } + res, err := s.db.ExecContext(context.Background(), + `UPDATE tool_actions SET status='consumed', consumed_at=?, updated_at=? + WHERE action_id=? AND binding_hash=? AND status='allowed' AND expires_at>=?`, + at.Unix(), at.Unix(), id, bindingHash, at.Unix()) + if err != nil { + return ToolAction{}, false, err + } + n, _ := res.RowsAffected() + if n == 0 { + if _, err := s.db.ExecContext(context.Background(), + `UPDATE tool_actions SET status='expired', decision='deny', + reason='permit expired before consumption', updated_at=? + WHERE action_id=? AND binding_hash=? AND status='allowed' AND expires_at 0 { - n, nerr := approval.NewNonce() - if nerr != nil { - log.Printf("[pipeline:%s][step:%s] approval nonce failed, recording unsigned: %v", pipeline.Name, step.Name, nerr) - } else { - nonce = n - sig = approval.Sign(approvalSecret, approval.Fields{ - Pipeline: pipeline.Name, Step: step.Name, DecidedAt: decidedAt.Unix(), - Decision: decision, OperatorID: opID, PayloadHash: payloadHash, - QuorumN: quorumN, QuorumGot: got, - }, nonce) - } - } - if e := state.RecordApprovalForRun(runID, pipeline.Name, step.Name, decidedAt, decision, opID, payloadHash, quorumN, got, nonce, sig); e != nil { + envelope := newApprovalEnvelope(approvalSecret, runID, pipeline.Name, step, + decidedAt, decidedAt.Add(approvalTimeout), decision, opID, payloadHash, + quorumN, got, "human-approval") + if e := state.RecordApprovalV2(envelope); e != nil { log.Printf("[pipeline:%s][step:%s] audit write failed: %v", pipeline.Name, step.Name, e) } } @@ -1754,8 +1746,11 @@ Description: We need an experienced LLM engineer to build a retrieval-augmented log.Printf("[pipeline:%s][step:%s] released by approval_policy (%s) — no operator prompt", pipeline.Name, step.Name, reason) obs.RecordApproval(pipeline.Name, step.Name, "policy_approve") if state != nil { - if e := state.RecordApprovalRow(runID, pipeline.Name, step.Name, time.Now(), - "policy_approve", 0, ph, quorumN, quorumN, "", "", reason); e != nil { + decidedAt := time.Now() + envelope := newApprovalEnvelope(approvalSecret, runID, pipeline.Name, step, + decidedAt, decidedAt.Add(approvalTimeout), "policy_approve", 0, ph, + quorumN, quorumN, reason) + if e := state.RecordApprovalV2(envelope); e != nil { log.Printf("[pipeline:%s][step:%s] audit write failed: %v", pipeline.Name, step.Name, e) } } @@ -2666,6 +2661,8 @@ func main() { os.Exit(runRunsCmd(os.Args[2:])) case "pending": os.Exit(runPendingCmd(os.Args[2:])) + case "receipts": + os.Exit(runReceiptsCmd(os.Args[2:])) case "hitl": os.Exit(runHitlCmd(os.Args[2:])) case "-h", "--help", "help": @@ -2677,6 +2674,7 @@ func main() { fmt.Println(" draftcat test dry-run a pipeline using fixtures//") fmt.Println(" draftcat runs [pipeline] [--json] recent runs + the approval decisions in each") fmt.Println(" draftcat pending [--json] approval gates waiting on a human right now") + fmt.Println(" draftcat receipts inspect and export verification-ready receipts") fmt.Println(" draftcat audit-verify check approval-receipt signatures (needs DRAFTCAT_APPROVAL_SECRET)") fmt.Println(" draftcat zk-receipt prove an approval without revealing its private fields") fmt.Println(" draftcat fhe-vote count encrypted approval votes without reading them") @@ -2864,6 +2862,9 @@ func main() { // engine starts, so no gate is left in an unknown state and no stale button // looks live. Needs the bot, hence its position after it. reconcileInterruptedApprovals(state, opChan) + if err := state.InterruptWebhookAdmissions(time.Now()); err != nil { + log.Printf("[webhook] reconcile admissions: %v", err) + } // Observability — structured span emission (off unless opted in). if cfg.Observ.Spans || os.Getenv("DRAFTCAT_TRACE") != "" { @@ -3299,16 +3300,93 @@ func newWebhookHandler(cfg *config.Config, sched *Scheduler, budget *BudgetTrack hookable := webhookPipelines(cfg) mux := http.NewServeMux() + mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + w.WriteHeader(http.StatusMethodNotAllowed) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte("{\"status\":\"ok\"}\n")) + }) + mux.HandleFunc("/readyz", func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + w.WriteHeader(http.StatusMethodNotAllowed) + return + } + ctx, cancel := context.WithTimeout(r.Context(), 2*time.Second) + defer cancel() + if state == nil || state.Ping(ctx) != nil { + http.Error(w, "not ready", http.StatusServiceUnavailable) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte("{\"status\":\"ready\"}\n")) + }) // Tool-call gate: an agent harness asks permission for one tool call. // Mounted on the same server as the webhook trigger so enabling it opens no // additional port. Default-deny lives in the handler. if cfg.ToolGate.Enabled { gate := newToolGate(cfg, gateChannel(bot)) - mux.HandleFunc(toolGatePath, gate.HandleCall) - mux.HandleFunc(toolGatePath+"/", gate.HandleStatus) + gateAuth := func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + const prefix = "Bearer " + got := r.Header.Get("Authorization") + if !strings.HasPrefix(got, prefix) || + subtle.ConstantTimeCompare([]byte(strings.TrimPrefix(got, prefix)), secret) != 1 { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + if r.Method == http.MethodPost { + body, _ := io.ReadAll(io.LimitReader(r.Body, maxBody)) + r.Body = io.NopCloser(bytes.NewReader(body)) + sigHeader := r.Header.Get(webhookSigHeader) + if sigHeader != "" || cfg.Webhook.RequireSignature { + if err := verifyWebhookSignature(sigHeader, body, secret, maxSkew, time.Now()); err != nil { + log.Printf("[tool-gate][security] signature rejected: %v", err) + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + } + } + next.ServeHTTP(w, r) + }) + } + mux.Handle(toolGatePath, gateAuth(http.HandlerFunc(gate.HandleCall))) + mux.Handle(toolGatePath+"/", gateAuth(http.HandlerFunc(gate.HandleStatus))) log.Printf("[tool-gate] enabled — %d tool(s) allowlisted, everything else denied", len(cfg.ToolGate.Tools)) } + mux.HandleFunc("/hooks/status/", func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + w.WriteHeader(http.StatusMethodNotAllowed) + return + } + const prefix = "Bearer " + got := r.Header.Get("Authorization") + if !strings.HasPrefix(got, prefix) || + subtle.ConstantTimeCompare([]byte(strings.TrimPrefix(got, prefix)), secret) != 1 { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + id := strings.TrimPrefix(r.URL.Path, "/hooks/status/") + if !validActionID(id) || state == nil { + http.Error(w, "admission not found", http.StatusNotFound) + return + } + a, err := state.WebhookAdmission(id) + if err != nil { + http.Error(w, "admission not found", http.StatusNotFound) + return + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "admission_id": a.ID, "pipeline": a.Pipeline, "body_hash": a.BodyHash, + "status": a.Status, "error": a.Error, + "created_at": a.CreatedAt.UTC().Format(time.RFC3339), + "updated_at": a.UpdatedAt.UTC().Format(time.RFC3339), + }) + }) + mux.HandleFunc("/hooks/", func(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) @@ -3353,8 +3431,24 @@ func newWebhookHandler(cfg *config.Config, sched *Scheduler, budget *BudgetTrack return } + admissionID := "wh_" + strings.TrimPrefix(newToolTicketID(), "tc_") + bodySum := sha256.Sum256(body) + bodyHash := "sha256:" + hex.EncodeToString(bodySum[:]) + now := time.Now() + if state != nil { + if err := state.BeginWebhookAdmission(statestore.WebhookAdmission{ + ID: admissionID, Pipeline: name, BodyHash: bodyHash, CreatedAt: now, UpdatedAt: now, + }); err != nil { + sched.SetRunning(name, false) + log.Printf("[webhook] durable admission failed: %v", err) + http.Error(w, "state unavailable", http.StatusServiceUnavailable) + return + } + _ = state.FinishWebhookAdmission(admissionID, "running", "", now) + } + log.Printf("[webhook] triggering pipeline %s (%d body bytes)", name, len(body)) - go func(p config.PipelineConfig, body []byte) { + go func(p config.PipelineConfig, body []byte, admissionID string) { defer sched.SetRunning(p.Name, false) seed := map[string]interface{}{"webhook_body": string(body)} if strings.TrimSpace(string(body)) != "" { @@ -3363,12 +3457,22 @@ func newWebhookHandler(cfg *config.Config, sched *Scheduler, budget *BudgetTrack if err := runPipeline(cfg, p, budget, gateChannel(bot), skills, seed); err != nil { log.Printf("[webhook] pipeline %s error: %v", p.Name, err) bot.Send(fmt.Sprintf("[draftcat] ERROR in %s (webhook): %s", p.Name, err)) + if state != nil { + _ = state.FinishWebhookAdmission(admissionID, "error", err.Error(), time.Now()) + } + } else if state != nil { + _ = state.FinishWebhookAdmission(admissionID, "completed", "", time.Now()) } sched.MarkRun(p.Name) - }(p, body) + }(p, body, admissionID) + w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusAccepted) - _, _ = w.Write([]byte("accepted\n")) + _ = json.NewEncoder(w).Encode(map[string]string{ + "admission_id": admissionID, + "status": "accepted", + "poll": "/hooks/status/" + admissionID, + }) }) return mux } diff --git a/model_policy.go b/model_policy.go new file mode 100644 index 0000000..3ff5d85 --- /dev/null +++ b/model_policy.go @@ -0,0 +1,88 @@ +package main + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "os" + "strings" + "time" + "unicode/utf8" + + "github.com/renezander030/draftcat/internal/config" +) + +func modelPolicyPreview(text string, max int) string { + if max <= 0 || utf8.RuneCountInString(text) <= max { + return text + } + runes := []rune(text) + return string(runes[:max]) + "..." +} + +func enforceModelPolicy(ctx context.Context, cfg *config.Config, role, phase, text string) error { + rule, err := cfg.ModelPolicy.Match(role, phase, text) + if err != nil { + return err + } + if rule == nil { + return nil + } + action := strings.ToLower(strings.TrimSpace(rule.Action)) + decision := "policy_deny" + var operatorID int64 + expires := time.Now() + if action == "review" { + if opChan == nil { + recordModelPolicyDecision(ctx, cfg, role, phase, text, rule, decision, operatorID, expires) + return fmt.Errorf("model policy %q requires review but no operator channel is running", rule.ID) + } + timeout, _ := time.ParseDuration(cfg.Timeouts.OperatorApproval) + if timeout <= 0 { + timeout = 4 * time.Hour + } + expires = time.Now().Add(timeout) + msg := fmt.Sprintf("[draftcat] Model %s requires review\n\nrule: %s\nrole: %s\nreason: %s\n\n%s", + phase, rule.ID, role, rule.Reason, modelPolicyPreview(text, cfg.ModelPolicy.PreviewLimit())) + reviewCtx, cancel := context.WithTimeout(ctx, timeout) + dec, reviewErr := opChan.SendForApproval(reviewCtx, msg, nil) + cancel() + if reviewErr == nil && dec.Action == "approve" { + decision, operatorID = "approve", dec.ApproverID + recordModelPolicyDecision(ctx, cfg, role, phase, text, rule, decision, operatorID, expires) + return nil + } + if reviewErr != nil { + decision = "timeout" + } else if dec.Action != "" { + decision = dec.Action + } + } + recordModelPolicyDecision(ctx, cfg, role, phase, text, rule, decision, operatorID, expires) + return fmt.Errorf("model %s blocked by policy %q: %s", phase, rule.ID, rule.Reason) +} + +func recordModelPolicyDecision(ctx context.Context, cfg *config.Config, role, phase, text string, + rule *config.ModelPolicyRule, decision string, operatorID int64, expires time.Time) { + if state == nil { + return + } + sum := sha256.Sum256([]byte(text)) + payloadHash := hex.EncodeToString(sum[:]) + step := config.StepConfig{Name: "model-" + phase + ":" + rule.ID, Type: "approval", Role: role, Risk: config.RiskHigh} + policy := rule.ID + ": " + rule.Reason + envelope := newApprovalEnvelope([]byte(os.Getenv("DRAFTCAT_APPROVAL_SECRET")), + runIDFromContext(ctx), pipelineFromContext(ctx), step, time.Now(), expires, + decision, operatorID, payloadHash, 1, boolCount(decision == "approve"), policy) + if err := state.RecordApprovalV2(envelope); err != nil { + return + } +} + +func boolCount(v bool) int { + if v { + return 1 + } + return 0 +} diff --git a/model_policy_test.go b/model_policy_test.go new file mode 100644 index 0000000..007cd19 --- /dev/null +++ b/model_policy_test.go @@ -0,0 +1,52 @@ +package main + +import ( + "context" + "path/filepath" + "testing" + + "github.com/renezander030/draftcat/internal/config" + statestore "github.com/renezander030/draftcat/internal/state" +) + +func TestModelPolicyDenyIsAudited(t *testing.T) { + st, err := statestore.OpenStateStore(filepath.Join(t.TempDir(), "state.db")) + if err != nil { + t.Fatal(err) + } + prevState, prevChannel := state, opChan + state, opChan = st, nil + t.Cleanup(func() { state, opChan = prevState, prevChannel; _ = st.Close() }) + cfg := &config.Config{ + ModelPolicy: config.ModelPolicyConfig{ + Rules: []config.ModelPolicyRule{{ + ID: "secrets", Phase: "input", Pattern: `(?i)password`, Action: "deny", Reason: "secret-like input", + }}, + }, + } + ctx := withRun(context.Background(), "run-1", "pipeline") + if err := enforceModelPolicy(ctx, cfg, "drafter", "input", "password=hidden"); err == nil { + t.Fatal("matching input was not denied") + } + rows, err := st.ApprovalsForRun("run-1") + if err != nil || len(rows) != 1 || rows[0].Version != 2 || rows[0].Decision != "policy_deny" || rows[0].BindingHash == "" { + t.Fatalf("rows=%+v err=%v", rows, err) + } +} + +func TestModelPolicyReviewCanRelease(t *testing.T) { + prev := opChan + opChan = &stubApprovalChannel{action: "approve", id: 7} + t.Cleanup(func() { opChan = prev }) + cfg := &config.Config{ + Timeouts: config.TimeoutConfig{OperatorApproval: "1s"}, + ModelPolicy: config.ModelPolicyConfig{ + Rules: []config.ModelPolicyRule{{ + ID: "review", Phase: "output", Pattern: `guarantee`, Action: "review", Reason: "claim review", + }}, + }, + } + if err := enforceModelPolicy(context.Background(), cfg, "drafter", "output", "we guarantee this"); err != nil { + t.Fatalf("approved review blocked output: %v", err) + } +} diff --git a/receipt_v2.go b/receipt_v2.go new file mode 100644 index 0000000..bab1155 --- /dev/null +++ b/receipt_v2.go @@ -0,0 +1,65 @@ +package main + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "strings" + "time" + + "github.com/renezander030/draftcat/internal/approval" + "github.com/renezander030/draftcat/internal/config" + statestore "github.com/renezander030/draftcat/internal/state" +) + +func hashCanonical(v interface{}) string { + b, _ := json.Marshal(v) + sum := sha256.Sum256(b) + return "sha256:" + hex.EncodeToString(sum[:]) +} + +func newApprovalEnvelope(secret []byte, runID, pipeline string, step config.StepConfig, + decidedAt, expiresAt time.Time, decision string, operatorID int64, payloadHash string, + quorumN, quorumGot int, policy string) statestore.ApprovalEnvelope { + actionID := strings.Trim(strings.Join([]string{runID, pipeline, step.Name}, ":"), ":") + if actionID == "" { + actionID = fmt.Sprintf("action:%d", decidedAt.UnixNano()) + } + policyHash := hashCanonical(struct { + Pipeline string `json:"pipeline"` + Step config.StepConfig `json:"step"` + Policy string `json:"policy"` + }{pipeline, step, policy}) + bindingHash := hashCanonical(struct { + Version int `json:"version"` + ActionID string `json:"action_id"` + PayloadHash string `json:"payload_hash"` + PolicyHash string `json:"policy_hash"` + ExpiresAt int64 `json:"expires_at"` + }{2, actionID, payloadHash, policyHash, expiresAt.Unix()}) + e := statestore.ApprovalEnvelope{ + ReceiptID: "rcpt_" + strings.TrimPrefix(newToolTicketID(), "tc_"), + RunID: runID, ActionID: actionID, Pipeline: pipeline, Step: step.Name, + DecidedAt: decidedAt, Decision: decision, OperatorID: operatorID, + PayloadHash: payloadHash, QuorumN: quorumN, QuorumGot: quorumGot, + Policy: policy, PolicyHash: policyHash, BindingHash: bindingHash, + ExpiresAt: expiresAt, Lifecycle: "decided", + } + if len(secret) == 0 { + return e + } + nonce, err := approval.NewNonce() + if err != nil { + return e + } + e.Nonce = nonce + e.Signature = approval.SignV2(secret, approval.FieldsV2{ + ReceiptID: e.ReceiptID, RunID: e.RunID, ActionID: e.ActionID, + Pipeline: e.Pipeline, Step: e.Step, DecidedAt: e.DecidedAt.Unix(), + Decision: e.Decision, OperatorID: e.OperatorID, PayloadHash: e.PayloadHash, + Policy: e.Policy, PolicyHash: e.PolicyHash, BindingHash: e.BindingHash, + ExpiresAt: e.ExpiresAt.Unix(), QuorumN: e.QuorumN, QuorumGot: e.QuorumGot, + }, nonce) + return e +} diff --git a/receipts_cmd.go b/receipts_cmd.go new file mode 100644 index 0000000..62afd98 --- /dev/null +++ b/receipts_cmd.go @@ -0,0 +1,261 @@ +package main + +import ( + "bufio" + "encoding/json" + "fmt" + "io" + "os" + "strconv" + "strings" + "time" + + statestore "github.com/renezander030/draftcat/internal/state" +) + +type receiptJSON struct { + Version int `json:"version"` + ReceiptID string `json:"receipt_id"` + RunID string `json:"run_id,omitempty"` + ActionID string `json:"action_id,omitempty"` + Pipeline string `json:"pipeline"` + Step string `json:"step"` + DecidedAt string `json:"decided_at"` + Decision string `json:"decision"` + OperatorID int64 `json:"operator_id"` + PayloadHash string `json:"payload_hash"` + Policy string `json:"policy,omitempty"` + PolicyHash string `json:"policy_hash,omitempty"` + BindingHash string `json:"binding_hash,omitempty"` + ExpiresAt string `json:"expires_at,omitempty"` + Lifecycle string `json:"lifecycle"` + QuorumN int `json:"quorum_n"` + QuorumGot int `json:"quorum_got"` + Nonce string `json:"nonce,omitempty"` + Signature string `json:"signature,omitempty"` + Verification string `json:"verification"` +} + +func receiptID(r statestore.ApprovalRecord) string { + if r.ReceiptID != "" { + return r.ReceiptID + } + return strconv.FormatInt(r.ID, 10) +} + +func receiptView(r statestore.ApprovalRecord, secret []byte) receiptJSON { + expires := "" + if !r.ExpiresAt.IsZero() { + expires = r.ExpiresAt.UTC().Format(time.RFC3339) + } + verification := statestore.VerifyApprovalRecord(secret, r) + if r.Signature != "" && len(secret) == 0 { + verification = "unverified" + } + return receiptJSON{ + Version: r.Version, ReceiptID: receiptID(r), RunID: r.RunID, ActionID: r.ActionID, + Pipeline: r.Pipeline, Step: r.Step, DecidedAt: r.DecidedAt.UTC().Format(time.RFC3339), + Decision: r.Decision, OperatorID: r.OperatorID, PayloadHash: r.PayloadHash, + Policy: r.Policy, PolicyHash: r.PolicyHash, BindingHash: r.BindingHash, + ExpiresAt: expires, Lifecycle: r.Lifecycle, QuorumN: r.QuorumN, QuorumGot: r.QuorumGot, + Nonce: r.Nonce, Signature: r.Signature, + Verification: verification, + } +} + +func runReceiptsCmd(args []string) int { + if len(args) == 0 { + fmt.Fprintln(os.Stderr, "Usage: draftcat receipts [options]") + return 2 + } + switch args[0] { + case "list": + return runReceiptsList(args[1:]) + case "show": + return runReceiptsShow(args[1:]) + case "export": + return runReceiptsExport(args[1:]) + case "-h", "--help", "help": + fmt.Println("Usage: draftcat receipts [options]") + fmt.Println(" list [--pipeline name] [--limit N] [--json] [--config path]") + fmt.Println(" show [--config path]") + fmt.Println(" export [--pipeline name] [--limit N] [--out path] [--config path]") + return 0 + default: + fmt.Fprintf(os.Stderr, "receipts: unknown command %q\n", args[0]) + return 2 + } +} + +type receiptOptions struct { + pipeline string + config string + out string + limit int + json bool + receiptID string +} + +func parseReceiptOptions(args []string, allowID bool) (receiptOptions, int) { + o := receiptOptions{config: "config.yaml", limit: 100} + for i := 0; i < len(args); i++ { + switch args[i] { + case "--pipeline": + if i+1 >= len(args) { + fmt.Fprintln(os.Stderr, "receipts: --pipeline requires a name") + return o, 2 + } + o.pipeline = args[i+1] + i++ + case "--config", "-config": + if i+1 >= len(args) { + fmt.Fprintln(os.Stderr, "receipts: --config requires a path") + return o, 2 + } + o.config = args[i+1] + i++ + case "--out": + if i+1 >= len(args) { + fmt.Fprintln(os.Stderr, "receipts: --out requires a path") + return o, 2 + } + o.out = args[i+1] + i++ + case "--limit": + if i+1 >= len(args) { + fmt.Fprintln(os.Stderr, "receipts: --limit requires a number") + return o, 2 + } + n, err := strconv.Atoi(args[i+1]) + if err != nil || n <= 0 { + fmt.Fprintln(os.Stderr, "receipts: --limit must be positive") + return o, 2 + } + o.limit = n + i++ + case "--json": + o.json = true + default: + if allowID && !strings.HasPrefix(args[i], "-") && o.receiptID == "" { + o.receiptID = args[i] + continue + } + fmt.Fprintf(os.Stderr, "receipts: unknown option %q\n", args[i]) + return o, 2 + } + } + return o, 0 +} + +func loadReceipts(o receiptOptions) (*statestore.StateStore, func(), []statestore.ApprovalRecord, int) { + st, closeStore, code := openStateForCmd(o.config) + if code != 0 { + return nil, closeStore, nil, code + } + var rows []statestore.ApprovalRecord + var err error + if o.pipeline == "" { + rows, err = st.AllApprovals(o.limit) + } else { + rows, err = st.ApprovalsForPipeline(o.pipeline, o.limit) + } + if err != nil { + closeStore() + fmt.Fprintf(os.Stderr, "receipts: read: %v\n", err) + return nil, func() {}, nil, 1 + } + return st, closeStore, rows, 0 +} + +func runReceiptsList(args []string) int { + o, code := parseReceiptOptions(args, false) + if code != 0 { + return code + } + _, closeStore, rows, code := loadReceipts(o) + if code != 0 { + return code + } + defer closeStore() + secret := []byte(os.Getenv("DRAFTCAT_APPROVAL_SECRET")) + views := make([]receiptJSON, 0, len(rows)) + for _, r := range rows { + views = append(views, receiptView(r, secret)) + } + if o.json { + b, _ := json.MarshalIndent(views, "", " ") + fmt.Println(string(b)) + return 0 + } + if len(views) == 0 { + fmt.Println("No receipts recorded yet.") + return 0 + } + for _, r := range views { + fmt.Printf("%s %-22s %-20s %-15s %s\n", r.DecidedAt, r.Pipeline, r.Step, r.Decision, r.ReceiptID) + } + return 0 +} + +func runReceiptsShow(args []string) int { + o, code := parseReceiptOptions(args, true) + if code != 0 { + return code + } + if o.receiptID == "" { + fmt.Fprintln(os.Stderr, "receipts show: receipt id required") + return 2 + } + st, closeStore, code := openStateForCmd(o.config) + if code != 0 { + return code + } + defer closeStore() + r, err := st.ApprovalByReceiptID(o.receiptID) + if err != nil { + fmt.Fprintf(os.Stderr, "receipts show: %v\n", err) + return 1 + } + b, _ := json.MarshalIndent(receiptView(r, []byte(os.Getenv("DRAFTCAT_APPROVAL_SECRET"))), "", " ") + fmt.Println(string(b)) + return 0 +} + +func runReceiptsExport(args []string) int { + o, code := parseReceiptOptions(args, false) + if code != 0 { + return code + } + _, closeStore, rows, code := loadReceipts(o) + if code != 0 { + return code + } + defer closeStore() + var dst io.Writer = os.Stdout + var f *os.File + if o.out != "" { + var err error + f, err = os.OpenFile(o.out, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o600) + if err != nil { + fmt.Fprintf(os.Stderr, "receipts export: %v\n", err) + return 1 + } + defer func() { _ = f.Close() }() + dst = f + } + w := bufio.NewWriter(dst) + secret := []byte(os.Getenv("DRAFTCAT_APPROVAL_SECRET")) + for i := len(rows) - 1; i >= 0; i-- { + b, err := json.Marshal(receiptView(rows[i], secret)) + if err != nil { + fmt.Fprintf(os.Stderr, "receipts export: row %d: %v\n", rows[i].ID, err) + continue + } + _, _ = w.Write(append(b, '\n')) + } + if err := w.Flush(); err != nil { + fmt.Fprintf(os.Stderr, "receipts export: %v\n", err) + return 1 + } + return 0 +} diff --git a/receipts_cmd_test.go b/receipts_cmd_test.go new file mode 100644 index 0000000..57d78fe --- /dev/null +++ b/receipts_cmd_test.go @@ -0,0 +1,37 @@ +package main + +import ( + "path/filepath" + "testing" + "time" + + "github.com/renezander030/draftcat/internal/config" + statestore "github.com/renezander030/draftcat/internal/state" +) + +func TestReceiptV2RoundTripAndView(t *testing.T) { + st, err := statestore.OpenStateStore(filepath.Join(t.TempDir(), "state.db")) + if err != nil { + t.Fatal(err) + } + defer func() { _ = st.Close() }() + secret := []byte("receipt-secret") + decided := time.Unix(1_800_000_000, 0) + e := newApprovalEnvelope(secret, "run-1", "pipeline", config.StepConfig{Name: "send", Risk: "high"}, + decided, decided.Add(time.Hour), "approve", 42, "sha256:payload", 1, 1, "human-approval") + if err := st.RecordApprovalV2(e); err != nil { + t.Fatal(err) + } + rows, err := st.AllApprovals(10) + if err != nil || len(rows) != 1 { + t.Fatalf("rows=%+v err=%v", rows, err) + } + v := receiptView(rows[0], secret) + if v.Version != 2 || v.ReceiptID == "" || v.ActionID == "" || v.BindingHash == "" || v.Verification != "ok" { + t.Fatalf("receipt=%+v", v) + } + got, err := st.ApprovalByReceiptID(v.ReceiptID) + if err != nil || got.ReceiptID != v.ReceiptID { + t.Fatalf("show=%+v err=%v", got, err) + } +} diff --git a/tool_gate.go b/tool_gate.go index 64b713c..5c8d110 100644 --- a/tool_gate.go +++ b/tool_gate.go @@ -39,23 +39,32 @@ import ( "context" "crypto/rand" "crypto/sha256" + "database/sql" "encoding/hex" "encoding/json" + "errors" "fmt" "io" "log" "net/http" + "os" "strings" "sync" "time" + "github.com/renezander030/draftcat/internal/approval" "github.com/renezander030/draftcat/internal/config" "github.com/renezander030/draftcat/internal/obs" + statestore "github.com/renezander030/draftcat/internal/state" ) // ToolCallRequest is what a harness sends to ask permission. type ToolCallRequest struct { - Tool string `json:"tool"` + // ActionID is the caller's stable idempotency identity for the intended + // side effect. Reusing it with the same binding returns the existing state; + // reusing it with changed inputs is rejected. + ActionID string `json:"action_id"` + Tool string `json:"tool"` // Args are the exact arguments the harness intends to call with. They are // hashed into the approval record, never stored raw. Args map[string]interface{} `json:"args"` @@ -72,14 +81,27 @@ type ToolCallRequest struct { // gate answers 202 pending with the approval_id instead of holding on, and // the approval keeps running server-side. Ignored in async mode. Wait string `json:"wait"` + // ExpiresAt optionally narrows the permit validity window. The gate never + // extends it beyond operator_approval. + ExpiresAt string `json:"expires_at"` + + policyHash string + bindingHash string + expires time.Time + providedActionID bool } // ToolCallResponse is the gate's answer. type ToolCallResponse struct { - Decision string `json:"decision"` // "allow" | "deny" | "pending" - Reason string `json:"reason"` - ArgsHash string `json:"args_hash"` - DecidedBy string `json:"decided_by,omitempty"` // "allowlist" | "policy" | "operator" | "repeat-guard" + ActionID string `json:"action_id,omitempty"` + Decision string `json:"decision"` // "allow" | "deny" | "pending" + State string `json:"state,omitempty"` // pending | allowed | denied | expired | consumed + Permit string `json:"permit,omitempty"` // "execute" only on the first successful consume + Reason string `json:"reason"` + ArgsHash string `json:"args_hash"` + PolicyHash string `json:"policy_hash,omitempty"` + BindingHash string `json:"binding_hash,omitempty"` + DecidedBy string `json:"decided_by,omitempty"` // "allowlist" | "policy" | "operator" | "repeat-guard" // Rule names the rule or condition behind a decision the gate made on its // own, so a denial can be traced to config without reading the audit log. Rule string `json:"rule,omitempty"` @@ -88,6 +110,7 @@ type ToolCallResponse struct { // can still collect the decision. ApprovalID string `json:"approval_id,omitempty"` Poll string `json:"poll,omitempty"` + Consume string `json:"consume,omitempty"` ExpiresAt string `json:"expires_at,omitempty"` } @@ -103,19 +126,22 @@ const toolTicketRetention = time.Hour // toolTicket is one human-path decision in flight or recently made. type toolTicket struct { - ID string - Tool string - Agent string - RunID string - ArgsHash string - Created time.Time - Expires time.Time + ID string + Tool string + Agent string + RunID string + ArgsHash string + PolicyHash string + BindingHash string + Created time.Time + Expires time.Time - done chan struct{} - mu sync.Mutex - decided bool - decidedAt time.Time - resp ToolCallResponse + done chan struct{} + mu sync.Mutex + decided bool + decidedAt time.Time + consumedAt time.Time + resp ToolCallResponse } func (t *toolTicket) result() (ToolCallResponse, bool) { @@ -136,15 +162,36 @@ func (t *toolTicket) resolve(resp ToolCallResponse, at time.Time) { close(t.done) } +func (t *toolTicket) consume(binding string, at time.Time) (ToolCallResponse, bool) { + t.mu.Lock() + defer t.mu.Unlock() + if binding == "" || binding != t.BindingHash || !t.decided || t.resp.State != "allowed" || at.After(t.Expires) { + return t.resp, false + } + t.consumedAt = at + t.resp.State = "consumed" + t.resp.Permit = "" + t.resp.Reason = "permit already consumed" + out := t.resp + out.Permit = "execute" + out.Reason = "permit consumed; execute this bound action once" + return out, true +} + // pendingResponse is what a caller sees while the human has not decided. func (t *toolTicket) pendingResponse() ToolCallResponse { return ToolCallResponse{ - Decision: "pending", - Reason: "awaiting operator decision", - ArgsHash: t.ArgsHash, - ApprovalID: t.ID, - Poll: toolGatePath + "/" + t.ID, - ExpiresAt: t.Expires.UTC().Format(time.RFC3339), + ActionID: t.ID, + Decision: "pending", + State: "pending", + Reason: "awaiting operator decision", + ArgsHash: t.ArgsHash, + ApprovalID: t.ID, + Poll: toolGatePath + "/" + t.ID, + Consume: toolGatePath + "/" + t.ID + "/consume", + PolicyHash: t.PolicyHash, + BindingHash: t.BindingHash, + ExpiresAt: t.Expires.UTC().Format(time.RFC3339), } } @@ -172,7 +219,7 @@ type toolGate struct { } func newToolGate(cfg *config.Config, ch OperatorChannel) *toolGate { - return &toolGate{ + g := &toolGate{ cfg: cfg, ch: ch, now: time.Now, @@ -180,6 +227,12 @@ func newToolGate(cfg *config.Config, ch OperatorChannel) *toolGate { recent: map[string]*repeatEntry{}, notified: map[string]time.Time{}, } + if state != nil { + if err := state.ExpireToolActions(time.Now()); err != nil { + log.Printf("[tool-gate] reconcile old permits: %v", err) + } + } + return g } // handleToolCall decides one tool call. Kept as the one-line constructor the @@ -224,6 +277,16 @@ func (g *toolGate) HandleCall(w http.ResponseWriter, r *http.Request) { }) return } + req.providedActionID = strings.TrimSpace(req.ActionID) != "" + if !req.providedActionID { + // Compatibility for v0.6 clients. Stable retries require callers to + // persist and resend the returned action_id. + req.ActionID = newToolTicketID() + } + if !validActionID(req.ActionID) { + http.Error(w, "action_id must be 1-128 URL-safe characters", http.StatusBadRequest) + return + } mode := strings.ToLower(strings.TrimSpace(req.Mode)) if mode != "" && mode != "sync" && mode != "async" { http.Error(w, `mode must be "sync" or "async"`, http.StatusBadRequest) @@ -242,6 +305,36 @@ func (g *toolGate) HandleCall(w http.ResponseWriter, r *http.Request) { argsHash := hashToolArgs(req.Args) rule, listed := g.cfg.ToolGate.Lookup(req.Tool) + policyHash := hashToolPolicy(req.Tool, rule, listed) + expires := g.now().Add(g.approvalWindow()) + if strings.TrimSpace(req.ExpiresAt) != "" { + requested, err := time.Parse(time.RFC3339, req.ExpiresAt) + if err != nil || !requested.After(g.now()) { + http.Error(w, "expires_at must be a future RFC3339 timestamp", http.StatusBadRequest) + return + } + if requested.Before(expires) { + expires = requested + } + } else { + g.mu.Lock() + existing := g.tickets[req.ActionID] + g.mu.Unlock() + if existing != nil { + expires = existing.Expires + } else if state != nil { + if a, err := state.ToolAction(req.ActionID); err == nil { + expires = a.ExpiresAt + } + } + } + bindingHash := hashToolBinding(req, argsHash, policyHash, expires) + req.policyHash, req.bindingHash, req.expires = policyHash, bindingHash, expires + replay, code, handled := g.reserveAction(req, argsHash, policyHash, bindingHash, expires) + if handled { + writeToolDecision(w, code, replay) + return + } // Default deny. An unlisted tool is refused whatever else is true. if !listed { @@ -249,10 +342,12 @@ func (g *toolGate) HandleCall(w http.ResponseWriter, r *http.Request) { recordToolDecision(g.cfg, req, argsHash, "deny", 0, "unlisted") reason := fmt.Sprintf("tool %q is not in tool_gate.tools — the gate denies by default", req.Tool) g.notifyDenial(req, argsHash, "unlisted", reason) - writeToolDecision(w, http.StatusOK, ToolCallResponse{ + resp := g.finishAction(req, expires, ToolCallResponse{ Decision: "deny", Reason: reason, ArgsHash: argsHash, DecidedBy: "allowlist", Rule: "not listed", + PolicyHash: policyHash, BindingHash: bindingHash, }) + writeToolDecision(w, http.StatusOK, resp) return } @@ -268,10 +363,12 @@ func (g *toolGate) HandleCall(w http.ResponseWriter, r *http.Request) { log.Printf("[tool-gate] DENY %s (args mismatch: %s) agent=%q", req.Tool, why, req.Agent) recordToolDecision(g.cfg, req, argsHash, "deny", 0, "args mismatch: "+why) g.notifyDenial(req, argsHash, "mismatch", reason) - writeToolDecision(w, http.StatusOK, ToolCallResponse{ + resp := g.finishAction(req, expires, ToolCallResponse{ Decision: "deny", Reason: reason, ArgsHash: argsHash, DecidedBy: "policy", Rule: "args." + why, + PolicyHash: policyHash, BindingHash: bindingHash, }) + writeToolDecision(w, http.StatusOK, resp) return } needHuman = true @@ -283,19 +380,22 @@ func (g *toolGate) HandleCall(w http.ResponseWriter, r *http.Request) { if !needHuman { log.Printf("[tool-gate] ALLOW %s (allowlisted, risk=%s) agent=%q", req.Tool, rule.RiskOf(), req.Agent) recordToolDecision(g.cfg, req, argsHash, "policy_approve", 0, "allowlisted risk="+rule.RiskOf()) - writeToolDecision(w, http.StatusOK, ToolCallResponse{ + resp := g.finishAction(req, expires, ToolCallResponse{ Decision: "allow", Reason: "allowlisted in tool_gate", ArgsHash: argsHash, DecidedBy: "allowlist", Rule: "listed", + PolicyHash: policyHash, BindingHash: bindingHash, }) + writeToolDecision(w, http.StatusOK, resp) return } // Needs a human. if g.ch == nil { - writeToolDecision(w, http.StatusOK, ToolCallResponse{ + resp := g.finishAction(req, expires, ToolCallResponse{ Decision: "deny", Reason: "tool requires approval but no operator channel is running", - ArgsHash: argsHash, + ArgsHash: argsHash, PolicyHash: policyHash, BindingHash: bindingHash, }) + writeToolDecision(w, http.StatusOK, resp) return } @@ -303,14 +403,32 @@ func (g *toolGate) HandleCall(w http.ResponseWriter, r *http.Request) { tk, verdict, ok := g.repeatCheck(req, rule, argsHash) if ok { // A remembered decision — no prompt. - writeToolDecision(w, http.StatusOK, verdict) + verdict.PolicyHash, verdict.BindingHash = policyHash, bindingHash + writeToolDecision(w, http.StatusOK, g.finishAction(req, expires, verdict)) + return + } + if tk != nil && tk.ID != req.ActionID { + if !req.providedActionID { + if state != nil { + _ = state.DecideToolAction(req.ActionID, "denied", "deny", "joined identical pending action", "repeat-guard", g.now()) + } + writeToolDecision(w, http.StatusAccepted, tk.pendingResponse()) + return + } + reason := "an identical action is already pending under action_id " + tk.ID + resp := g.finishAction(req, expires, ToolCallResponse{ + Decision: "deny", Reason: reason, ArgsHash: argsHash, + PolicyHash: policyHash, BindingHash: bindingHash, + DecidedBy: "repeat-guard", Rule: "pending_duplicate", + }) + writeToolDecision(w, http.StatusConflict, resp) return } if tk == nil { // Fresh ask. The approval context is the request's own in the plain // sync case (a harness that hangs up cancels the gate, as before) and // detached whenever the caller may legitimately come back later. - tk = g.newTicket(req, argsHash) + tk = g.newTicket(req, argsHash, policyHash, bindingHash, expires) parent := r.Context() if async || wait > 0 { parent = context.Background() @@ -343,8 +461,9 @@ func (g *toolGate) HandleCall(w http.ResponseWriter, r *http.Request) { } resp, decided := tk.result() if !decided { - resp = ToolCallResponse{Decision: "deny", Reason: "approval timed out — the gate denies rather than assumes yes", - ArgsHash: argsHash, ApprovalID: tk.ID} + resp = ToolCallResponse{ActionID: tk.ID, Decision: "deny", State: "denied", + Reason: "approval timed out - the gate denies rather than assumes yes", + ArgsHash: argsHash, PolicyHash: tk.PolicyHash, BindingHash: tk.BindingHash, ApprovalID: tk.ID} } writeToolDecision(w, http.StatusOK, resp) case <-r.Context().Done(): @@ -353,26 +472,41 @@ func (g *toolGate) HandleCall(w http.ResponseWriter, r *http.Request) { } } -// HandleStatus is GET /gate/tool-call/[?wait=30s]. +// HandleStatus serves GET /gate/tool-call/ and the atomic +// POST /gate/tool-call//consume transition. func (g *toolGate) HandleStatus(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodGet { - w.WriteHeader(http.StatusMethodNotAllowed) + rest := strings.Trim(strings.TrimPrefix(r.URL.Path, toolGatePath), "/") + parts := strings.Split(rest, "/") + if rest == "" || len(parts) > 2 || !validActionID(parts[0]) { + http.Error(w, "approval id required", http.StatusNotFound) return } - id := strings.Trim(strings.TrimPrefix(r.URL.Path, toolGatePath), "/") - if id == "" || strings.Contains(id, "/") { - http.Error(w, "approval id required", http.StatusNotFound) + id := parts[0] + if len(parts) == 2 { + if parts[1] != "consume" || r.Method != http.MethodPost { + w.WriteHeader(http.StatusMethodNotAllowed) + return + } + g.handleConsume(w, r, id) + return + } + if r.Method != http.MethodGet { + w.WriteHeader(http.StatusMethodNotAllowed) return } g.mu.Lock() tk := g.tickets[id] g.mu.Unlock() if tk == nil { - // Unknown here means the gate restarted or the id is stale. Either - // way there is no decision to hand over, and the safe answer is no. + if state != nil { + if a, err := state.ToolAction(id); err == nil { + writeToolDecision(w, http.StatusOK, responseFromToolAction(a)) + return + } + } writeToolDecision(w, http.StatusNotFound, ToolCallResponse{ - Decision: "deny", Reason: "unknown or expired approval id — the gate has no decision for it; ask again", - ApprovalID: id, + ActionID: id, Decision: "deny", State: "denied", + Reason: "unknown action id - the gate has no decision for it; ask again", ApprovalID: id, }) return } @@ -393,19 +527,75 @@ func (g *toolGate) HandleStatus(w http.ResponseWriter, r *http.Request) { } } if resp, decided := tk.result(); decided { + if resp.State == "allowed" && g.now().After(tk.Expires) { + resp.Decision, resp.State, resp.Consume, resp.Permit = "deny", "expired", "", "" + resp.Reason = "permit expired before consumption" + } writeToolDecision(w, http.StatusOK, resp) return } writeToolDecision(w, http.StatusOK, tk.pendingResponse()) } +func (g *toolGate) handleConsume(w http.ResponseWriter, r *http.Request, id string) { + var body struct { + BindingHash string `json:"binding_hash"` + } + if err := json.NewDecoder(io.LimitReader(r.Body, 4096)).Decode(&body); err != nil { + http.Error(w, "malformed json", http.StatusBadRequest) + return + } + now := g.now() + if state != nil { + a, consumed, err := state.ConsumeToolAction(id, body.BindingHash, now) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + writeToolDecision(w, http.StatusNotFound, ToolCallResponse{ActionID: id, Decision: "deny", State: "denied", Reason: "unknown action id"}) + } else { + http.Error(w, "state unavailable", http.StatusServiceUnavailable) + } + return + } + if consumed { + recordToolConsumption(a, now) + resp := responseFromToolAction(a) + resp.Decision, resp.Permit = "allow", "execute" + resp.Reason = "permit consumed; execute this bound action once" + g.mu.Lock() + if tk := g.tickets[id]; tk != nil { + _, _ = tk.consume(body.BindingHash, now) + } + g.mu.Unlock() + writeToolDecision(w, http.StatusOK, resp) + return + } + writeToolDecision(w, http.StatusConflict, responseFromToolAction(a)) + return + } + g.mu.Lock() + tk := g.tickets[id] + g.mu.Unlock() + if tk == nil { + writeToolDecision(w, http.StatusNotFound, ToolCallResponse{ActionID: id, Decision: "deny", State: "denied", Reason: "unknown action id"}) + return + } + resp, ok := tk.consume(body.BindingHash, now) + if !ok { + resp.Permit = "" + writeToolDecision(w, http.StatusConflict, resp) + return + } + writeToolDecision(w, http.StatusOK, resp) +} + // newTicket registers a fresh human-path decision and remembers the call for // the repeat guard. Also the moment old tickets are swept. -func (g *toolGate) newTicket(req ToolCallRequest, argsHash string) *toolTicket { +func (g *toolGate) newTicket(req ToolCallRequest, argsHash, policyHash, bindingHash string, expires time.Time) *toolTicket { now := g.now() tk := &toolTicket{ - ID: newToolTicketID(), Tool: req.Tool, Agent: req.Agent, RunID: req.RunID, ArgsHash: argsHash, - Created: now, Expires: now.Add(g.approvalWindow()), done: make(chan struct{}), + ID: req.ActionID, Tool: req.Tool, Agent: req.Agent, RunID: req.RunID, ArgsHash: argsHash, + PolicyHash: policyHash, BindingHash: bindingHash, + Created: now, Expires: expires, done: make(chan struct{}), } g.mu.Lock() defer g.mu.Unlock() @@ -553,16 +743,29 @@ func (g *toolGate) askHuman(parent context.Context, tk *toolTicket, req ToolCall log.Printf("[tool-gate] DENY %s (%s) agent=%q", req.Tool, reason, req.Agent) recordToolDecision(g.cfg, req, tk.ArgsHash, "deny", dec.ApproverID, reason) tk.resolve(ToolCallResponse{ - Decision: "deny", Reason: reason, ArgsHash: tk.ArgsHash, DecidedBy: "operator", ApprovalID: tk.ID, + ActionID: tk.ID, Decision: "deny", State: "denied", Reason: reason, + ArgsHash: tk.ArgsHash, PolicyHash: tk.PolicyHash, BindingHash: tk.BindingHash, + DecidedBy: "operator", ApprovalID: tk.ID, }, now) + if state != nil { + _ = state.DecideToolAction(tk.ID, "denied", "deny", reason, "operator", now) + } return } log.Printf("[tool-gate] ALLOW %s (operator %d) agent=%q", req.Tool, dec.ApproverID, req.Agent) recordToolDecision(g.cfg, req, tk.ArgsHash, "approve", dec.ApproverID, "operator approved") tk.resolve(ToolCallResponse{ - Decision: "allow", Reason: "operator approved", ArgsHash: tk.ArgsHash, DecidedBy: "operator", ApprovalID: tk.ID, + ActionID: tk.ID, Decision: "allow", State: "allowed", + Reason: "operator approved; consume the permit before executing", + ArgsHash: tk.ArgsHash, PolicyHash: tk.PolicyHash, BindingHash: tk.BindingHash, + DecidedBy: "operator", ApprovalID: tk.ID, + Consume: toolGatePath + "/" + tk.ID + "/consume", + ExpiresAt: tk.Expires.UTC().Format(time.RFC3339), }, now) + if state != nil { + _ = state.DecideToolAction(tk.ID, "allowed", "allow", "operator approved", "operator", now) + } } // notifyDenial tells the operator about a refusal the gate made on its own. @@ -621,18 +824,221 @@ func hashToolArgs(args map[string]interface{}) string { return "sha256:" + hex.EncodeToString(sum[:]) } +func validActionID(id string) bool { + if len(id) == 0 || len(id) > 128 { + return false + } + for _, r := range id { + if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || + (r >= '0' && r <= '9') || r == '-' || r == '_' || r == '.' || r == ':' { + continue + } + return false + } + return true +} + +func hashToolPolicy(tool string, rule config.ToolRule, listed bool) string { + b, _ := json.Marshal(struct { + Tool string `json:"tool"` + Listed bool `json:"listed"` + Rule config.ToolRule `json:"rule"` + }{Tool: tool, Listed: listed, Rule: rule}) + sum := sha256.Sum256(b) + return "sha256:" + hex.EncodeToString(sum[:]) +} + +func hashToolBinding(req ToolCallRequest, argsHash, policyHash string, expires time.Time) string { + b, _ := json.Marshal(struct { + Version int `json:"version"` + ActionID string `json:"action_id"` + Agent string `json:"agent"` + Tool string `json:"tool"` + RunID string `json:"run_id"` + ArgsHash string `json:"args_hash"` + PolicyHash string `json:"policy_hash"` + ExpiresAt int64 `json:"expires_at"` + }{2, req.ActionID, req.Agent, req.Tool, req.RunID, argsHash, policyHash, expires.Unix()}) + sum := sha256.Sum256(b) + return "sha256:" + hex.EncodeToString(sum[:]) +} + +func responseFromToolAction(a statestore.ToolAction) ToolCallResponse { + decision := a.Decision + if decision == "" { + decision = "pending" + } + resp := ToolCallResponse{ + ActionID: a.ActionID, Decision: decision, State: a.Status, Reason: a.Reason, + ArgsHash: a.ArgsHash, PolicyHash: a.PolicyHash, BindingHash: a.BindingHash, + DecidedBy: a.DecidedBy, ApprovalID: a.ActionID, + Poll: toolGatePath + "/" + a.ActionID, + ExpiresAt: a.ExpiresAt.UTC().Format(time.RFC3339), + } + if a.Status == "pending" && resp.Reason == "" { + resp.Reason = "awaiting operator decision" + } + if a.Status == "allowed" { + if time.Now().After(a.ExpiresAt) { + resp.Decision, resp.State = "deny", "expired" + resp.Reason = "permit expired before consumption" + } else { + resp.Consume = toolGatePath + "/" + a.ActionID + "/consume" + if resp.Reason == "" { + resp.Reason = "decision allows this action; consume the permit before executing" + } + } + } + if a.Status == "consumed" { + resp.Permit = "" + resp.Reason = "permit already consumed" + } + if a.Status == "expired" { + resp.Decision = "deny" + } + return resp +} + +// reserveAction is the idempotency gate. The first request reserves its action +// id; matching retries return the existing state and drift is rejected. +func (g *toolGate) reserveAction(req ToolCallRequest, argsHash, policyHash, bindingHash string, expires time.Time) (ToolCallResponse, int, bool) { + g.mu.Lock() + tk := g.tickets[req.ActionID] + g.mu.Unlock() + if tk != nil { + if tk.BindingHash != bindingHash { + return ToolCallResponse{ActionID: req.ActionID, Decision: "deny", State: "denied", + Reason: "action_id is already bound to different action data"}, http.StatusConflict, true + } + if resp, decided := tk.result(); decided { + if resp.State == "allowed" && g.now().After(tk.Expires) { + resp.Decision, resp.State, resp.Consume, resp.Permit = "deny", "expired", "", "" + resp.Reason = "permit expired before consumption" + } + return resp, http.StatusOK, true + } + return tk.pendingResponse(), http.StatusAccepted, true + } + if state == nil { + return ToolCallResponse{}, 0, false + } + a, created, err := state.ReserveToolAction(statestore.ToolAction{ + ActionID: req.ActionID, Tool: req.Tool, Agent: req.Agent, RunID: req.RunID, + ArgsHash: argsHash, PolicyHash: policyHash, BindingHash: bindingHash, + CreatedAt: g.now(), UpdatedAt: g.now(), ExpiresAt: expires, + }) + if err != nil { + return ToolCallResponse{ActionID: req.ActionID, Decision: "deny", State: "denied", Reason: err.Error()}, http.StatusConflict, true + } + if !created { + code := http.StatusOK + if a.Status == "pending" { + code = http.StatusAccepted + } + return responseFromToolAction(a), code, true + } + return ToolCallResponse{}, 0, false +} + +func (g *toolGate) finishAction(req ToolCallRequest, expires time.Time, resp ToolCallResponse) ToolCallResponse { + now := g.now() + resp.ActionID = req.ActionID + resp.ApprovalID = req.ActionID + resp.ExpiresAt = expires.UTC().Format(time.RFC3339) + resp.Poll = toolGatePath + "/" + req.ActionID + status := "denied" + if resp.Decision == "allow" { + status = "allowed" + resp.Consume = toolGatePath + "/" + req.ActionID + "/consume" + resp.Reason += "; consume the permit before executing" + } + resp.State = status + tk := &toolTicket{ + ID: req.ActionID, Tool: req.Tool, Agent: req.Agent, RunID: req.RunID, + ArgsHash: resp.ArgsHash, PolicyHash: resp.PolicyHash, BindingHash: resp.BindingHash, + Created: now, Expires: expires, done: make(chan struct{}), + } + tk.resolve(resp, now) + g.mu.Lock() + g.tickets[tk.ID] = tk + g.mu.Unlock() + if state != nil { + _ = state.DecideToolAction(req.ActionID, status, resp.Decision, resp.Reason, resp.DecidedBy, now) + } + return resp +} + func recordToolDecision(cfg *config.Config, req ToolCallRequest, argsHash, decision string, operator int64, reason string) { obs.RecordApproval("tool-gate", req.Tool, decision) if state == nil { return } - if err := state.RecordApprovalRow(req.RunID, "tool-gate", req.Tool, time.Now(), - decision, operator, argsHash, 1, 1, "", "", reason); err != nil { + decidedAt := time.Now() + receiptID := "rcpt_" + strings.TrimPrefix(newToolTicketID(), "tc_") + expires := req.expires + if expires.IsZero() { + expires = decidedAt.Add(4 * time.Hour) + } + quorumGot := 0 + if decision == "approve" || decision == "policy_approve" { + quorumGot = 1 + } + envelope := statestore.ApprovalEnvelope{ + ReceiptID: receiptID, RunID: req.RunID, ActionID: req.ActionID, + Pipeline: "tool-gate", Step: req.Tool, DecidedAt: decidedAt, + Decision: decision, OperatorID: operator, PayloadHash: argsHash, + QuorumN: 1, QuorumGot: quorumGot, Policy: reason, PolicyHash: req.policyHash, + BindingHash: req.bindingHash, ExpiresAt: expires, Lifecycle: "decided", + } + secret := []byte(os.Getenv("DRAFTCAT_APPROVAL_SECRET")) + if len(secret) > 0 { + nonce, err := approval.NewNonce() + if err == nil { + envelope.Nonce = nonce + envelope.Signature = approval.SignV2(secret, approval.FieldsV2{ + ReceiptID: envelope.ReceiptID, RunID: envelope.RunID, ActionID: envelope.ActionID, + Pipeline: envelope.Pipeline, Step: envelope.Step, DecidedAt: envelope.DecidedAt.Unix(), + Decision: envelope.Decision, OperatorID: envelope.OperatorID, PayloadHash: envelope.PayloadHash, + Policy: envelope.Policy, PolicyHash: envelope.PolicyHash, BindingHash: envelope.BindingHash, + ExpiresAt: envelope.ExpiresAt.Unix(), QuorumN: envelope.QuorumN, QuorumGot: envelope.QuorumGot, + }, nonce) + } + } + if err := state.RecordApprovalV2(envelope); err != nil { log.Printf("[tool-gate] audit write failed: %v", err) } _ = cfg } +func recordToolConsumption(a statestore.ToolAction, at time.Time) { + if state == nil { + return + } + e := statestore.ApprovalEnvelope{ + ReceiptID: "rcpt_" + strings.TrimPrefix(newToolTicketID(), "tc_"), + RunID: a.RunID, ActionID: a.ActionID, Pipeline: "tool-gate", Step: a.Tool, + DecidedAt: at, Decision: "consume", PayloadHash: a.ArgsHash, + QuorumN: 1, QuorumGot: 1, Policy: "permit-consume", PolicyHash: a.PolicyHash, + BindingHash: a.BindingHash, ExpiresAt: a.ExpiresAt, Lifecycle: "consumed", + } + secret := []byte(os.Getenv("DRAFTCAT_APPROVAL_SECRET")) + if len(secret) > 0 { + if nonce, err := approval.NewNonce(); err == nil { + e.Nonce = nonce + e.Signature = approval.SignV2(secret, approval.FieldsV2{ + ReceiptID: e.ReceiptID, RunID: e.RunID, ActionID: e.ActionID, + Pipeline: e.Pipeline, Step: e.Step, DecidedAt: e.DecidedAt.Unix(), + Decision: e.Decision, OperatorID: e.OperatorID, PayloadHash: e.PayloadHash, + Policy: e.Policy, PolicyHash: e.PolicyHash, BindingHash: e.BindingHash, + ExpiresAt: e.ExpiresAt.Unix(), QuorumN: e.QuorumN, QuorumGot: e.QuorumGot, + }, nonce) + } + } + if err := state.RecordApprovalV2(e); err != nil { + log.Printf("[tool-gate] consumption receipt write failed: %v", err) + } +} + func writeToolDecision(w http.ResponseWriter, code int, resp ToolCallResponse) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(code) diff --git a/tool_gate_policy_test.go b/tool_gate_policy_test.go index 7d708d5..729f4a8 100644 --- a/tool_gate_policy_test.go +++ b/tool_gate_policy_test.go @@ -6,6 +6,7 @@ import ( "net/http" "net/http/httptest" "path/filepath" + "strconv" "strings" "sync" "testing" @@ -75,6 +76,18 @@ func getGate(t *testing.T, g *toolGate, id, query string) (int, ToolCallResponse return rec.Code, resp } +func consumeGate(t *testing.T, g *toolGate, id, binding string) (int, ToolCallResponse) { + t.Helper() + rec := httptest.NewRecorder() + body := `{"binding_hash":` + strconv.Quote(binding) + `}` + g.HandleStatus(rec, httptest.NewRequest(http.MethodPost, toolGatePath+"/"+id+"/consume", strings.NewReader(body))) + var resp ToolCallResponse + if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil { + t.Fatalf("decode %q: %v", rec.Body.String(), err) + } + return rec.Code, resp +} + func glob(pattern string) config.ArgConstraint { return config.ArgConstraint{Glob: pattern} } // --- argument constraints --- @@ -349,6 +362,36 @@ func TestToolGate_BadModeIsRejected(t *testing.T) { } } +func TestToolGate_ActionBindingIdempotencyAndConsumeOnce(t *testing.T) { + st, err := statestore.OpenStateStore(filepath.Join(t.TempDir(), "state.db")) + if err != nil { + t.Fatal(err) + } + prev := state + state = st + t.Cleanup(func() { state = prev; _ = st.Close() }) + + g := newToolGate(gateCfg(config.ToolRule{Name: "send_email"}), nil) + body := `{"action_id":"send-42","tool":"send_email","agent":"a","args":{"to":"anna@example.com"}}` + _, first := postGate(t, g, body) + if first.State != "allowed" || first.BindingHash == "" || first.Consume == "" { + t.Fatalf("decision = %+v", first) + } + _, retry := postGate(t, g, body) + if retry.ActionID != first.ActionID || retry.BindingHash != first.BindingHash { + t.Fatalf("retry changed binding: first=%+v retry=%+v", first, retry) + } + if code, drift := postGate(t, g, `{"action_id":"send-42","tool":"send_email","args":{"to":"mallory@example.com"}}`); code != http.StatusConflict || drift.Decision != "deny" { + t.Fatalf("drift code=%d response=%+v", code, drift) + } + if code, permit := consumeGate(t, g, first.ActionID, first.BindingHash); code != http.StatusOK || permit.Permit != "execute" || permit.State != "consumed" { + t.Fatalf("first consume code=%d response=%+v", code, permit) + } + if code, permit := consumeGate(t, g, first.ActionID, first.BindingHash); code != http.StatusConflict || permit.Permit != "" { + t.Fatalf("second consume code=%d response=%+v", code, permit) + } +} + // --- durable rows --- // A tool-call gate waiting on a human is written to pending_approvals like a diff --git a/webhook_test.go b/webhook_test.go index 862727e..4bae0e3 100644 --- a/webhook_test.go +++ b/webhook_test.go @@ -1,13 +1,16 @@ package main import ( + "encoding/json" "net/http" "net/http/httptest" + "path/filepath" "strings" "testing" "time" "github.com/renezander030/draftcat/internal/config" + statestore "github.com/renezander030/draftcat/internal/state" ) func TestSchedulerTryStart(t *testing.T) { @@ -136,3 +139,107 @@ func TestWebhookAcceptsValid(t *testing.T) { } t.Error("pipeline did not finish / release running flag within timeout") } + +func TestListenerHealthAndReadiness(t *testing.T) { + prev := state + state = nil + t.Cleanup(func() { state = prev }) + h, _ := testHandler(t) + for path, want := range map[string]int{"/healthz": http.StatusOK, "/readyz": http.StatusServiceUnavailable} { + rr := httptest.NewRecorder() + h.ServeHTTP(rr, httptest.NewRequest(http.MethodGet, path, nil)) + if rr.Code != want { + t.Fatalf("%s = %d, want %d", path, rr.Code, want) + } + } + st, err := statestore.OpenStateStore(filepath.Join(t.TempDir(), "state.db")) + if err != nil { + t.Fatal(err) + } + state = st + t.Cleanup(func() { _ = st.Close() }) + h, _ = testHandler(t) + rr := httptest.NewRecorder() + h.ServeHTTP(rr, httptest.NewRequest(http.MethodGet, "/readyz", nil)) + if rr.Code != http.StatusOK { + t.Fatalf("readyz = %d, want 200", rr.Code) + } +} + +func TestToolGateRoutesRequireBearer(t *testing.T) { + prev := state + state = nil + t.Cleanup(func() { state = prev }) + cfg := &config.Config{Timeouts: config.TimeoutConfig{OperatorApproval: "1m"}} + cfg.Webhook.Enabled = true + cfg.Webhook.SetSecret("s3cret") + cfg.ToolGate = config.ToolGateConfig{Enabled: true, Tools: []config.ToolRule{{Name: "read_calendar"}}} + h := newWebhookHandler(cfg, newScheduler(nil), &BudgetTracker{dayStart: time.Now()}, &TGBot{}, nil) + body := `{"action_id":"read-1","tool":"read_calendar"}` + if rr := post(h, toolGatePath, "", body); rr.Code != http.StatusUnauthorized { + t.Fatalf("unauthenticated gate = %d, want 401", rr.Code) + } + if rr := post(h, toolGatePath, "Bearer s3cret", body); rr.Code != http.StatusOK { + t.Fatalf("authenticated gate = %d (%s), want 200", rr.Code, rr.Body.String()) + } +} + +func TestToolGatePostUsesConfiguredBodySignature(t *testing.T) { + st := newTempStateStore(t) + prev := state + state = st + t.Cleanup(func() { state = prev }) + cfg := &config.Config{Timeouts: config.TimeoutConfig{OperatorApproval: "1m"}} + cfg.Webhook.Enabled = true + cfg.Webhook.RequireSignature = true + cfg.Webhook.SetSecret("s3cret") + cfg.ToolGate = config.ToolGateConfig{Enabled: true, Tools: []config.ToolRule{{Name: "read_calendar"}}} + h := newWebhookHandler(cfg, newScheduler(nil), &BudgetTracker{dayStart: time.Now()}, &TGBot{}, nil) + body := `{"action_id":"read-signed","tool":"read_calendar"}` + if rr := post(h, toolGatePath, "Bearer s3cret", body); rr.Code != http.StatusUnauthorized { + t.Fatalf("unsigned gate POST = %d, want 401", rr.Code) + } + req := httptest.NewRequest(http.MethodPost, toolGatePath, strings.NewReader(body)) + req.Header.Set("Authorization", "Bearer s3cret") + req.Header.Set(webhookSigHeader, signBody([]byte("s3cret"), time.Now().Unix(), []byte(body))) + rr := httptest.NewRecorder() + h.ServeHTTP(rr, req) + if rr.Code != http.StatusOK { + t.Fatalf("signed gate POST = %d (%s), want 200", rr.Code, rr.Body.String()) + } +} + +func TestWebhookWritesAdmissionBeforeAccepted(t *testing.T) { + st, err := statestore.OpenStateStore(filepath.Join(t.TempDir(), "state.db")) + if err != nil { + t.Fatal(err) + } + prev := state + state = st + t.Cleanup(func() { state = prev; _ = st.Close() }) + h, _ := testHandler(t) + rr := post(h, "/hooks/ping", "Bearer s3cret", `{"hello":"world"}`) + if rr.Code != http.StatusAccepted { + t.Fatalf("code=%d body=%s", rr.Code, rr.Body.String()) + } + var body struct { + AdmissionID string `json:"admission_id"` + } + if err := json.Unmarshal(rr.Body.Bytes(), &body); err != nil || body.AdmissionID == "" { + t.Fatalf("response=%q err=%v", rr.Body.String(), err) + } + deadline := time.Now().Add(2 * time.Second) + for { + a, err := st.WebhookAdmission(body.AdmissionID) + if err != nil || a.BodyHash == "" { + t.Fatalf("admission=%+v err=%v", a, err) + } + if a.Status == "completed" { + break + } + if time.Now().After(deadline) { + t.Fatalf("admission stayed %q", a.Status) + } + time.Sleep(10 * time.Millisecond) + } +}