diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 1586679..f51d405 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -11,8 +11,33 @@ permissions: contents: read packages: write +concurrency: + group: release-${{ github.ref }} + cancel-in-progress: false + jobs: + verify: + if: startsWith(github.ref, 'refs/tags/v') + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-go@v5 + with: { go-version-file: go.mod } + - uses: actions/setup-node@v4 + with: { node-version: "24" } + - name: Check release version + run: | + test "v$(node -p 'require("./package.json").version')" = "$GITHUB_REF_NAME" + test "$(go run . --version)" = "draftcat ${GITHUB_REF_NAME#v}" + - run: go test -short -count=1 -timeout 60s ./... + - run: go test -short -tags voice -count=1 -timeout 60s ./... + - run: go run . validate + - run: npm ci --ignore-scripts + - run: npm test + - run: npm pack --dry-run + binaries: + needs: verify if: startsWith(github.ref, 'refs/tags/v') runs-on: ubuntu-latest strategy: @@ -27,7 +52,7 @@ jobs: steps: - uses: actions/checkout@v4 - uses: actions/setup-go@v5 - with: { go-version: "1.25" } + with: { go-version-file: go.mod } - name: Build native binary env: GOOS: ${{ matrix.goos }} @@ -35,7 +60,7 @@ jobs: CGO_ENABLED: "0" run: | asset="draftcat-${GITHUB_REF_NAME}-${{ matrix.platform }}-${{ matrix.arch }}${{ matrix.extension }}" - go build -trimpath -buildvcs=false -ldflags="-s -w" -o "$asset" . + go build -trimpath -buildvcs=false -ldflags="-s -w -X main.version=${GITHUB_REF_NAME#v}" -o "$asset" . gzip -n "$asset" - uses: actions/upload-artifact@v4 with: @@ -61,9 +86,15 @@ jobs: GH_TOKEN: ${{ github.token }} run: | sha256sum draftcat-*.gz > SHA256SUMS - gh release create "$GITHUB_REF_NAME" --repo "$GITHUB_REPOSITORY" --generate-notes --title "$GITHUB_REF_NAME" draftcat-*.gz SHA256SUMS + if gh release view "$GITHUB_REF_NAME" --repo "$GITHUB_REPOSITORY" >/dev/null 2>&1; then + gh release upload "$GITHUB_REF_NAME" --repo "$GITHUB_REPOSITORY" --clobber draftcat-*.gz SHA256SUMS + else + gh release create "$GITHUB_REF_NAME" --repo "$GITHUB_REPOSITORY" --generate-notes --title "$GITHUB_REF_NAME" draftcat-*.gz SHA256SUMS + fi image: + if: startsWith(github.ref, 'refs/tags/v') + needs: verify runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 @@ -91,3 +122,27 @@ jobs: push: true tags: ${{ steps.meta.outputs.tags }} labels: ${{ steps.meta.outputs.labels }} + + npm: + if: startsWith(github.ref, 'refs/tags/v') + needs: release + runs-on: ubuntu-latest + permissions: + contents: read + id-token: write + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-node@v4 + with: + node-version: "24" + registry-url: https://registry.npmjs.org + - run: npm ci --ignore-scripts + - run: npm test + - name: Smoke-test the published native asset + run: | + node npm/install.js + test "$(node npm/draftcat.js --version)" = "draftcat ${GITHUB_REF_NAME#v}" + - name: Publish npm package + env: + NODE_AUTH_TOKEN: ${{ secrets.NPM_TOKEN }} + run: npm publish --access public --provenance diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..79f8315 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,19 @@ +# Changelog + +## 0.8.0 + +- Add durable, pipeline-scoped `Idempotency-Key` webhook retries. Matching bodies return the original admission; changed bodies are rejected. +- Claim signed webhook replay identities atomically, including concurrent requests. +- Upgrade older SQLite stores transactionally before indexing new columns. +- Refuse consume requests when the tool rule, operator channel, allowed reviewers, or approval timeout changed after the decision. +- Reject oversized and unreadable request bodies, unknown tool envelope fields, duplicate JSON keys, and trailing JSON. +- Preserve exact tool argument numbers for approval hashes and compare numeric policy bounds without rounding. +- Persist completed and failed pipeline runs with their run IDs; join approvals by exact identity while preserving legacy history. +- Add `draftcat receipts verify [--json]` to verify exported signed receipt fields without SQLite. +- Open audit and inspection commands in read-only mode, without creating or migrating databases. + +Release packaging includes `--version`, synchronized npm metadata, and a tag workflow that tests before building native assets, publishing the container, and publishing npm after a native installer smoke test. + +### Upgrade notes + +Start the engine once to migrate an older state store before using read-only audit commands. Existing v1/v2 receipt signatures remain unchanged. Unconsumed permits from v0.7.0 require a new approval because the effective policy binding now includes the operator authorization configuration. Tool request envelopes are stricter and retain numeric token spelling; send the same spelling when retrying an action ID. diff --git a/README.md b/README.md index 76bf1f7..9e15eb7 100644 --- a/README.md +++ b/README.md @@ -65,6 +65,8 @@ 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.8.0:** webhook retries can carry a durable `Idempotency-Key`, signed replay identities are claimed atomically, and tool permits recheck their policy before execution. Requests reject oversized or ambiguous data and retain exact numbers. Older state stores upgrade safely; completed and failed runs carry exact approval identities. Audit commands read without modifying the database, and `draftcat receipts verify` checks exported JSONL offline. See the [upgrade and reliability guide](docs/reliability.md). +> > **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.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. @@ -138,6 +140,12 @@ draftcat --help The installer downloads the matching Linux, macOS, or Windows binary and verifies it against the checksums attached to the GitHub release. No Go toolchain is required. +With a Go toolchain, install the CLI from the module: + +```bash +go install github.com/renezander030/draftcat@latest +``` + Or build from source: ```bash diff --git a/docs/action-receipts.md b/docs/action-receipts.md index 2216c70..ecf0717 100644 --- a/docs/action-receipts.md +++ b/docs/action-receipts.md @@ -32,7 +32,7 @@ decision fields with HMAC-SHA256: - quorum result - policy and policy digest - binding digest -- permit expiry and lifecycle +- permit expiry - nonce See [`internal/approval/receipt.go`](../internal/approval/receipt.go). If any @@ -107,6 +107,17 @@ Set `DRAFTCAT_APPROVAL_SECRET` while reading to receive `verification: ok` or `tampered`. Signed rows without the key report `unverified`; unsigned rows report `unsigned` explicitly. +Verify exported records without opening SQLite: + +```bash +draftcat receipts verify receipts.jsonl --json +cat receipts.jsonl | draftcat receipts verify - +``` + +This requires `DRAFTCAT_APPROVAL_SECRET`, checks each signed v1/v2 field set, +and returns nonzero for unsigned or tampered rows. It does not certify export +completeness or unsigned `lifecycle` metadata. See the [reliability guide](reliability.md). + ## Design rule Do not let the model decide whether the approval boundary was satisfied. The diff --git a/docs/reliability.md b/docs/reliability.md new file mode 100644 index 0000000..8bcaea7 --- /dev/null +++ b/docs/reliability.md @@ -0,0 +1,86 @@ +# Reliable requests and audit inspection + +## Retry a webhook without repeating the action + +Choose one opaque key for one intended pipeline execution and keep it for every retry: + +```bash +curl -X POST https://draftcat.example/hooks/invoice-due-diligence \ + -H "Authorization: Bearer $DRAFTCAT_WEBHOOK_SECRET" \ + -H "Idempotency-Key: invoice-4821" \ + -d '{"invoice":"4821"}' +``` + +The key is scoped to the pipeline and must be 1–128 URL-safe characters (letters, +digits, underscore, hyphen, dot, or colon). SQLite stores its hash and the exact +body hash. Matching retries return HTTP 202 with the original `admission_id`, +current status, and poll path even after completion or restart. A changed body +returns 409. Missing keys retain the previous behavior. Retain the key while the +admission exists; keys do not automatically expire. A retry never resumes an +interrupted admission: inspect it and explicitly choose a fresh key for a new run. + +Bearer authentication is checked on every retry. When signatures are enabled, +every retry also needs an authentic signature within the clock-skew window. +A matching, already-admitted retry may reuse the original signature while it is +still timely; it only reads the existing admission. Starting new work claims the +signature atomically. Reusing that signature for another admission is refused. + +## Send complete, unambiguous tool requests + +Oversized HTTP request bodies return 413 instead of being truncated; unreadable +bodies return 400. `webhook.max_body_bytes` applies to webhook and tool-gate POSTs. +The tool gate also limits direct requests to 1 MiB and consume bodies to 4 KiB. +Unknown tool envelope fields, duplicate keys (including nested arguments), and +trailing JSON, and nesting above 128 levels return 400 before creating or consuming a permit. Webhook payloads +remain arbitrary bytes; the strict JSON envelope rules apply to the tool gate. + +Tool argument numbers retain their JSON spelling, including integers above 2^53. +Distinct large integers therefore cannot share a rounded approval hash. Use the +same numeric spelling on an action ID retry: `1`, `1.0`, and `1e0` bind different +representations. Numeric min/max constraints compare exact decimal values and +reject non-finite strings such as `NaN`. YAML numeric bounds keep their existing +float64 configuration type; comparisons retain the supplied argument's exact digits. + +## Reapprove after a policy change + +Consumption compares the saved policy digest with the current effective tool +policy, including operator-channel settings, permitted reviewers, and the +approval window. A mismatch returns 409 without an execution permit. Request a +new action ID and obtain a fresh approval. This check also applies to permits +loaded after restart. Upgrading from v0.7.0 changes the policy digest, so obtain +fresh approval for any unconsumed permits. + +## Upgrade and inspect state + +Engine startup migrates older SQLite stores in one transaction. Indexes on new +columns are created after those columns exist, and previous receipt signatures +are retained. Completed and failed pipeline runs are recorded with the run ID +used by their approvals. `draftcat runs --json` includes `run_id` for new runs; +identity-less historical records use the old timestamp join, restricted to +identity-less approvals. + +`runs`, `pending`, `receipts list|show|export`, and `audit-verify` open existing +state in read-only mode. A missing path fails without creating a new database. +Start the engine once to migrate an older store before inspecting it. Inspection +can read a live WAL-backed database; it does not run schema migrations. + +## Verify exported receipt fields offline + +```bash +draftcat receipts export --out receipts.jsonl +draftcat receipts verify receipts.jsonl --json +cat receipts.jsonl | draftcat receipts verify - +``` + +Set `DRAFTCAT_APPROVAL_SECRET` to the same signing key used by the instance. +The verifier opens no state database and recomputes each v1/v2 signature instead +of trusting the export's `verification` field. Exit 0 means all listed receipts +have valid signatures; unsigned or tampered rows return 1; a missing signing key +or invalid command returns 2. Malformed, oversized, empty, or unsupported-version +input fails. The JSON report includes line number, receipt ID, and verdict. + +HMAC verification requires the signing secret. It verifies the fields covered by +the corresponding receipt version; it does not attest that a JSONL file is +complete, prove actual delivery, or authenticate unsigned display metadata such +as `lifecycle`. Keep the secret within the trusted operator boundary. Use +`zk-receipt` for the separate experimental privacy-preserving proof flow. diff --git a/internal/config/config.go b/internal/config/config.go index a3e6744..71daf61 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -8,6 +8,8 @@ package config import ( "encoding/json" "fmt" + "math" + "math/big" "path" "regexp" "sort" @@ -573,7 +575,7 @@ type ToolRule struct { } // ArgConstraint is one condition on one argument. Every non-empty field must -// hold. Values are compared in their string form (numbers without exponent, +// hold. Values are compared in their string form (JSON number spelling is retained, // booleans as true/false, anything structured as compact JSON) except min/max, // which need a number. type ArgConstraint struct { @@ -631,14 +633,14 @@ func (c ArgConstraint) Check(v interface{}, present bool) (bool, string) { } } if c.Max != nil || c.Min != nil { - f, ok := argNumber(v) + f, ok := exactArgNumber(v) if !ok { return false, fmt.Sprintf("%q is not a number", clip(s)) } - if c.Max != nil && f > *c.Max { + if c.Max != nil && (math.IsNaN(*c.Max) || math.IsInf(*c.Max, 0) || f.Cmp(exactBound(*c.Max)) > 0) { return false, fmt.Sprintf("%s exceeds max %s", s, strconv.FormatFloat(*c.Max, 'f', -1, 64)) } - if c.Min != nil && f < *c.Min { + if c.Min != nil && (math.IsNaN(*c.Min) || math.IsInf(*c.Min, 0) || f.Cmp(exactBound(*c.Min)) < 0) { return false, fmt.Sprintf("%s is below min %s", s, strconv.FormatFloat(*c.Min, 'f', -1, 64)) } } @@ -673,25 +675,25 @@ func ArgString(v interface{}) string { } } -func argNumber(v interface{}) (float64, bool) { - switch x := v.(type) { - case float64: - return x, true - case float32: - return float64(x), true - case int: - return float64(x), true - case int64: - return float64(x), true - case json.Number: - f, err := x.Float64() - return f, err == nil - case string: - f, err := strconv.ParseFloat(strings.TrimSpace(x), 64) - return f, err == nil - default: - return 0, false +func exactBound(v float64) *big.Rat { + r, _ := new(big.Rat).SetString(strconv.FormatFloat(v, 'g', -1, 64)) + return r +} + +func exactArgNumber(v interface{}) (*big.Rat, bool) { + s := strings.TrimSpace(ArgString(v)) + if i := strings.IndexAny(s, "eE"); i >= 0 { + exponent, err := strconv.Atoi(s[i+1:]) + if err != nil || exponent < -4096 || exponent > 4096 { + return nil, false + } + } + // Bound exponent work and reject NaN/Inf while retaining exact decimal digits. + f, err := strconv.ParseFloat(s, 64) + if err != nil || math.IsNaN(f) || math.IsInf(f, 0) { + return nil, false } + return new(big.Rat).SetString(s) } func clip(s string) string { diff --git a/internal/config/precise_numbers_test.go b/internal/config/precise_numbers_test.go new file mode 100644 index 0000000..c58c064 --- /dev/null +++ b/internal/config/precise_numbers_test.go @@ -0,0 +1,25 @@ +package config + +import ( + "encoding/json" + "testing" +) + +func TestExactNumericConstraintBoundary(t *testing.T) { + max := float64(9007199254740992) + c := ArgConstraint{Max: &max} + for _, v := range []interface{}{json.Number("9007199254740993"), "9007199254740993", "NaN", "Inf", "-Inf", "1e-1000000000"} { + if ok, _ := c.Check(v, true); ok { + t.Errorf("unsafe boundary %v accepted", v) + } + } + if ok, why := c.Check(json.Number("9007199254740992"), true); !ok { + t.Fatalf("exact bound rejected: %s", why) + } + min := 1.1 + max = 1.1 + c = ArgConstraint{Min: &min, Max: &max} + if ok, why := c.Check(json.Number("1.1"), true); !ok { + t.Fatalf("decimal bound rejected: %s", why) + } +} diff --git a/internal/state/reliability_test.go b/internal/state/reliability_test.go new file mode 100644 index 0000000..4681d7d --- /dev/null +++ b/internal/state/reliability_test.go @@ -0,0 +1,153 @@ +package state + +import ( + "context" + "database/sql" + "os" + "path/filepath" + "sync" + "sync/atomic" + "testing" + "time" +) + +func TestUpgradeStoreBeforeRunIDColumns(t *testing.T) { + path := filepath.Join(t.TempDir(), "old.db") + db, err := sql.Open("sqlite", path) + if err != nil { + t.Fatal(err) + } + _, err = db.ExecContext(context.Background(), `CREATE TABLE pipeline_runs (id INTEGER PRIMARY KEY, pipeline TEXT, started_at INTEGER, ended_at INTEGER, status TEXT, error_text TEXT); + CREATE TABLE action_approvals (id INTEGER PRIMARY KEY, pipeline TEXT, step TEXT, decided_at INTEGER, decision TEXT, operator_id INTEGER, payload_hash TEXT, quorum_n INTEGER, quorum_got INTEGER); + INSERT INTO pipeline_runs VALUES (1,'p',100,101,'ok',''); + INSERT INTO action_approvals VALUES (1,'p','review',100,'approve',7,'payload',1,1);`) + if err != nil { + t.Fatal(err) + } + if err := db.Close(); err != nil { + t.Fatal(err) + } + st, err := OpenStateStore(path) + if err != nil { + t.Fatal(err) + } + defer func() { _ = st.Close() }() + rows, err := st.AllApprovals(10) + if err != nil || len(rows) != 1 || rows[0].Version != 1 || rows[0].PayloadHash != "payload" { + t.Fatalf("legacy approvals=%+v err=%v", rows, err) + } + runs, err := st.AllRecentRuns(10) + if err != nil || len(runs) != 1 || runs[0].RunID != "" { + t.Fatalf("legacy runs=%+v err=%v", runs, err) + } + if err := st.RecordRunForID("new-run", "p", time.Now(), time.Now(), nil); err != nil { + t.Fatal(err) + } + if err := st.Close(); err != nil { + t.Fatal(err) + } + reopened, err := OpenStateStore(path) + if err != nil { + t.Fatalf("repeat upgrade: %v", err) + } + _ = reopened.Close() +} + +func TestAtomicReplayClaimAcrossConnections(t *testing.T) { + path := filepath.Join(t.TempDir(), "state.db") + first, err := OpenStateStore(path) + if err != nil { + t.Fatal(err) + } + defer func() { _ = first.Close() }() + second, err := OpenStateStore(path) + if err != nil { + t.Fatal(err) + } + defer func() { _ = second.Close() }() + var winners atomic.Int32 + var wg sync.WaitGroup + for i := 0; i < 24; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + st := first + if i%2 == 1 { + st = second + } + won, err := st.TryMarkSeen("hooks", "sig", "shared", time.Now()) + if err != nil { + t.Error(err) + } + if won { + winners.Add(1) + } + }(i) + } + wg.Wait() + if winners.Load() != 1 { + t.Fatalf("winners=%d", winners.Load()) + } +} + +func TestDurableWebhookRetryIdentity(t *testing.T) { + path := filepath.Join(t.TempDir(), "state.db") + st, err := OpenStateStore(path) + if err != nil { + t.Fatal(err) + } + now := time.Now() + a := WebhookAdmission{ID: "wh_first", Pipeline: "p", BodyHash: "body-a", IdempotencyKey: "hash-key", CreatedAt: now, UpdatedAt: now} + saved, created, err := st.AdmitWebhook(a) + if err != nil || !created { + t.Fatalf("first=%+v created=%v err=%v", saved, created, err) + } + if err := st.Close(); err != nil { + t.Fatal(err) + } + st, err = OpenStateStore(path) + if err != nil { + t.Fatal(err) + } + defer func() { _ = st.Close() }() + a.ID = "wh_retry" + saved, created, err = st.AdmitWebhook(a) + if err != nil || created || saved.ID != "wh_first" { + t.Fatalf("retry=%+v created=%v err=%v", saved, created, err) + } + a.BodyHash = "body-b" + if _, _, err := st.AdmitWebhook(a); err == nil { + t.Fatal("changed retry payload accepted") + } + a.Pipeline = "other" + if _, created, err := st.AdmitWebhook(a); err != nil || !created { + t.Fatalf("pipeline scope created=%v err=%v", created, err) + } +} + +func TestAuditStoreReadOnlyAndMissingPath(t *testing.T) { + path := filepath.Join(t.TempDir(), "missing.db") + if st, err := OpenStateStoreReadOnly(path); err == nil { + _ = st.Close() + t.Fatal("missing store opened") + } + if _, err := os.Stat(path); !os.IsNotExist(err) { + t.Fatalf("missing file created: %v", err) + } + st, err := OpenStateStore(path) + if err != nil { + t.Fatal(err) + } + _ = st.Close() + ro, err := OpenStateStoreReadOnly(path) + if err != nil { + t.Fatal(err) + } + defer func() { _ = ro.Close() }() + if _, err := ro.AllRecentRuns(10); err != nil { + t.Fatal(err) + } + if err := ro.RecordRun("p", time.Now(), time.Now(), nil); err == nil { + t.Fatal("read-only audit connection wrote a row") + } +} diff --git a/internal/state/state.go b/internal/state/state.go index 8a3fa34..d0baa46 100644 --- a/internal/state/state.go +++ b/internal/state/state.go @@ -4,6 +4,8 @@ import ( "context" "database/sql" "fmt" + "net/url" + "path/filepath" "strings" "time" @@ -29,7 +31,11 @@ func OpenStateStore(path string) (*StateStore, error) { if err != nil { return nil, fmt.Errorf("open state store %s: %w", path, err) } + // Keep the connection-level pragmas on every operation in this store. + // SQLite permits one writer; separate store handles can share the WAL. + db.SetMaxOpenConns(1) if _, err := db.Exec(`PRAGMA journal_mode=WAL; PRAGMA synchronous=NORMAL; PRAGMA busy_timeout=5000;`); err != nil { + _ = db.Close() return nil, fmt.Errorf("state store pragmas: %w", err) } if err := initStateSchema(db); err != nil { @@ -39,7 +45,39 @@ func OpenStateStore(path string) (*StateStore, error) { return &StateStore{db: db}, nil } +// OpenStateStoreReadOnly inspects an existing store without creating or migrating it. +// Older stores must first be opened by the engine to apply their migrations. +func OpenStateStoreReadOnly(path string) (*StateStore, error) { + absolute, err := filepath.Abs(path) + if err != nil { + return nil, err + } + uriPath := filepath.ToSlash(absolute) + if !strings.HasPrefix(uriPath, "/") { + uriPath = "/" + uriPath + } + u := url.URL{Scheme: "file", Path: uriPath} + q := u.Query() + q.Set("mode", "ro") + q.Set("_pragma", "busy_timeout(5000)") + u.RawQuery = q.Encode() + db, err := sql.Open("sqlite", u.String()) + if err != nil { + return nil, err + } + if err := db.PingContext(context.Background()); err != nil { + _ = db.Close() + return nil, err + } + return &StateStore{db: db}, nil +} + func initStateSchema(db *sql.DB) error { + tx, err := db.BeginTx(context.Background(), nil) + if err != nil { + return err + } + defer func() { _ = tx.Rollback() }() const schema = ` CREATE TABLE IF NOT EXISTS seen_items ( pipeline TEXT NOT NULL, @@ -58,7 +96,6 @@ CREATE TABLE IF NOT EXISTS pipeline_runs ( run_id TEXT NOT NULL DEFAULT '' -- minted at run start; joins to action_approvals ); CREATE INDEX IF NOT EXISTS idx_runs_pipeline ON pipeline_runs(pipeline, started_at DESC); -CREATE INDEX IF NOT EXISTS idx_runs_runid ON pipeline_runs(run_id); CREATE TABLE IF NOT EXISTS action_approvals ( id INTEGER PRIMARY KEY AUTOINCREMENT, pipeline TEXT NOT NULL, @@ -82,7 +119,6 @@ CREATE TABLE IF NOT EXISTS action_approvals ( 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); CREATE TABLE IF NOT EXISTS pending_approvals ( id INTEGER PRIMARY KEY AUTOINCREMENT, pipeline TEXT NOT NULL, @@ -116,6 +152,7 @@ CREATE TABLE IF NOT EXISTS webhook_admissions ( id TEXT PRIMARY KEY, pipeline TEXT NOT NULL, body_hash TEXT NOT NULL, + idempotency_key TEXT NOT NULL DEFAULT '', status TEXT NOT NULL, error_text TEXT NOT NULL DEFAULT '', created_at INTEGER NOT NULL, @@ -123,7 +160,7 @@ CREATE TABLE IF NOT EXISTS webhook_admissions ( ); CREATE INDEX IF NOT EXISTS idx_webhook_admissions_status ON webhook_admissions(status, created_at); ` - if _, err := db.ExecContext(context.Background(), schema); err != nil { + if _, err := tx.ExecContext(context.Background(), schema); err != nil { return err } // Migrate stores created before signed receipts existed: add the columns if @@ -147,12 +184,33 @@ CREATE INDEX IF NOT EXISTS idx_webhook_admissions_status ON webhook_admissions(s `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'`, + `ALTER TABLE webhook_admissions ADD COLUMN idempotency_key TEXT NOT NULL DEFAULT ''`, } { - if _, err := db.ExecContext(context.Background(), alter); err != nil && !strings.Contains(err.Error(), "duplicate column name") { + if _, err := tx.ExecContext(context.Background(), alter); err != nil && !strings.Contains(err.Error(), "duplicate column name") { return err } } - return nil + if _, err := tx.ExecContext(context.Background(), ` + CREATE INDEX IF NOT EXISTS idx_runs_runid ON pipeline_runs(run_id); + CREATE INDEX IF NOT EXISTS idx_approvals_runid ON action_approvals(run_id); + CREATE UNIQUE INDEX IF NOT EXISTS idx_webhook_idempotency ON webhook_admissions(pipeline, idempotency_key) WHERE idempotency_key <> ''; + `); err != nil { + return err + } + return tx.Commit() +} + +// TryMarkSeen atomically claims one item. Exactly one concurrent caller wins. +func (s *StateStore) TryMarkSeen(pipeline, scope, id string, at time.Time) (bool, error) { + if s == nil || s.db == nil { + return false, fmt.Errorf("state store unavailable") + } + res, err := s.db.ExecContext(context.Background(), `INSERT OR IGNORE INTO seen_items (pipeline, scope, item_id, seen_at) VALUES (?, ?, ?, ?)`, pipeline, scope, id, at.Unix()) + if err != nil { + return false, err + } + n, err := res.RowsAffected() + return n == 1, err } // FilterUnseen returns the subset of ids not previously marked as seen for @@ -222,6 +280,11 @@ func (s *StateStore) MarkSeen(pipeline, scope string, ids []string) error { // RecordRun appends a pipeline run record. Failures here are surfaced but // must not halt the engine — observability is best-effort. func (s *StateStore) RecordRun(pipeline string, started, ended time.Time, runErr error) error { + return s.RecordRunForID("", pipeline, started, ended, runErr) +} + +// RecordRunForID records the exact identity carried by this run's approvals. +func (s *StateStore) RecordRunForID(runID, pipeline string, started, ended time.Time, runErr error) error { status := "ok" var errText string if runErr != nil { @@ -229,8 +292,8 @@ func (s *StateStore) RecordRun(pipeline string, started, ended time.Time, runErr errText = runErr.Error() } _, err := s.db.Exec( - `INSERT INTO pipeline_runs (pipeline, started_at, ended_at, status, error_text) VALUES (?, ?, ?, ?, ?)`, - pipeline, started.Unix(), ended.Unix(), status, errText, + `INSERT INTO pipeline_runs (run_id, pipeline, started_at, ended_at, status, error_text) VALUES (?, ?, ?, ?, ?, ?)`, + runID, pipeline, started.Unix(), ended.Unix(), status, errText, ) return err } @@ -238,6 +301,7 @@ func (s *StateStore) RecordRun(pipeline string, started, ended time.Time, runErr // RecentRuns returns the last n runs for a pipeline, newest first. Used by // the /status operator command. type RunRecord struct { + RunID string Pipeline string StartedAt time.Time EndedAt time.Time @@ -247,7 +311,7 @@ type RunRecord struct { func (s *StateStore) RecentRuns(pipeline string, n int) ([]RunRecord, error) { rows, err := s.db.Query( - `SELECT pipeline, started_at, ended_at, status, COALESCE(error_text,'') FROM pipeline_runs WHERE pipeline=? ORDER BY started_at DESC LIMIT ?`, + `SELECT run_id, pipeline, started_at, ended_at, status, COALESCE(error_text,'') FROM pipeline_runs WHERE pipeline=? ORDER BY started_at DESC LIMIT ?`, pipeline, n, ) if err != nil { @@ -258,7 +322,7 @@ func (s *StateStore) RecentRuns(pipeline string, n int) ([]RunRecord, error) { for rows.Next() { var r RunRecord var st, en int64 - if err := rows.Scan(&r.Pipeline, &st, &en, &r.Status, &r.Error); err != nil { + if err := rows.Scan(&r.RunID, &r.Pipeline, &st, &en, &r.Status, &r.Error); err != nil { return nil, err } r.StartedAt = time.Unix(st, 0) @@ -287,7 +351,7 @@ func (s *StateStore) AllRecentRuns(n int) ([]RunRecord, error) { return nil, nil } rows, err := s.db.QueryContext(context.Background(), - `SELECT pipeline, started_at, ended_at, status, COALESCE(error_text,'') + `SELECT run_id, pipeline, started_at, ended_at, status, COALESCE(error_text,'') FROM pipeline_runs ORDER BY started_at DESC LIMIT ?`, n) if err != nil { return nil, err @@ -297,7 +361,7 @@ func (s *StateStore) AllRecentRuns(n int) ([]RunRecord, error) { for rows.Next() { var r RunRecord var st, en int64 - if err := rows.Scan(&r.Pipeline, &st, &en, &r.Status, &r.Error); err != nil { + if err := rows.Scan(&r.RunID, &r.Pipeline, &st, &en, &r.Status, &r.Error); err != nil { return nil, err } r.StartedAt = time.Unix(st, 0) @@ -838,24 +902,57 @@ func (s *StateStore) ExpireToolActions(at time.Time) error { // WebhookAdmission is the durable evidence that an HTTP trigger was accepted. type WebhookAdmission struct { - ID string - Pipeline string - BodyHash string - Status string - Error string - CreatedAt time.Time - UpdatedAt time.Time + IdempotencyKey string // SHA-256 of the caller key, never its raw value. + ID string + Pipeline string + BodyHash string + Status string + Error string + CreatedAt time.Time + UpdatedAt time.Time } // BeginWebhookAdmission writes the accepted record before the handler returns 202. func (s *StateStore) BeginWebhookAdmission(a WebhookAdmission) error { + _, _, err := s.AdmitWebhook(a) + return err +} + +// AdmitWebhook atomically reserves a pipeline-scoped retry identity. +func (s *StateStore) AdmitWebhook(a WebhookAdmission) (WebhookAdmission, bool, error) { if s == nil || s.db == nil { - return fmt.Errorf("state store unavailable") + return a, false, fmt.Errorf("state store unavailable") } - _, err := s.db.ExecContext(context.Background(), - `INSERT INTO webhook_admissions (id, pipeline, body_hash, status, created_at, updated_at) - VALUES (?, ?, ?, 'accepted', ?, ?)`, a.ID, a.Pipeline, a.BodyHash, a.CreatedAt.Unix(), a.UpdatedAt.Unix()) - return err + res, err := s.db.ExecContext(context.Background(), `INSERT OR IGNORE INTO webhook_admissions + (id, pipeline, body_hash, idempotency_key, status, created_at, updated_at) + VALUES (?, ?, ?, ?, 'accepted', ?, ?)`, a.ID, a.Pipeline, a.BodyHash, a.IdempotencyKey, a.CreatedAt.Unix(), a.UpdatedAt.Unix()) + if err != nil { + return a, false, err + } + n, err := res.RowsAffected() + if err != nil { + return a, false, err + } + var saved WebhookAdmission + if a.IdempotencyKey == "" { + saved, err = s.WebhookAdmission(a.ID) + } else { + saved, err = s.WebhookAdmissionForKey(a.Pipeline, a.IdempotencyKey) + } + if err != nil { + return saved, false, err + } + if saved.Pipeline != a.Pipeline || saved.BodyHash != a.BodyHash { + return saved, false, fmt.Errorf("idempotency key is already bound to another request body") + } + return saved, n == 1, nil +} + +func (s *StateStore) WebhookAdmissionForKey(pipeline, key string) (WebhookAdmission, error) { + if s == nil || s.db == nil { + return WebhookAdmission{}, sql.ErrNoRows + } + return scanWebhookAdmission(s.db.QueryRowContext(context.Background(), `SELECT id, pipeline, body_hash, idempotency_key, status, error_text, created_at, updated_at FROM webhook_admissions WHERE pipeline=? AND idempotency_key=?`, pipeline, key).Scan) } // FinishWebhookAdmission records the terminal pipeline outcome. @@ -874,12 +971,13 @@ func (s *StateStore) WebhookAdmission(id string) (WebhookAdmission, error) { if s == nil || s.db == nil { return WebhookAdmission{}, sql.ErrNoRows } + return scanWebhookAdmission(s.db.QueryRowContext(context.Background(), `SELECT id, pipeline, body_hash, idempotency_key, status, error_text, created_at, updated_at FROM webhook_admissions WHERE id=?`, id).Scan) +} + +func scanWebhookAdmission(scan func(...interface{}) error) (WebhookAdmission, error) { var a WebhookAdmission var created, updated int64 - err := s.db.QueryRowContext(context.Background(), - `SELECT id, pipeline, body_hash, status, error_text, created_at, updated_at - FROM webhook_admissions WHERE id=?`, id).Scan( - &a.ID, &a.Pipeline, &a.BodyHash, &a.Status, &a.Error, &created, &updated) + err := scan(&a.ID, &a.Pipeline, &a.BodyHash, &a.IdempotencyKey, &a.Status, &a.Error, &created, &updated) if err != nil { return a, err } @@ -887,6 +985,12 @@ func (s *StateStore) WebhookAdmission(id string) (WebhookAdmission, error) { return a, nil } +// RevokeToolAction closes an allowed permit whose policy no longer applies. +func (s *StateStore) RevokeToolAction(id, policy string, at time.Time) error { + _, err := s.db.ExecContext(context.Background(), `UPDATE tool_actions SET status='expired', decision='deny', reason='policy changed; request a new approval', updated_at=? WHERE action_id=? AND policy_hash=? AND status='allowed'`, at.Unix(), id, policy) + return err +} + // InterruptWebhookAdmissions makes accepted work visible after a restart. func (s *StateStore) InterruptWebhookAdmissions(at time.Time) error { if s == nil || s.db == nil { diff --git a/main.go b/main.go index d352bf2..84f9e38 100644 --- a/main.go +++ b/main.go @@ -6,8 +6,10 @@ import ( "crypto/hmac" "crypto/sha256" "crypto/subtle" + "database/sql" "encoding/hex" "encoding/json" + "errors" "fmt" "github.com/renezander030/draftcat/internal/config" skillsapi "github.com/renezander030/draftcat/internal/skills" @@ -1253,7 +1255,16 @@ func (t *TGBot) getUpdates() ([]TGUpdate, error) { func runPipeline(cfg *config.Config, pipeline config.PipelineConfig, budget *BudgetTracker, ch OperatorChannel, skills *skillsapi.SkillRegistry, seed map[string]interface{}) (err error) { // One identity for this run, minted before any work happens so every // approval and audit row produced below can be joined back to it. - runID := newRunID(time.Now()) + startedAt := time.Now() + runID := newRunID(startedAt) + runStore := state + defer func() { + if runStore != nil { + if recordErr := runStore.RecordRunForID(runID, pipeline.Name, startedAt, time.Now(), err); recordErr != nil { + log.Printf("[pipeline:%s] record run: %v", pipeline.Name, recordErr) + } + } + }() log.Printf("[pipeline:%s] starting (run %s)", pipeline.Name, runID) budget.tokensUsedPipeline = 0 budget.costPipeline = 0 @@ -2615,7 +2626,7 @@ func runAuditVerify(args []string) int { statePath = "./state.db" } - st, err := statestore.OpenStateStore(statePath) + st, err := statestore.OpenStateStoreReadOnly(statePath) if err != nil { fmt.Fprintf(os.Stderr, "open state store %s: %v\n", statePath, err) return 1 @@ -2647,6 +2658,9 @@ func main() { // Subcommand dispatch. The bare form `draftcat [config.yaml] [skills/]` still runs the engine. if len(os.Args) > 1 { switch os.Args[1] { + case "version", "--version", "-v": + fmt.Println("draftcat " + version) + return case "validate": os.Exit(validate.Run(os.Args[2:])) case "test": @@ -2674,7 +2688,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 receipts inspect, export, or verify 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") @@ -3217,56 +3231,66 @@ const webhookSigHeader = "X-Draftcat-Signature" // Errors are deliberately specific for the log and generic for the caller — the // handler returns a bare 401 so a prober learns nothing about which check failed. func verifyWebhookSignature(header string, body, secret []byte, maxSkewSeconds int64, now time.Time) error { + sig, err := authenticateWebhookSignature(header, body, secret, maxSkewSeconds, now) + if err != nil { + return err + } + return claimWebhookSignature(sig, now) +} + +func claimWebhookSignature(sig string, now time.Time) error { + if state != nil { + won, err := state.TryMarkSeen(webhookReplayScope, "sig", sig, now) + if err != nil { + return fmt.Errorf("replay record failed: %w", err) + } + if !won { + return fmt.Errorf("signature already used (replay)") + } + } + return nil +} + +func authenticateWebhookSignature(header string, body, secret []byte, maxSkewSeconds int64, now time.Time) (string, error) { if header == "" { - return fmt.Errorf("missing %s header", webhookSigHeader) + return "", fmt.Errorf("missing %s header", webhookSigHeader) } var tsPart, sigPart string for _, field := range strings.Split(header, ",") { field = strings.TrimSpace(field) switch { case strings.HasPrefix(field, "t="): + if tsPart != "" { + return "", fmt.Errorf("duplicate timestamp") + } tsPart = strings.TrimPrefix(field, "t=") case strings.HasPrefix(field, "v1="): + if sigPart != "" { + return "", fmt.Errorf("duplicate signature") + } sigPart = strings.TrimPrefix(field, "v1=") } } if tsPart == "" || sigPart == "" { - return fmt.Errorf("malformed signature header (want t=,v1=)") + return "", fmt.Errorf("malformed signature header") } ts, err := strconv.ParseInt(tsPart, 10, 64) if err != nil { - return fmt.Errorf("unparseable timestamp %q", tsPart) + return "", fmt.Errorf("unparseable timestamp") } - if skew := now.Unix() - ts; skew > maxSkewSeconds || skew < -maxSkewSeconds { - return fmt.Errorf("timestamp outside %ds window (skew %ds)", maxSkewSeconds, skew) + // Compare bounds instead of subtracting an attacker-controlled timestamp. + if ts < now.Unix()-maxSkewSeconds || ts > now.Unix()+maxSkewSeconds { + return "", fmt.Errorf("timestamp outside %ds window", maxSkewSeconds) } - mac := hmac.New(sha256.New, secret) mac.Write([]byte(tsPart)) mac.Write([]byte(".")) mac.Write(body) want := hex.EncodeToString(mac.Sum(nil)) if subtle.ConstantTimeCompare([]byte(want), []byte(sigPart)) != 1 { - return fmt.Errorf("signature mismatch") + return "", fmt.Errorf("signature mismatch") } - - // Replay guard. Within the skew window a valid signature is otherwise - // reusable, so each one is spent exactly once. Reuses the existing dedup - // table rather than adding a second store. With no state store configured - // the signature still authenticates — we just cannot promise single-use. - if state != nil { - unseen, ferr := state.FilterUnseen(webhookReplayScope, "sig", []string{sigPart}) - if ferr != nil { - return fmt.Errorf("replay check failed: %w", ferr) - } - if len(unseen) == 0 { - return fmt.Errorf("signature already used (replay)") - } - if merr := state.MarkSeen(webhookReplayScope, "sig", []string{sigPart}); merr != nil { - return fmt.Errorf("replay record failed: %w", merr) - } - } - return nil + return sigPart, nil } // webhookReplayScope namespaces spent webhook signatures in seen_items. The @@ -3298,6 +3322,7 @@ func newWebhookHandler(cfg *config.Config, sched *Scheduler, budget *BudgetTrack maxSkew = 300 } hookable := webhookPipelines(cfg) + var admissionMu sync.Mutex mux := http.NewServeMux() mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { @@ -3337,7 +3362,10 @@ func newWebhookHandler(cfg *config.Config, sched *Scheduler, budget *BudgetTrack return } if r.Method == http.MethodPost { - body, _ := io.ReadAll(io.LimitReader(r.Body, maxBody)) + body, ok := readRequestBody(w, r, maxBody) + if !ok { + return + } r.Body = io.NopCloser(bytes.NewReader(body)) sigHeader := r.Header.Get(webhookSigHeader) if sigHeader != "" || cfg.Webhook.RequireSignature { @@ -3403,16 +3431,21 @@ func newWebhookHandler(cfg *config.Config, sched *Scheduler, budget *BudgetTrack // The body must be read before the signature can be checked, since the // signature covers it — that binding is the point. - body, _ := io.ReadAll(io.LimitReader(r.Body, maxBody)) + body, ok := readRequestBody(w, r, maxBody) + if !ok { + return + } // A signature is verified whenever it is present, and demanded when // require_signature is on. Verifying an unrequested-but-present header // means a signer that starts emitting bad signatures fails loudly // instead of being silently ignored. sigHeader := r.Header.Get(webhookSigHeader) + var signature string if sigHeader != "" || cfg.Webhook.RequireSignature { - if err := verifyWebhookSignature(sigHeader, body, secret, maxSkew, time.Now()); err != nil { - log.Printf("[webhook][security] signature rejected: %v", err) + var err error + signature, err = authenticateWebhookSignature(sigHeader, body, secret, maxSkew, time.Now()) + if err != nil { http.Error(w, "unauthorized", http.StatusUnauthorized) return } @@ -3425,26 +3458,76 @@ func newWebhookHandler(cfg *config.Config, sched *Scheduler, budget *BudgetTrack return } + key := r.Header.Get("Idempotency-Key") + if key != "" && !validActionID(key) { + http.Error(w, "Idempotency-Key must be 1-128 URL-safe characters", http.StatusBadRequest) + return + } + keyHash := "" + if key != "" { + if state == nil { + http.Error(w, "state unavailable", http.StatusServiceUnavailable) + return + } + sum := sha256.Sum256([]byte(key)) + keyHash = hex.EncodeToString(sum[:]) + } + bodySum := sha256.Sum256(body) + bodyHash := "sha256:" + hex.EncodeToString(bodySum[:]) + admissionMu.Lock() + defer admissionMu.Unlock() + if keyHash != "" { + existing, err := state.WebhookAdmissionForKey(name, keyHash) + if err == nil { + if existing.BodyHash != bodyHash { + http.Error(w, "idempotency key is already bound to another request body", http.StatusConflict) + return + } + writeWebhookAdmission(w, existing) + return + } + if !errors.Is(err, sql.ErrNoRows) { + http.Error(w, "state unavailable", http.StatusServiceUnavailable) + return + } + } ok, reason := sched.TryStart(name) if !ok { http.Error(w, reason, http.StatusConflict) return } - + if signature != "" { + if err := claimWebhookSignature(signature, time.Now()); err != nil { + sched.SetRunning(name, false) + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + } admissionID := "wh_" + strings.TrimPrefix(newToolTicketID(), "tc_") - bodySum := sha256.Sum256(body) - bodyHash := "sha256:" + hex.EncodeToString(bodySum[:]) now := time.Now() + admission := statestore.WebhookAdmission{ID: admissionID, Pipeline: name, BodyHash: bodyHash, IdempotencyKey: keyHash, Status: "accepted", CreatedAt: now, UpdatedAt: now} if state != nil { - if err := state.BeginWebhookAdmission(statestore.WebhookAdmission{ - ID: admissionID, Pipeline: name, BodyHash: bodyHash, CreatedAt: now, UpdatedAt: now, - }); err != nil { + saved, created, err := state.AdmitWebhook(admission) + if err != nil { sched.SetRunning(name, false) log.Printf("[webhook] durable admission failed: %v", err) + if saved.ID != "" { + http.Error(w, "idempotency key is already bound to another request body", http.StatusConflict) + } else { + http.Error(w, "state unavailable", http.StatusServiceUnavailable) + } + return + } + if !created { + sched.SetRunning(name, false) + writeWebhookAdmission(w, saved) + return + } + if err := state.FinishWebhookAdmission(admissionID, "running", "", now); err != nil { + sched.SetRunning(name, false) 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)) @@ -3466,17 +3549,17 @@ func newWebhookHandler(cfg *config.Config, sched *Scheduler, budget *BudgetTrack sched.MarkRun(p.Name) }(p, body, admissionID) - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(http.StatusAccepted) - _ = json.NewEncoder(w).Encode(map[string]string{ - "admission_id": admissionID, - "status": "accepted", - "poll": "/hooks/status/" + admissionID, - }) + writeWebhookAdmission(w, admission) }) return mux } +func writeWebhookAdmission(w http.ResponseWriter, a statestore.WebhookAdmission) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusAccepted) + _ = json.NewEncoder(w).Encode(map[string]string{"admission_id": a.ID, "status": a.Status, "poll": "/hooks/status/" + a.ID}) +} + func resolveEnv(names ...string) string { for _, name := range names { if v := os.Getenv(name); v != "" { diff --git a/package-lock.json b/package-lock.json index 3308c29..5a4e570 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "draftcat", - "version": "0.7.0", + "version": "0.8.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "draftcat", - "version": "0.7.0", + "version": "0.8.0", "cpu": [ "x64", "arm64" diff --git a/package.json b/package.json index 7ae1261..f714cc3 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "draftcat", - "version": "0.7.0", + "version": "0.8.0", "description": "Governed AI pipelines with human approval gates", "license": "MIT", "author": "Rene Zander", diff --git a/receipts_cmd.go b/receipts_cmd.go index 62afd98..00cc854 100644 --- a/receipts_cmd.go +++ b/receipts_cmd.go @@ -65,7 +65,7 @@ func receiptView(r statestore.ApprovalRecord, secret []byte) receiptJSON { func runReceiptsCmd(args []string) int { if len(args) == 0 { - fmt.Fprintln(os.Stderr, "Usage: draftcat receipts [options]") + fmt.Fprintln(os.Stderr, "Usage: draftcat receipts [options]") return 2 } switch args[0] { @@ -75,11 +75,14 @@ func runReceiptsCmd(args []string) int { return runReceiptsShow(args[1:]) case "export": return runReceiptsExport(args[1:]) + case "verify": + return runReceiptsVerify(args[1:]) case "-h", "--help", "help": - fmt.Println("Usage: draftcat receipts [options]") + 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]") + fmt.Println(" verify [--json] check signed fields without opening SQLite") return 0 default: fmt.Fprintf(os.Stderr, "receipts: unknown command %q\n", args[0]) diff --git a/receipts_verify.go b/receipts_verify.go new file mode 100644 index 0000000..aa98151 --- /dev/null +++ b/receipts_verify.go @@ -0,0 +1,135 @@ +package main + +import ( + "bufio" + "encoding/json" + "fmt" + "io" + "os" + "strings" + "time" + + statestore "github.com/renezander030/draftcat/internal/state" +) + +type receiptCheck struct { + Line int `json:"line"` + ReceiptID string `json:"receipt_id"` + Verification string `json:"verification"` +} + +func exportedReceiptRecord(v receiptJSON) (statestore.ApprovalRecord, error) { + if v.Version != 1 && v.Version != 2 { + return statestore.ApprovalRecord{}, fmt.Errorf("unsupported receipt version %d", v.Version) + } + decided, err := time.Parse(time.RFC3339, v.DecidedAt) + if err != nil { + return statestore.ApprovalRecord{}, fmt.Errorf("invalid decided_at: %w", err) + } + var expires time.Time + if v.ExpiresAt != "" { + expires, err = time.Parse(time.RFC3339, v.ExpiresAt) + if err != nil { + return statestore.ApprovalRecord{}, fmt.Errorf("invalid expires_at: %w", err) + } + } + return statestore.ApprovalRecord{ + Version: v.Version, ReceiptID: v.ReceiptID, RunID: v.RunID, ActionID: v.ActionID, + Pipeline: v.Pipeline, Step: v.Step, DecidedAt: decided, Decision: v.Decision, + OperatorID: v.OperatorID, PayloadHash: v.PayloadHash, Policy: v.Policy, PolicyHash: v.PolicyHash, + BindingHash: v.BindingHash, ExpiresAt: expires, Lifecycle: v.Lifecycle, + QuorumN: v.QuorumN, QuorumGot: v.QuorumGot, Nonce: v.Nonce, Signature: v.Signature, + }, nil +} + +func verifyReceiptStream(src io.Reader, secret []byte) ([]receiptCheck, error) { + scanner := bufio.NewScanner(src) + scanner.Buffer(make([]byte, 4096), 1<<20) + checks := []receiptCheck{} + for line := 1; scanner.Scan(); line++ { + if strings.TrimSpace(scanner.Text()) == "" { + continue + } + var v receiptJSON + if err := decodeStrictJSON(scanner.Bytes(), &v); err != nil { + return checks, fmt.Errorf("line %d: %w", line, err) + } + record, err := exportedReceiptRecord(v) + if err != nil { + return checks, fmt.Errorf("line %d: %w", line, err) + } + verdict := statestore.VerifyApprovalRecord(secret, record) + if v.Signature != "" && len(secret) == 0 { + verdict = "unverified" + } + checks = append(checks, receiptCheck{Line: line, ReceiptID: v.ReceiptID, Verification: verdict}) + } + if err := scanner.Err(); err != nil { + return checks, err + } + if len(checks) == 0 { + return checks, fmt.Errorf("no receipts in input") + } + return checks, nil +} + +func runReceiptsVerify(args []string) int { + path := "" + jsonOut := false + for _, arg := range args { + switch arg { + case "--json": + jsonOut = true + case "--help", "-h": + fmt.Println("Usage: draftcat receipts verify [--json]") + return 0 + default: + if path != "" || (strings.HasPrefix(arg, "-") && arg != "-") { + fmt.Fprintln(os.Stderr, "receipts verify: expected one JSONL file or -") + return 2 + } + path = arg + } + } + if path == "" { + fmt.Fprintln(os.Stderr, "Usage: draftcat receipts verify [--json]") + return 2 + } + secret := []byte(os.Getenv("DRAFTCAT_APPROVAL_SECRET")) + if len(secret) == 0 { + fmt.Fprintln(os.Stderr, "DRAFTCAT_APPROVAL_SECRET is not set — cannot verify receipts") + return 2 + } + var src io.Reader = os.Stdin + if path != "-" { + f, err := os.Open(path) + if err != nil { + fmt.Fprintf(os.Stderr, "receipts verify: %v\n", err) + return 1 + } + defer func() { _ = f.Close() }() + src = f + } + checks, err := verifyReceiptStream(src, secret) + if err != nil { + fmt.Fprintf(os.Stderr, "receipts verify: %v\n", err) + return 1 + } + code := 0 + for _, c := range checks { + if c.Verification != "ok" { + code = 1 + } + } + if jsonOut { + if err := json.NewEncoder(os.Stdout).Encode(checks); err != nil { + fmt.Fprintf(os.Stderr, "receipts verify: %v\n", err) + return 1 + } + } else { + for _, c := range checks { + fmt.Printf("%-9s line=%d receipt=%s\n", c.Verification, c.Line, c.ReceiptID) + } + } + return code +} diff --git a/reliability_test.go b/reliability_test.go new file mode 100644 index 0000000..a074010 --- /dev/null +++ b/reliability_test.go @@ -0,0 +1,290 @@ +package main + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/renezander030/draftcat/internal/config" + statestore "github.com/renezander030/draftcat/internal/state" +) + +func useReliabilityStore(t *testing.T) *statestore.StateStore { + t.Helper() + st := newTempStateStore(t) + prev := state + state = st + t.Cleanup(func() { state = prev }) + return st +} + +func TestWebhookKeyRetriesReturnSameAdmission(t *testing.T) { + st := useReliabilityStore(t) + h, sched := testHandler(t) + request := func(body, key string) *httptest.ResponseRecorder { + r := httptest.NewRequest(http.MethodPost, "/hooks/ping", strings.NewReader(body)) + r.Header.Set("Authorization", "Bearer s3cret") + r.Header.Set("Idempotency-Key", key) + r.Header.Set(webhookSigHeader, signBody([]byte("s3cret"), time.Now().Unix(), []byte(body))) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, r) + return rec + } + first := request(`{"amount":1}`, "invoice-1") + if first.Code != 202 { + t.Fatalf("first=%d %s", first.Code, first.Body) + } + var initial map[string]string + if err := json.Unmarshal(first.Body.Bytes(), &initial); err != nil { + t.Fatal(err) + } + var wg sync.WaitGroup + for i := 0; i < 12; i++ { + wg.Add(1) + go func() { + defer wg.Done() + retry := request(`{"amount":1}`, "invoice-1") + var v map[string]string + if err := json.Unmarshal(retry.Body.Bytes(), &v); err != nil { + t.Error(err) + } + if retry.Code != 202 || v["admission_id"] != initial["admission_id"] { + t.Errorf("retry=%d %s", retry.Code, retry.Body) + } + }() + } + wg.Wait() + if changed := request(`{"amount":2}`, "invoice-1"); changed.Code != 409 { + t.Fatalf("changed=%d", changed.Code) + } + waitFor(t, "pipeline completion", func() bool { + for _, p := range sched.GetAll() { + if p.Name == "ping" { + return !p.Running + } + } + return false + }) + runs, err := st.AllRecentRuns(100) + if err != nil || len(runs) != 1 { + t.Fatalf("executed runs=%d err=%v", len(runs), err) + } +} + +type brokenBody struct{} + +func (brokenBody) Read([]byte) (int, error) { return 0, errors.New("read failed") } +func (brokenBody) Close() error { return nil } + +func TestRequestBoundariesDoNotAdmitTruncatedData(t *testing.T) { + useReliabilityStore(t) + cfg := gateCfg(config.ToolRule{Name: "send"}) + cfg.Webhook.MaxBodyBytes = 32 + cfg.Webhook.SetSecret("s3cret") + cfg.Pipelines = []config.PipelineConfig{{Name: "ping", Schedule: "webhook"}} + h := newWebhookHandler(cfg, newScheduler(cfg.Pipelines), &BudgetTracker{}, &TGBot{}, nil) + for _, route := range []string{"/hooks/ping", toolGatePath, toolGatePath + "/a/consume"} { + req := httptest.NewRequest(http.MethodPost, route, strings.NewReader(strings.Repeat("x", 33))) + req.Header.Set("Authorization", "Bearer s3cret") + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != 413 { + t.Errorf("%s oversize=%d", route, rec.Code) + } + req = httptest.NewRequest(http.MethodPost, route, nil) + req.Body = brokenBody{} + req.Header.Set("Authorization", "Bearer s3cret") + rec = httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != 400 { + t.Errorf("%s unreadable=%d", route, rec.Code) + } + } + // Direct gate use has its own independent bound as well. + g := newToolGate(cfg, nil) + rec := httptest.NewRecorder() + g.HandleCall(rec, httptest.NewRequest(http.MethodPost, toolGatePath, strings.NewReader(strings.Repeat("x", (1<<20)+1)))) + if rec.Code != 413 { + t.Fatalf("direct gate oversize=%d", rec.Code) + } +} + +func TestStrictToolRequestDoesNotSpendPermitOnMalformedConsume(t *testing.T) { + useReliabilityStore(t) + g := newToolGate(gateCfg(config.ToolRule{Name: "send"}), nil) + for _, body := range []string{`{"tool":"send","typo":true}`, `{"tool":"send"} {}`, `{"tool":"send","args":{"id":1,"id":2}}`, `{"tool":"send","tool":"other"}`} { + rec := httptest.NewRecorder() + g.HandleCall(rec, httptest.NewRequest(http.MethodPost, toolGatePath, strings.NewReader(body))) + if rec.Code != 400 { + t.Errorf("malformed request %s accepted: %d", body, rec.Code) + } + } + _, allowed := postGate(t, g, `{"action_id":"a","tool":"send"}`) + for _, suffix := range []string{` {}`, `,"extra":true}`, `,"binding_hash":"other"}`} { + body := fmt.Sprintf(`{"binding_hash":%q}`, allowed.BindingHash) + if strings.HasPrefix(suffix, ",") { + body = strings.TrimSuffix(body, "}") + suffix + } else { + body += suffix + } + rec := httptest.NewRecorder() + g.HandleStatus(rec, httptest.NewRequest(http.MethodPost, toolGatePath+"/a/consume", strings.NewReader(body))) + if rec.Code != 400 { + t.Errorf("malformed consume %s accepted: %d", body, rec.Code) + } + } + if code, resp := consumeGate(t, g, "a", allowed.BindingHash); code != 200 || resp.Permit != "execute" { + t.Fatalf("valid permit lost: %d %+v", code, resp) + } +} + +func TestLargeToolArgumentsHaveDifferentIdentities(t *testing.T) { + useReliabilityStore(t) + g := newToolGate(gateCfg(config.ToolRule{Name: "send"}), nil) + _, a := postGate(t, g, `{"action_id":"same","tool":"send","args":{"id":9007199254740992}}`) + code, _ := postGate(t, g, `{"action_id":"same","tool":"send","args":{"id":9007199254740993}}`) + if code != 409 { + t.Fatalf("large integer collision accepted: %d", code) + } + _, b := postGate(t, g, `{"action_id":"different","tool":"send","args":{"id":9007199254740993}}`) + if a.ArgsHash == b.ArgsHash { + t.Fatal("different exact integers share a hash") + } +} + +func TestPermitRejectsChangedPolicyAfterRestart(t *testing.T) { + useReliabilityStore(t) + for _, change := range []string{"rule", "operators", "channel", "window"} { + t.Run(change, func(t *testing.T) { + cfg := gateCfg(config.ToolRule{Name: "send"}) + g := newToolGate(cfg, nil) + _, a := postGate(t, g, fmt.Sprintf(`{"action_id":%q,"tool":"send"}`, change)) + switch change { + case "rule": + cfg.ToolGate.Tools[0].RequireApproval = true + case "operators": + cfg.Telegram.Security.AllowedUsers = []int64{42} + case "channel": + cfg.Relay.URL = "https://relay.example/dispatch" + case "window": + cfg.Timeouts.OperatorApproval = "1s" + } + restarted := newToolGate(cfg, nil) + if code, resp := consumeGate(t, restarted, a.ActionID, a.BindingHash); code != 409 || resp.Permit != "" || resp.Decision != "deny" { + t.Fatalf("changed policy consumed: %d %+v", code, resp) + } + }) + } +} + +func TestEnginePersistsRunsAndExactApprovalJoin(t *testing.T) { + st := useReliabilityStore(t) + cfg := &config.Config{} + if err := runPipeline(cfg, config.PipelineConfig{Name: "zero"}, &BudgetTracker{}, nil, nil, nil); err != nil { + t.Fatal(err) + } + runs, err := st.AllRecentRuns(10) + if err != nil || len(runs) != 1 || runs[0].RunID == "" { + t.Fatalf("engine run=%+v err=%v", runs, err) + } + at := runs[0].StartedAt + for _, id := range []string{runs[0].RunID, "different-run", ""} { + if err := st.RecordApprovalForRun(id, "zero", id, at, "approve", 7, "hash", 1, 1, "", ""); err != nil { + t.Fatal(err) + } + } + joined := approvalsDuring(st, runs[0]) + if len(joined) != 1 || joined[0].Step != runs[0].RunID { + t.Fatalf("inexact correlation=%+v", joined) + } + // A budget-blocked run must leave a terminal error record too. + cfg.Budgets.PerStepTokens = 1 + err = runPipeline(cfg, config.PipelineConfig{Name: "failure", Steps: []config.StepConfig{{Name: "ai", Type: "ai", Skill: "missing"}}}, &BudgetTracker{}, nil, nil, nil) + if err == nil { + t.Fatal("expected budget failure") + } + failures, err := st.RecentRuns("failure", 10) + if err != nil || len(failures) != 1 || failures[0].Status != "error" { + t.Fatalf("failed run=%+v err=%v", failures, err) + } +} + +func TestOfflineReceiptVerificationDoesNotTrustExportVerdict(t *testing.T) { + st := useReliabilityStore(t) + secret := []byte("secret") + now := time.Now() + e := newApprovalEnvelope(secret, "r", "p", config.StepConfig{Name: "send"}, now, now.Add(time.Hour), "approve", 42, "hash", 1, 1, "human") + if err := st.RecordApprovalV2(e); err != nil { + t.Fatal(err) + } + rows, err := st.AllApprovals(10) + if err != nil { + t.Fatal(err) + } + v := receiptView(rows[0], nil) + v.Verification = "tampered" + raw, err := json.Marshal(v) + if err != nil { + t.Fatal(err) + } + checks, err := verifyReceiptStream(bytes.NewReader(raw), secret) + if err != nil || len(checks) != 1 || checks[0].Verification != "ok" { + t.Fatalf("intact=%+v err=%v", checks, err) + } + v.PayloadHash = "edited" + v.Verification = "ok" + raw, _ = json.Marshal(v) + checks, err = verifyReceiptStream(bytes.NewReader(raw), secret) + if err != nil || checks[0].Verification != "tampered" { + t.Fatalf("edited=%+v err=%v", checks, err) + } + v.Signature = "" + raw, _ = json.Marshal(v) + checks, err = verifyReceiptStream(bytes.NewReader(raw), secret) + if err != nil || checks[0].Verification != "unsigned" { + t.Fatalf("unsigned=%+v err=%v", checks, err) + } + for _, bad := range []string{"", string(raw) + " {}", `{"version":99,"decided_at":"2026-10-01T00:00:00Z"}`} { + if _, err := verifyReceiptStream(strings.NewReader(bad), secret); err == nil { + t.Errorf("bad input accepted: %q", bad) + } + } + // The CLI must work with no decision database and return nonzero on unsigned rows. + t.Setenv("DRAFTCAT_STATE_PATH", filepath.Join(t.TempDir(), "absent.db")) + t.Setenv("DRAFTCAT_APPROVAL_SECRET", string(secret)) + path := filepath.Join(t.TempDir(), "receipts.jsonl") + if err := os.WriteFile(path, raw, 0600); err != nil { + t.Fatal(err) + } + if code := runReceiptsVerify([]string{path, "--json"}); code != 1 { + t.Fatalf("unsigned CLI exit=%d", code) + } + if _, err := os.Stat(os.Getenv("DRAFTCAT_STATE_PATH")); !os.IsNotExist(err) { + t.Fatal("offline verifier touched database") + } +} + +func TestStrictJSONDepthBound(t *testing.T) { + body := []byte(`{"tool":"send","args":{"nested":` + strings.Repeat("[", 150) + "0" + strings.Repeat("]", 150) + "}}") + var req ToolCallRequest + if err := decodeStrictJSON(body, &req); err == nil { + t.Fatal("unbounded nesting accepted") + } +} + +func TestUnknownConsumeRemainsStructured(t *testing.T) { + useReliabilityStore(t) + g := newToolGate(gateCfg(config.ToolRule{Name: "send"}), nil) + if code, resp := consumeGate(t, g, "absent", "binding"); code != 404 || resp.Decision != "deny" { + t.Fatalf("unknown consume=%d %+v", code, resp) + } +} diff --git a/request_body.go b/request_body.go new file mode 100644 index 0000000..c6849f7 --- /dev/null +++ b/request_body.go @@ -0,0 +1,82 @@ +package main + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" +) + +func readRequestBody(w http.ResponseWriter, r *http.Request, limit int64) ([]byte, bool) { + body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, limit)) + if err != nil { + var tooLarge *http.MaxBytesError + if errors.As(err, &tooLarge) { + http.Error(w, "request body too large", http.StatusRequestEntityTooLarge) + } else { + http.Error(w, "unreadable body", http.StatusBadRequest) + } + return nil, false + } + return body, true +} + +// decodeStrictJSON preserves number tokens and rejects ambiguous envelopes. +func decodeStrictJSON(body []byte, target interface{}) error { + tokens := json.NewDecoder(bytes.NewReader(body)) + tokens.UseNumber() + if err := uniqueJSONValue(tokens, 0); err != nil { + return err + } + if _, err := tokens.Token(); err != io.EOF { + return fmt.Errorf("trailing JSON content") + } + dec := json.NewDecoder(bytes.NewReader(body)) + dec.UseNumber() + dec.DisallowUnknownFields() + return dec.Decode(target) +} + +func uniqueJSONValue(dec *json.Decoder, depth int) error { + if depth > 128 { + return fmt.Errorf("JSON nesting exceeds 128 levels") + } + token, err := dec.Token() + if err != nil { + return err + } + switch token { + case json.Delim('{'): + keys := map[string]bool{} + for dec.More() { + token, err := dec.Token() + if err != nil { + return err + } + key, ok := token.(string) + if !ok { + return fmt.Errorf("object key must be a string") + } + if keys[key] { + return fmt.Errorf("duplicate JSON key %q", key) + } + keys[key] = true + if err := uniqueJSONValue(dec, depth+1); err != nil { + return err + } + } + _, err = dec.Token() + return err + case json.Delim('['): + for dec.More() { + if err := uniqueJSONValue(dec, depth+1); err != nil { + return err + } + } + _, err = dec.Token() + return err + } + return nil +} diff --git a/runs_cmd.go b/runs_cmd.go index 8eb19db..8c9d7ec 100644 --- a/runs_cmd.go +++ b/runs_cmd.go @@ -33,6 +33,7 @@ import ( // was decided, by whom. type runJSON struct { + RunID string `json:"run_id,omitempty"` Pipeline string `json:"pipeline"` StartedAt string `json:"started_at"` EndedAt string `json:"ended_at"` @@ -120,6 +121,7 @@ func runRunsCmd(args []string) int { out := make([]runJSON, 0, len(runs)) for _, r := range runs { out = append(out, runJSON{ + RunID: r.RunID, Pipeline: r.Pipeline, StartedAt: r.StartedAt.Format(time.RFC3339), EndedAt: r.EndedAt.Format(time.RFC3339), @@ -170,18 +172,22 @@ func runRunsCmd(args []string) int { return 0 } -// approvalsDuring attaches the approval decisions recorded inside a run's -// window. There is no run_id on action_approvals, but the scheduler refuses to -// start a pipeline that is already running, so runs of one pipeline never -// overlap and the timestamp window is an unambiguous join. +// approvalsDuring joins by exact run ID. Historical rows without an identity +// retain the legacy timestamp join, restricted to other identity-less rows. func approvalsDuring(st *statestore.StateStore, r statestore.RunRecord) []approvalJSON { - recs, err := st.ApprovalsForPipeline(r.Pipeline, 1000) + var recs []statestore.ApprovalRecord + var err error + if r.RunID != "" { + recs, err = st.ApprovalsForRun(r.RunID) + } else { + recs, err = st.ApprovalsForPipeline(r.Pipeline, 1000) + } if err != nil { return nil } var out []approvalJSON for _, a := range recs { - if a.DecidedAt.Before(r.StartedAt) || a.DecidedAt.After(r.EndedAt) { + if r.RunID == "" && (a.RunID != "" || a.DecidedAt.Before(r.StartedAt) || a.DecidedAt.After(r.EndedAt)) { continue } out = append(out, approvalJSON{ @@ -199,7 +205,7 @@ func approvalsDuring(st *statestore.StateStore, r statestore.RunRecord) []approv } // openStateForCmd resolves the state path exactly like the engine does — env -// override, then config, then ./state.db — and opens it read-only enough for a +// override, then config, then ./state.db — and opens it read-only for a // reporting command. func openStateForCmd(configPath string) (*statestore.StateStore, func(), int) { statePath := strings.TrimSpace(os.Getenv("DRAFTCAT_STATE_PATH")) @@ -217,7 +223,7 @@ func openStateForCmd(configPath string) (*statestore.StateStore, func(), int) { if statePath == "" { statePath = "./state.db" } - st, err := statestore.OpenStateStore(statePath) + st, err := statestore.OpenStateStoreReadOnly(statePath) if err != nil { fmt.Fprintf(os.Stderr, "runs: open state store %s: %v\n", statePath, err) return nil, func() {}, 1 diff --git a/tool_gate.go b/tool_gate.go index 5c8d110..6dd1432 100644 --- a/tool_gate.go +++ b/tool_gate.go @@ -44,7 +44,6 @@ import ( "encoding/json" "errors" "fmt" - "io" "log" "net/http" "os" @@ -261,13 +260,12 @@ func (g *toolGate) HandleCall(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusMethodNotAllowed) return } - body, err := io.ReadAll(io.LimitReader(r.Body, 1<<20)) - if err != nil { - http.Error(w, "unreadable body", http.StatusBadRequest) + body, ok := readRequestBody(w, r, 1<<20) + if !ok { return } var req ToolCallRequest - if err := json.Unmarshal(body, &req); err != nil { + if err := decodeStrictJSON(body, &req); err != nil { http.Error(w, "malformed json", http.StatusBadRequest) return } @@ -305,7 +303,7 @@ 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) + policyHash := g.policyHash(req.Tool, rule, listed) expires := g.now().Add(g.approvalWindow()) if strings.TrimSpace(req.ExpiresAt) != "" { requested, err := time.Parse(time.RFC3339, req.ExpiresAt) @@ -541,12 +539,40 @@ func (g *toolGate) handleConsume(w http.ResponseWriter, r *http.Request, id stri var body struct { BindingHash string `json:"binding_hash"` } - if err := json.NewDecoder(io.LimitReader(r.Body, 4096)).Decode(&body); err != nil { + raw, ok := readRequestBody(w, r, 4096) + if !ok { + return + } + if err := decodeStrictJSON(raw, &body); err != nil { http.Error(w, "malformed json", http.StatusBadRequest) return } now := g.now() if state != nil { + saved, err := state.ToolAction(id) + 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 saved.BindingHash != body.BindingHash { + writeToolDecision(w, http.StatusConflict, responseFromToolAction(saved)) + return + } + rule, listed := g.cfg.ToolGate.Lookup(saved.Tool) + if saved.PolicyHash != g.policyHash(saved.Tool, rule, listed) { + if err := state.RevokeToolAction(id, saved.PolicyHash, now); err != nil { + http.Error(w, "state unavailable", http.StatusServiceUnavailable) + return + } + resp := responseFromToolAction(saved) + resp.Decision, resp.State, resp.Permit, resp.Consume, resp.Reason = "deny", "expired", "", "", "policy changed; request a new approval" + writeToolDecision(w, http.StatusConflict, resp) + return + } a, consumed, err := state.ConsumeToolAction(id, body.BindingHash, now) if err != nil { if errors.Is(err, sql.ErrNoRows) { @@ -579,6 +605,13 @@ func (g *toolGate) handleConsume(w http.ResponseWriter, r *http.Request, id stri writeToolDecision(w, http.StatusNotFound, ToolCallResponse{ActionID: id, Decision: "deny", State: "denied", Reason: "unknown action id"}) return } + rule, listed := g.cfg.ToolGate.Lookup(tk.Tool) + if tk.PolicyHash != g.policyHash(tk.Tool, rule, listed) { + resp, _ := tk.result() + resp.Decision, resp.State, resp.Permit, resp.Consume, resp.Reason = "deny", "expired", "", "", "policy changed; request a new approval" + writeToolDecision(w, http.StatusConflict, resp) + return + } resp, ok := tk.consume(body.BindingHash, now) if !ok { resp.Permit = "" @@ -838,6 +871,18 @@ func validActionID(id string) bool { return true } +func (g *toolGate) policyHash(tool string, rule config.ToolRule, listed bool) string { + b, _ := json.Marshal(struct { + ToolPolicy string + Enabled bool + ApprovalWindow int64 + Telegram config.TelegramConfig + Relay config.RelayConfig + }{hashToolPolicy(tool, rule, listed), g.cfg.ToolGate.Enabled, int64(g.approvalWindow()), g.cfg.Telegram, g.cfg.Relay}) + sum := sha256.Sum256(b) + return "sha256:" + hex.EncodeToString(sum[:]) +} + func hashToolPolicy(tool string, rule config.ToolRule, listed bool) string { b, _ := json.Marshal(struct { Tool string `json:"tool"` diff --git a/version.go b/version.go new file mode 100644 index 0000000..35739ef --- /dev/null +++ b/version.go @@ -0,0 +1,4 @@ +package main + +// version is overridden by the native release build. +var version = "0.8.0"