diff --git a/CHANGELOG.md b/CHANGELOG.md index 90ef469..06841b3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,27 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Removed + +- Governed delivery. The signed-envelope frames `TypeGoverned` (8) and + `TypeGovernedFileStream` (9) carried decisions issued by the hosted control + plane, which has been retired; the daemon stopped configuring the receiver + gates in v1.14.0, so the code could no longer be reached. Removed: every + exported `Governed*` identifier (the frame and stream-INIT envelopes and + their codecs, the verifier and receipt-recorder interfaces, the payload + hash helpers, `GovernedRetentionPolicy`), `DecisionFrameVerifier`, + `Client.SendGoverned*`, `BuildStreamInitPayload`, and the `ServiceConfig` + fields `RequireGoverned`, `GovernedVerifier`, `GovernedStreamVerifier`, + `RequireGovernedReceipts`, `GovernedReceiptRecorder`, + `GovernedContentInspector`, `RequireGovernedContentInspection`, + `GovernedTransferQuota`, `GovernedRetentionPolicies`, `RetentionStateDir` + and `RetentionSweepInterval`. The module no longer imports + `github.com/pilot-protocol/common/decision`. +- Frame types 8 and 9 stay reserved and keep their constants and names. A + receiver refuses them like any unsupported type: nothing is stored and the + sender gets `ERR GOVERNED save failed: unsupported frame type 8` (or + `ERR GOVERNED_FILESTREAM ... type 9`); the connection stays open. + ### Fixed - The inbox byte cap no longer deletes the whole inbox. With the default diff --git a/README.md b/README.md index 1f3212a..710d1e9 100644 --- a/README.md +++ b/README.md @@ -30,10 +30,9 @@ rt.Register(dataexchange.NewService(dataexchange.ServiceConfig{})) | File | What it does | |---|---| -| `dataexchange.go` | Wire format: `Frame`, `WriteFrame`, `ReadFrame`, `TraceFrame`, `TypeText/Binary/JSON/File/Trace/Governed/Tagged`, `TypeName`. | +| `dataexchange.go` | Wire format: `Frame`, `WriteFrame`, `ReadFrame`, `TraceFrame`, `TypeText/Binary/JSON/File/Trace/Tagged`, `TypeName`. | | `tagged.go` | Optional request/reply correlation on the wire: `Frame.MessageID` / `Frame.ReplyTo` (`TypeTagged`), `NewMessageID`, `ValidMessageID`. | | `client.go` | `Client` — `Dial`, `Send` (waits for the ACK, falls back for older receivers) and send helpers. | -| `governed.go` | Signed decision envelope, receiver-side verifier, and enforceable transport constraints. | | `server.go` | `Server` — accept loop and handler dispatch. | | `service.go` | `*Service` — `coreapi.Service` adapter. Build tag `!no_dataexchange`. | | `inbox_budget.go` | Inbox caps: running byte total, oldest-first eviction to 90% of the byte cap, file-count cap. | @@ -96,8 +95,8 @@ a frame can carry two optional fields: | `Frame.ReplyTo` | `reply_to` | The `message_id` of the message this one answers. | IDs are 1–128 characters of `A-Z a-z 0-9 - _ . :`. They are correlation -metadata only: they are not authenticated (a governed frame's signature does -not cover them), so never use them for authorization. +metadata only: they are not authenticated, so never use them for +authorization. **If you run a service** (an agent that answers requests): when a request's inbox record (or `message.received` event) has a `message_id`, send your @@ -147,8 +146,8 @@ untagged (`SendResult.Tagged == false`), and a raw `WriteFrame` returns `ErrTaggedFrameTooLarge` without writing anything. Whenever `Tagged` is false, the reply cannot carry `reply_to`; match it on -the sender alone. The fields can accompany text, JSON, binary, file, -governed and trace frames. On `TypeFileStream` frames they are ignored. +the sender alone. The fields can accompany text, JSON, binary, file and +trace frames. On `TypeFileStream` frames they are ignored. Wire layout of a tagged frame, for other languages: @@ -169,89 +168,19 @@ without a `MessageID` are always stored, as before, unless the operator sets `ServiceConfig.DedupeContentWindow` (off by default) to also drop byte-identical copies from the same sender inside a short window. -## Governed delivery - -`TypeGoverned` wraps one text, JSON, binary, or single-frame file delivery in -the sender's signed `decision.Intent` and signed `decision.Decision`. The -intent payload hash binds the frame type, filename, and exact bytes; the -receiver verifies the signatures, tenant authority state, local deterministic -ceiling, exact local destination, and any applicable transport constraints -before it writes to disk. A workflow-approved action uses the same short-lived -execution Decision as an ordinary allowed action—there is no reusable -transport permit. - -Large resumable files use `TypeGovernedFileStream`: the signed envelope binds -the exact `TypeFileStream` INIT (filename, declared length, full SHA-256, -chunk size, and transfer ID). Required receivers admit later chunks only for -that verified transfer on the same connection. Compute the file hash first, -call `BuildStreamInitPayload`, sign an Intent using -`GovernedStreamPayloadHash`, then call `SendGovernedFileStream`. - -For a required typed-disclosure profile, build a `decision.DisclosureBinding` -whose content hash, byte length, filename, and stream transfer ID match the -file, bind its canonical hash in the signed Intent, and use -`SendGovernedWithDisclosure` or `SendGovernedFileStreamWithDisclosure`. A -`DecisionFrameVerifier` with `RequireDisclosure` rejects governed messages, -single-frame files, and resumable stream INITs that omit this evidence. -When a receipt recorder is configured, typed deliveries require its V2 -disclosure-evidence method; the resulting signed receipt binds the canonical -disclosure hash without retaining the file or message body. - -Set `ServiceConfig.GovernedVerifier` to `DecisionFrameVerifier` (or an -equivalent local verifier). Once all senders have been upgraded, set -`ServiceConfig.RequireGoverned` to reject unsigned legacy deliveries. Roll out -in that order: upgraded receiver with verification available, upgraded -senders, then required mode. An older receiver does not understand -`TypeGoverned`; a required receiver intentionally rejects `TypeTrace` and -raw `TypeFileStream` INIT frames. Governed stream INITs are supported. - -For auditable enterprise ingress, configure `GovernedReceiptRecorder` and set -`RequireGovernedReceipts`. The service writes the received message/file first, -then requires the recorder to durably capture the exact signed Intent and -Decision before it emits the success ACK or delivery event. If recording fails, -the staged file is removed and the sender receives an error rather than a -successful but unreceipted delivery. - -### Local content inspection - -`ServiceConfig.GovernedContentInspector` is an optional receiver-local hook -that runs after the signed envelope and local policy ceiling are verified but -before a message/file is released. Set -`RequireGovernedContentInspection` to make startup fail unless the hook is -present; an inspection error removes a staged file or rejects the message. -`decision.PresidioInspector` is the included OSS adapter for bounded text, -JSON, XML, YAML, and form content. It rejects unsupported binary/document -types rather than truncating or silently skipping them. The inspector is local -to the receiver; neither the decision authority nor the sender's authority -receives plaintext for this check. - -Typed disclosure binding V2 adds a tenant-defined `retention_class` to the -same Intent hash. A signed policy can select allowed classes; the receiver sees -the bound metadata. Configure `GovernedRetentionPolicies` to map those classes -to local expiry durations. The service writes an owner-only retention journal -before a governed message/file becomes accepted (and before a streamed file's -final rename), then removes the content after expiry across restarts. An -unknown, V1, or unconfigured class is rejected when retention is enabled. -This is deletion retention, not a legal-hold or WORM-storage implementation. - -### Per-agent transfer quotas - -`ServiceConfig.GovernedTransferQuota` admits a bounded number of bytes and/or -actions for each signed `Intent.AgentID` in a fixed local window. It is charged -only after governed verification, including the declared bytes of a verified -stream INIT; a peer address cannot select or reset another agent's budget. -Quota is deliberately charged for an admitted attempt even if later local DLP -or receipt persistence rejects it, so repeatedly failing submissions cannot -turn the scanner into an unmetered denial-of-service target. - -The v1 action mapping is: - -| Frame | Intent action | -|---|---| -| text | `data.send.text` | -| JSON | `data.send.json` | -| binary | `data.send.binary` | -| file | `file.share` | +## Reserved frame types + +Types 8 (`TypeGoverned`) and 9 (`TypeGovernedFileStream`) carried deliveries +wrapped in an envelope signed by the hosted control plane. That control plane +has been retired and the envelope code was removed with it. The two numbers +stay reserved and will not be given to a new frame type, because senders built +before the removal can still emit them. + +A receiver treats both like any frame type it does not implement: it stores +nothing, publishes no event, answers +`ERR GOVERNED save failed: unsupported frame type 8` (or +`ERR GOVERNED_FILESTREAM save failed: unsupported frame type 9`), and keeps +the connection open for further frames. ## Build tags diff --git a/client.go b/client.go index 0ab2659..c5c8359 100644 --- a/client.go +++ b/client.go @@ -9,7 +9,6 @@ import ( "strings" "time" - "github.com/pilot-protocol/common/decision" "github.com/pilot-protocol/common/driver" "github.com/pilot-protocol/common/protocol" ) @@ -65,8 +64,9 @@ const duplicateAckSuffix = " (duplicate)" // "ACK UNKNOWN(10) bytes": a success ACK for a frame it dropped. // - dataexchange v0.2.2 and later, before TypeTagged, reject an unknown // type and answer "ERR UNKNOWN(10) save failed: unsupported frame type 10". -// A RequireGoverned receiver answers "ERR UNKNOWN(10) save failed: -// unsigned legacy frame ..." instead, also without storing it. +// A receiver of that era configured to require governed envelopes (since +// removed) answers "ERR UNKNOWN(10) save failed: unsigned legacy frame ..." +// instead, also without storing it. // // A receiver that knows TypeTagged unwraps the frame in ReadFrame and // acknowledges the inner type (which can never be TypeTagged), so its ACK @@ -177,113 +177,6 @@ func (c *Client) SendFile(name string, data []byte) error { return WriteFrame(c.conn, &Frame{Type: TypeFile, Filename: name, Payload: data}) } -// SendGoverned sends a frame with exact signed intent/decision evidence. -// The remote service must have RequireGoverned enabled to enforce it. -func (c *Client) SendGoverned(frame *Frame, intent decision.Intent, result decision.Decision) error { - governed, err := NewGovernedFrame(frame, intent, result) - if err != nil { - return err - } - envelope, err := EncodeGovernedFrame(governed) - if err != nil { - return err - } - return WriteFrame(c.conn, envelope) -} - -// SendGovernedWithDisclosure sends a single governed message or file with a -// typed disclosure binding. The Intent and Decision must already have been -// obtained for the exact binding. -func (c *Client) SendGovernedWithDisclosure(frame *Frame, intent decision.Intent, result decision.Decision, disclosure decision.DisclosureBinding) error { - governed, err := NewGovernedFrameWithDisclosure(frame, intent, result, disclosure) - if err != nil { - return err - } - envelope, err := EncodeGovernedFrame(governed) - if err != nil { - return err - } - return WriteFrame(c.conn, envelope) -} - -// SendGovernedFileStream sends a large resumable file with a signed authority -// decision bound to its exact INIT metadata. Callers compute the intent payload -// hash with GovernedStreamPayloadHash over the deterministic INIT payload (use -// BuildStreamInitPayload after calculating the file hash), then pass the same -// signed intent and decision here. The receiver accepts chunks only for that -// verified transfer ID and records its receipt on successful completion. -func (c *Client) SendGovernedFileStream(name string, r io.ReadSeeker, size int64, intent decision.Intent, result decision.Decision, stepTimeout time.Duration) (*StreamResult, error) { - return c.SendGovernedFileStreamWithAuthorizer(name, r, size, func(_ []byte) (decision.Intent, decision.Decision, error) { - return intent, result, nil - }, stepTimeout) -} - -// SendGovernedFileStreamWithDisclosure sends a governed resumable file whose -// Intent and authority Decision bind the supplied typed disclosure metadata. -// The metadata must match the final INIT (content hash, size, filename, and -// transfer ID) exactly; callers can derive these with BuildStreamInitPayload. -func (c *Client) SendGovernedFileStreamWithDisclosure(name string, r io.ReadSeeker, size int64, intent decision.Intent, result decision.Decision, disclosure decision.DisclosureBinding, stepTimeout time.Duration) (*StreamResult, error) { - return streamSendWithInit(c.conn, name, r, size, stepTimeout, func(id [transferIDLen]byte, declaredSize uint64, hash [32]byte, chunkSize uint32, filename string) (*Frame, error) { - init := encodeInit(id, declaredSize, hash, chunkSize, filename) - governed, err := NewGovernedStreamInitWithDisclosure(init, intent, result, disclosure) - if err != nil { - return nil, err - } - return EncodeGovernedStreamInit(governed) - }) -} - -// GovernedStreamAuthorizer receives the exact FileStream INIT payload after -// its stable transfer ID and content hash have been calculated. It must return -// a short-lived signed Intent and Decision bound to that exact payload. The -// callback runs before any stream frame is written, so a deny or unavailable -// authority cannot produce a partial transfer. -type GovernedStreamAuthorizer func(initPayload []byte) (decision.Intent, decision.Decision, error) - -// GovernedStreamDisclosureAuthorizer returns typed disclosure evidence for -// the exact INIT. Hosted federation clients use this after uploading the full -// file content and before the first stream byte is released to the peer. -type GovernedStreamDisclosureAuthorizer func(initPayload []byte) (decision.Intent, decision.Decision, decision.DisclosureBinding, error) - -// SendGovernedFileStreamWithAuthorizer obtains an exact signed decision only -// after the resumable stream's INIT bytes are known. This is the preferred -// sender API when the decision comes from an online authority. -func (c *Client) SendGovernedFileStreamWithAuthorizer(name string, r io.ReadSeeker, size int64, authorize GovernedStreamAuthorizer, stepTimeout time.Duration) (*StreamResult, error) { - if authorize == nil { - return nil, fmt.Errorf("dataexchange: governed stream authorizer is required") - } - return streamSendWithInit(c.conn, name, r, size, stepTimeout, func(id [transferIDLen]byte, declaredSize uint64, hash [32]byte, chunkSize uint32, filename string) (*Frame, error) { - init := encodeInit(id, declaredSize, hash, chunkSize, filename) - intent, result, err := authorize(append([]byte(nil), init.Payload...)) - if err != nil { - return nil, err - } - governed, err := NewGovernedStreamInit(init, intent, result) - if err != nil { - return nil, err - } - return EncodeGovernedStreamInit(governed) - }) -} - -func (c *Client) SendGovernedFileStreamWithDisclosureAuthorizer(name string, r io.ReadSeeker, size int64, authorize GovernedStreamDisclosureAuthorizer, stepTimeout time.Duration) (*StreamResult, error) { - if authorize == nil { - return nil, fmt.Errorf("dataexchange: governed stream disclosure authorizer is required") - } - return streamSendWithInit(c.conn, name, r, size, stepTimeout, func(id [transferIDLen]byte, declaredSize uint64, hash [32]byte, chunkSize uint32, filename string) (*Frame, error) { - init := encodeInit(id, declaredSize, hash, chunkSize, filename) - intent, result, disclosure, err := authorize(append([]byte(nil), init.Payload...)) - if err != nil { - return nil, err - } - governed, err := NewGovernedStreamInitWithDisclosure(init, intent, result, disclosure) - if err != nil { - return nil, err - } - return EncodeGovernedStreamInit(governed) - }) -} - // SendTrace wraps data in a TypeTrace frame with the current nanosecond clock. // Returns sentAtNs so the caller can correlate it against the timing ACK. func (c *Client) SendTrace(innerType uint32, data []byte) (sentAtNs int64, err error) { diff --git a/dataexchange.go b/dataexchange.go index d6e3fe0..8bffc8b 100644 --- a/dataexchange.go +++ b/dataexchange.go @@ -33,13 +33,14 @@ const ( // peer that does not understand TypeFileStream never sends INIT-ACK, so // the sender falls back to TypeFile. TypeFileStream uint32 = 7 - // TypeGoverned carries a regular data frame together with the sender's - // signed intent and authority decision. It is opt-in; enterprise receivers - // can require this envelope before persisting a message or file. - TypeGoverned uint32 = 8 - // TypeGovernedFileStream carries the signed authorization evidence for a - // TypeFileStream INIT. Subsequent chunks are accepted only while bound to - // that verified transfer ID on the same connection. + // TypeGoverned and TypeGovernedFileStream are reserved. They carried a + // data frame (8) or a TypeFileStream INIT (9) inside an envelope signed + // by the hosted control plane, which has been retired; the envelope + // codec and its receiver gates were removed with it. The numbers must + // not be reused: senders built before the removal can still emit them, + // and a receiver refuses both as unsupported frame types without + // storing anything. + TypeGoverned uint32 = 8 TypeGovernedFileStream uint32 = 9 // TypeTagged wraps another frame with optional correlation metadata // (Frame.MessageID / Frame.ReplyTo). Callers never set it directly: diff --git a/dedupe.go b/dedupe.go index c9d4e6f..7d59341 100644 --- a/dedupe.go +++ b/dedupe.go @@ -181,7 +181,7 @@ func (d *deliveryDedupe) len() int { // Streams have their own resume protocol and trace frames are diagnostics. func dedupeEligible(frameType uint32) bool { switch frameType { - case TypeText, TypeJSON, TypeBinary, TypeFile, TypeGoverned: + case TypeText, TypeJSON, TypeBinary, TypeFile: return true default: return false diff --git a/filestream.go b/filestream.go index 4876b52..e200d89 100644 --- a/filestream.go +++ b/filestream.go @@ -74,19 +74,6 @@ const ( // fresh connection. var ErrStreamUnsupported = errors.New("dataexchange: peer does not support TypeFileStream") -// BuildStreamInitPayload returns the deterministic INIT payload that a caller -// must bind in a signed file.share Intent before SendGovernedFileStream. The -// transfer ID is derived from the supplied full SHA-256, matching the sender's -// resumable-transfer state machine. -func BuildStreamInitPayload(name string, size int64, fullHash [32]byte) ([]byte, error) { - if size < 0 || !validGovernedFilename(name) { - return nil, fmt.Errorf("dataexchange: governed stream requires a non-negative size and safe filename") - } - var id [transferIDLen]byte - copy(id[:], fullHash[:transferIDLen]) - return append([]byte(nil), encodeInit(id, uint64(size), fullHash, uint32(StreamChunkSize), name).Payload...), nil -} - // --- control-frame codec --------------------------------------------------- func encodeStreamFrame(kind byte, id [transferIDLen]byte, body []byte) *Frame { @@ -218,8 +205,9 @@ func streamSend(conn frameRW, name string, r io.ReadSeeker, size int64, stepTime }) } -// streamInitBuilder allows the sender to substitute a governed INIT envelope -// while keeping the exact same chunk/ACK/resume state machine. +// streamInitBuilder builds the INIT frame for a transfer, so the frame that +// opens a stream can vary while the chunk/ACK/resume state machine stays +// the same. type streamInitBuilder func([transferIDLen]byte, uint64, [32]byte, uint32, string) (*Frame, error) func streamSendWithInit(conn frameRW, name string, r io.ReadSeeker, size int64, stepTimeout time.Duration, buildInit streamInitBuilder) (*StreamResult, error) { @@ -424,8 +412,8 @@ type StreamReceiver struct { receivedDir string onSaved func(name, path string, size int64) // onPrepare runs after full-content integrity verification but before the - // atomic rename. It lets a governed service durably record retention work - // for the final path before that path can become visible after a crash. + // atomic rename. It lets a caller durably record work for the final path + // before that path can become visible after a crash. onPrepare func([transferIDLen]byte, string, string, int64) error // onCommit runs after integrity verification and atomic rename, but before // onSaved. A required receipt recorder uses it to make the visible file @@ -477,15 +465,15 @@ func NewStreamReceiverWithQuota(receivedDir string, nameSuffix func(base string) } // NewStreamReceiverWithQuotaAndCommit extends the normal receiver with a -// transactional commit hook. It is intended for governed streams: the hook -// records evidence after the final file is durable but before consumers are -// notified. A hook failure removes the final file and reports COMPLETE failure. +// transactional commit hook: the hook runs after the final file is durable +// but before consumers are notified. A hook failure removes the final file +// and reports COMPLETE failure. func NewStreamReceiverWithQuotaAndCommit(receivedDir string, nameSuffix func(base string) string, onSaved func(name, path string, size int64), onCommit func([transferIDLen]byte, string, string, int64) error, quotaBytes int64) *StreamReceiver { return NewStreamReceiverWithQuotaAndPrepareAndCommit(receivedDir, nameSuffix, onSaved, nil, onCommit, quotaBytes) } // NewStreamReceiverWithQuotaAndPrepareAndCommit adds a pre-rename durable -// preparation hook to the governed commit path. A preparation error leaves no +// preparation hook to the commit path. A preparation error leaves no // final file visible; callers may keep the partial for retry/inspection. func NewStreamReceiverWithQuotaAndPrepareAndCommit(receivedDir string, nameSuffix func(base string) string, onSaved func(name, path string, size int64), onPrepare, onCommit func([transferIDLen]byte, string, string, int64) error, quotaBytes int64) *StreamReceiver { if nameSuffix == nil { diff --git a/governed.go b/governed.go deleted file mode 100644 index 4eadb6c..0000000 --- a/governed.go +++ /dev/null @@ -1,318 +0,0 @@ -// SPDX-License-Identifier: AGPL-3.0-or-later - -package dataexchange - -import ( - "bytes" - "context" - "crypto/sha256" - "encoding/binary" - "encoding/json" - "fmt" - "io" - "path/filepath" - "strconv" - "strings" - "unicode/utf8" - - "github.com/pilot-protocol/common/coreapi" - "github.com/pilot-protocol/common/decision" -) - -const governedFrameVersion uint16 = 1 - -// GovernedFrame transports an ordinary data frame with the exact signed -// authorization evidence for that frame. PayloadHash is computed over frame -// type, filename, and bytes, so a decision for text cannot be replayed as a -// file or against different contents. -type GovernedFrame struct { - Version uint16 `json:"version"` - Type uint32 `json:"type"` - Filename string `json:"filename,omitempty"` - Payload []byte `json:"payload"` - Disclosure *decision.DisclosureBinding `json:"disclosure,omitempty"` - Intent decision.Intent `json:"intent"` - Decision decision.Decision `json:"decision"` -} - -// GovernedFrameVerifier maps a trusted transport peer and governed frame into -// a local enforcement decision. The service invokes it before any disk write. -type GovernedFrameVerifier interface { - VerifyGovernedFrame(context.Context, coreapi.Addr, GovernedFrame) error -} - -// GovernedReceiptRecorder durably records a receiver-side enforcement receipt -// for a verified governed delivery. It receives the exact signed objects that -// authorized the data frame; implementations must fail closed on a recording -// error when their deployment requires evidence before acknowledgement. -type GovernedReceiptRecorder interface { - RecordGovernedReceipt(context.Context, decision.Intent, decision.Decision) error -} - -// GovernedDisclosureReceiptRecorder is the evidence extension for a -// disclosure-bound transport. A configured recorder must implement this when -// it receives typed metadata: silently emitting a V1 receipt would lose the -// evidence a required disclosure profile relies on. -type GovernedDisclosureReceiptRecorder interface { - RecordGovernedDisclosureReceipt(context.Context, decision.Intent, decision.Decision, decision.DisclosureBinding) error -} - -func recordGovernedReceipt(ctx context.Context, recorder GovernedReceiptRecorder, intent decision.Intent, result decision.Decision, disclosure *decision.DisclosureBinding) error { - if recorder == nil { - return fmt.Errorf("dataexchange: governed receipt recorder is not configured") - } - if disclosure == nil { - return recorder.RecordGovernedReceipt(ctx, intent, result) - } - typed, supported := recorder.(GovernedDisclosureReceiptRecorder) - if !supported { - return fmt.Errorf("dataexchange: governed receipt recorder does not support disclosure evidence") - } - return typed.RecordGovernedDisclosureReceipt(ctx, intent, result, *disclosure) -} - -// DecisionFrameVerifier is the reference receiver-side verifier. Resource -// must return the exact local resource identifier expected for this incoming -// frame (for example, "agent:finance/inbox"). This prevents a sender from -// reusing a valid decision for a different local destination. -type DecisionFrameVerifier struct { - Enforcer *decision.Enforcer - Resource func(coreapi.Addr, *Frame) string - RequireDisclosure bool -} - -func NewGovernedFrame(frame *Frame, intent decision.Intent, result decision.Decision) (GovernedFrame, error) { - if frame == nil { - return GovernedFrame{}, fmt.Errorf("dataexchange: governed frame is required") - } - governed := GovernedFrame{ - Version: governedFrameVersion, Type: frame.Type, Filename: frame.Filename, - Payload: append([]byte(nil), frame.Payload...), Intent: intent, Decision: result, - } - if err := governed.Validate(); err != nil { - return GovernedFrame{}, err - } - return governed, nil -} - -// NewGovernedFrameWithDisclosure creates a governed frame whose Intent binds -// a canonical disclosure metadata object. The caller must obtain the Decision -// for the disclosure-bound Intent; this constructor never changes authority. -func NewGovernedFrameWithDisclosure(frame *Frame, intent decision.Intent, result decision.Decision, disclosure decision.DisclosureBinding) (GovernedFrame, error) { - if frame == nil { - return GovernedFrame{}, fmt.Errorf("dataexchange: governed frame is required") - } - disclosure.Labels = append([]string(nil), disclosure.Labels...) - governed := GovernedFrame{ - Version: governedFrameVersion, Type: frame.Type, Filename: frame.Filename, - Payload: append([]byte(nil), frame.Payload...), Disclosure: &disclosure, Intent: intent, Decision: result, - } - if err := governed.Validate(); err != nil { - return GovernedFrame{}, err - } - return governed, nil -} - -func (frame GovernedFrame) Validate() error { - if frame.Version != governedFrameVersion || !governedDataType(frame.Type) { - return fmt.Errorf("dataexchange: invalid governed frame type") - } - if frame.Type == TypeFile { - if !validGovernedFilename(frame.Filename) { - return fmt.Errorf("dataexchange: invalid governed filename") - } - } else if frame.Filename != "" { - return fmt.Errorf("dataexchange: filename is only valid for file frames") - } - if uint64(len(frame.Payload)) > uint64(MaxFrameSize) { - return fmt.Errorf("dataexchange: governed payload exceeds maximum frame size") - } - if err := frame.Intent.Validate(); err != nil { - return fmt.Errorf("dataexchange: invalid governed intent: %w", err) - } - if frame.Intent.Signature == "" || frame.Decision.Signature == "" { - return fmt.Errorf("dataexchange: governed intent and decision must be signed") - } - if frame.Intent.Action != GovernedAction(frame.Type) { - return fmt.Errorf("dataexchange: governed intent action does not match frame type") - } - if err := frame.verifyPayloadBinding(); err != nil { - return err - } - if err := frame.Decision.Validate(); err != nil { - return fmt.Errorf("dataexchange: invalid governed decision: %w", err) - } - return nil -} - -func (frame GovernedFrame) verifyPayloadBinding() error { - if frame.Disclosure == nil { - if frame.Intent.PayloadHash != GovernedPayloadHash(frame.Type, frame.Filename, frame.Payload) { - return fmt.Errorf("dataexchange: governed intent payload binding mismatch") - } - return nil - } - if frame.Disclosure.ContentHash != decision.HashPayload(frame.Payload) || frame.Disclosure.DeclaredBytes != uint64(len(frame.Payload)) || frame.Disclosure.Filename != frame.Filename || frame.Disclosure.TransferID != "" { - return fmt.Errorf("dataexchange: governed disclosure does not match frame") - } - if err := frame.Disclosure.VerifyIntent(frame.Intent); err != nil { - return fmt.Errorf("dataexchange: governed disclosure intent binding: %w", err) - } - return nil -} - -func (frame GovernedFrame) DataFrame() *Frame { - return &Frame{Type: frame.Type, Filename: frame.Filename, Payload: append([]byte(nil), frame.Payload...)} -} - -func EncodeGovernedFrame(frame GovernedFrame) (*Frame, error) { - if err := frame.Validate(); err != nil { - return nil, err - } - body, err := json.Marshal(frame) - if err != nil { - return nil, fmt.Errorf("dataexchange: encode governed frame: %w", err) - } - if uint64(len(body)) > uint64(MaxFrameSize) { - return nil, fmt.Errorf("dataexchange: governed envelope exceeds maximum frame size") - } - return &Frame{Type: TypeGoverned, Payload: body}, nil -} - -func DecodeGovernedFrame(frame *Frame) (GovernedFrame, error) { - if frame == nil || frame.Type != TypeGoverned || uint64(len(frame.Payload)) > uint64(MaxFrameSize) { - return GovernedFrame{}, fmt.Errorf("dataexchange: invalid governed envelope") - } - decoder := json.NewDecoder(bytes.NewReader(frame.Payload)) - decoder.DisallowUnknownFields() - var governed GovernedFrame - if err := decoder.Decode(&governed); err != nil { - return GovernedFrame{}, fmt.Errorf("dataexchange: decode governed envelope: %w", err) - } - var trailing any - if err := decoder.Decode(&trailing); err != io.EOF { - return GovernedFrame{}, fmt.Errorf("dataexchange: trailing governed envelope data") - } - if err := governed.Validate(); err != nil { - return GovernedFrame{}, err - } - return governed, nil -} - -func (verifier DecisionFrameVerifier) VerifyGovernedFrame(ctx context.Context, remote coreapi.Addr, governed GovernedFrame) error { - if verifier.Enforcer == nil || verifier.Resource == nil { - return fmt.Errorf("dataexchange: decision frame verifier is not initialized") - } - if err := governed.Validate(); err != nil { - return err - } - if verifier.RequireDisclosure && governed.Disclosure == nil { - return fmt.Errorf("dataexchange: governed disclosure is required") - } - frame := governed.DataFrame() - resource := verifier.Resource(remote, frame) - if resource == "" || governed.Intent.Resource != resource { - return fmt.Errorf("dataexchange: governed intent resource binding mismatch") - } - var verifyErr error - if governed.Disclosure != nil { - verifyErr = verifier.Enforcer.VerifyDisclosure(ctx, governed.Intent, governed.Decision, *governed.Disclosure) - } else { - verifyErr = verifier.Enforcer.Verify(ctx, governed.Intent, governed.Decision) - } - if verifyErr != nil { - return fmt.Errorf("dataexchange: verify governed decision: %w", verifyErr) - } - switch governed.Decision.Outcome { - case decision.Allow: - return nil - case decision.Constrain: - return enforceFrameConstraints(governed.Decision.Constraints, remote, frame) - default: - return fmt.Errorf("dataexchange: governed decision outcome %q cannot permit delivery", governed.Decision.Outcome) - } -} - -// GovernedAction returns the only signed Intent action that can authorize a -// regular governed data frame of frameType. An empty result means the type is -// not eligible for a governed envelope. -func GovernedAction(frameType uint32) string { - switch frameType { - case TypeText: - return "data.send.text" - case TypeJSON: - return "data.send.json" - case TypeBinary: - return "data.send.binary" - case TypeFile: - return "file.share" - default: - return "" - } -} - -func governedAction(frameType uint32) string { return GovernedAction(frameType) } - -func governedDataType(frameType uint32) bool { return GovernedAction(frameType) != "" } - -// GovernedPayloadHash returns the exact payload binding required by a -// TypeGoverned Intent. It includes the frame kind and filename as well as the -// bytes, so callers must use it when creating the Intent before SendGoverned. -func GovernedPayloadHash(frameType uint32, filename string, payload []byte) string { - var header [12]byte - binary.BigEndian.PutUint32(header[:4], frameType) - binary.BigEndian.PutUint32(header[4:8], uint32(len(filename))) - binary.BigEndian.PutUint32(header[8:12], uint32(len(payload))) - hash := sha256.New() - _, _ = hash.Write([]byte("pilot-dataexchange-governed-frame-v1\x00")) - _, _ = hash.Write(header[:]) - _, _ = hash.Write([]byte(filename)) - _, _ = hash.Write(payload) - return decision.HashPayload(hash.Sum(nil)) -} - -func validGovernedFilename(name string) bool { - return name != "" && utf8.ValidString(name) && len(name) <= maxFilenameLen && filepath.Base(name) == name && !strings.ContainsAny(name, "/\\") -} - -func enforceFrameConstraints(constraints []decision.Constraint, remote coreapi.Addr, frame *Frame) error { - attributes := map[string]string{ - "sender": remote.String(), "frame_type": TypeName(frame.Type), "bytes": strconv.Itoa(len(frame.Payload)), "filename": frame.Filename, - } - for _, constraint := range constraints { - actual, found := attributes[constraint.Key] - if !found { - return fmt.Errorf("dataexchange: constraint %q has no enforceable frame attribute", constraint.Key) - } - switch constraint.Operator { - case "eq": - if actual != constraint.Value { - return fmt.Errorf("dataexchange: constraint %s rejected", constraint.Key) - } - case "one_of": - matched := false - for _, allowed := range strings.Split(constraint.Value, ",") { - if actual == strings.TrimSpace(allowed) { - matched = true - break - } - } - if !matched { - return fmt.Errorf("dataexchange: constraint %s rejected", constraint.Key) - } - case "max", "min": - value, valueErr := strconv.ParseUint(actual, 10, 64) - limit, limitErr := strconv.ParseUint(constraint.Value, 10, 64) - if valueErr != nil || limitErr != nil || (constraint.Operator == "max" && value > limit) || (constraint.Operator == "min" && value < limit) { - return fmt.Errorf("dataexchange: numeric constraint %s rejected", constraint.Key) - } - case "require": - if constraint.Value != "" && actual != constraint.Value { - return fmt.Errorf("dataexchange: required constraint %s rejected", constraint.Key) - } - default: - return fmt.Errorf("dataexchange: constraint operator %q is not enforceable", constraint.Operator) - } - } - return nil -} diff --git a/governed_replay.go b/governed_replay.go deleted file mode 100644 index 5a15ef9..0000000 --- a/governed_replay.go +++ /dev/null @@ -1,71 +0,0 @@ -// SPDX-License-Identifier: AGPL-3.0-or-later - -package dataexchange - -import ( - "fmt" - "sync" - "time" - - "github.com/pilot-protocol/common/decision" -) - -// maxGovernedReplayEntries bounds the receiver-side replay cache. Entries are -// keyed on the signed intent and expire at the intent's own ExpiresAt (<= the -// 5-minute MaxIntentTTL), so under legitimate signed traffic the cache drains -// continuously. The cap is a backstop against pathological retention; on -// overflow a new transfer is refused (fail closed) rather than admitted -// un-deduplicated. -const maxGovernedReplayEntries = 1 << 20 - -// governedReplayGuard rejects a second delivery of the same signed governed -// intent. A verified governed envelope is otherwise a bearer capability: any -// peer that observed one could re-send the exact bytes within the intent TTL, -// producing duplicate authorized deliveries and re-charging the signing -// agent's quota. Dedup is keyed on (tenant, agent, intent id) — all -// signature-authenticated fields — so a replay cannot dodge it by presenting a -// different transport peer. A legitimate retry must carry a fresh intent -// (fresh nonce/id); reusing a signed intent IS the replay pattern. -type governedReplayGuard struct { - mu sync.Mutex - seen map[string]int64 // key -> intent ExpiresAt (unix seconds) - now func() time.Time -} - -func newGovernedReplayGuard() *governedReplayGuard { - return &governedReplayGuard{seen: make(map[string]int64), now: time.Now} -} - -func governedReplayKey(intent decision.Intent) string { - return intent.TenantID + "\x1f" + intent.AgentID + "\x1f" + intent.ID -} - -// admit records the intent as delivered and returns an error if it was already -// delivered (replay) or if the cache is saturated. A blank intent ID is a -// no-op (upstream freshness/nonce checks still apply). -func (g *governedReplayGuard) admit(intent decision.Intent) error { - if intent.ID == "" { - return nil - } - now := g.now().Unix() - key := governedReplayKey(intent) - g.mu.Lock() - defer g.mu.Unlock() - for k, exp := range g.seen { - if exp <= now { - delete(g.seen, k) - } - } - if exp, ok := g.seen[key]; ok && exp > now { - return fmt.Errorf("governed intent already delivered (replay rejected)") - } - if len(g.seen) >= maxGovernedReplayEntries { - return fmt.Errorf("governed replay cache saturated") - } - expiresAt := intent.ExpiresAt - if expiresAt <= now { - expiresAt = now + int64(decision.MaxIntentTTL/time.Second) - } - g.seen[key] = expiresAt - return nil -} diff --git a/governed_stream.go b/governed_stream.go deleted file mode 100644 index 67437ba..0000000 --- a/governed_stream.go +++ /dev/null @@ -1,285 +0,0 @@ -// SPDX-License-Identifier: AGPL-3.0-or-later - -package dataexchange - -import ( - "bytes" - "context" - "encoding/binary" - "encoding/hex" - "encoding/json" - "fmt" - "io" - "strconv" - "strings" - - "github.com/pilot-protocol/common/coreapi" - "github.com/pilot-protocol/common/decision" -) - -const governedStreamVersion uint16 = 1 - -// GovernedStreamInit binds a signed authority decision to the exact INIT -// frame for a resumable file stream. The INIT includes the full file hash, -// filename, declared length, and transfer ID; individual chunks are accepted -// only after this envelope has been verified locally. -type GovernedStreamInit struct { - Version uint16 `json:"version"` - InitPayload []byte `json:"init_payload"` - Disclosure *decision.DisclosureBinding `json:"disclosure,omitempty"` - Intent decision.Intent `json:"intent"` - Decision decision.Decision `json:"decision"` -} - -// GovernedStreamVerifier verifies a stream INIT before the receiver creates a -// partial file. It is intentionally distinct from GovernedFrameVerifier so -// embedders that do not support large files need not accidentally authorize -// them. -type GovernedStreamVerifier interface { - VerifyGovernedStreamInit(context.Context, coreapi.Addr, GovernedStreamInit) error -} - -func NewGovernedStreamInit(init *Frame, intent decision.Intent, result decision.Decision) (GovernedStreamInit, error) { - if init == nil || init.Type != TypeFileStream { - return GovernedStreamInit{}, fmt.Errorf("dataexchange: stream INIT frame is required") - } - governed := GovernedStreamInit{ - Version: governedStreamVersion, InitPayload: append([]byte(nil), init.Payload...), Intent: intent, Decision: result, - } - if err := governed.Validate(); err != nil { - return GovernedStreamInit{}, err - } - return governed, nil -} - -// NewGovernedStreamInitWithDisclosure creates a governed resumable transfer -// whose signed Intent binds typed metadata as well as the content hash, -// filename, declared bytes, and stable transfer ID in its INIT frame. -func NewGovernedStreamInitWithDisclosure(init *Frame, intent decision.Intent, result decision.Decision, disclosure decision.DisclosureBinding) (GovernedStreamInit, error) { - if init == nil || init.Type != TypeFileStream { - return GovernedStreamInit{}, fmt.Errorf("dataexchange: stream INIT frame is required") - } - disclosure.Labels = append([]string(nil), disclosure.Labels...) - governed := GovernedStreamInit{ - Version: governedStreamVersion, InitPayload: append([]byte(nil), init.Payload...), Disclosure: &disclosure, Intent: intent, Decision: result, - } - if err := governed.Validate(); err != nil { - return GovernedStreamInit{}, err - } - return governed, nil -} - -func (stream GovernedStreamInit) Validate() error { - if stream.Version != governedStreamVersion || len(stream.InitPayload) == 0 || uint64(len(stream.InitPayload)) > uint64(MaxFrameSize) { - return fmt.Errorf("dataexchange: invalid governed stream INIT") - } - _, id, body, ok := decodeStreamFrame(&Frame{Type: TypeFileStream, Payload: stream.InitPayload}) - if !ok || stream.InitPayload[0] != streamKindInit { - return fmt.Errorf("dataexchange: governed stream must contain a file-stream INIT") - } - size, hash, chunkSize, name, valid := decodeInit(body) - if !valid || !validGovernedFilename(name) || chunkSize == 0 || size > uint64(^uint64(0)>>1) { - return fmt.Errorf("dataexchange: invalid governed stream metadata") - } - if !bytes.Equal(id[:], hash[:transferIDLen]) { - return fmt.Errorf("dataexchange: governed stream transfer ID does not match content hash") - } - if err := stream.Intent.Validate(); err != nil { - return fmt.Errorf("dataexchange: invalid governed stream intent: %w", err) - } - if stream.Intent.Signature == "" || stream.Decision.Signature == "" { - return fmt.Errorf("dataexchange: governed stream intent and decision must be signed") - } - if stream.Intent.Action != "file.share" { - return fmt.Errorf("dataexchange: governed stream intent action mismatch") - } - if stream.Disclosure == nil && stream.Intent.PayloadHash != GovernedStreamPayloadHash(stream.InitPayload) { - return fmt.Errorf("dataexchange: governed stream intent binding mismatch") - } - if stream.Disclosure != nil { - if stream.Disclosure.ContentHash != hex.EncodeToString(hash[:]) || stream.Disclosure.DeclaredBytes != size || stream.Disclosure.Filename != name || stream.Disclosure.TransferID != hex.EncodeToString(id[:]) { - return fmt.Errorf("dataexchange: governed stream disclosure does not match INIT") - } - if err := stream.Disclosure.VerifyIntent(stream.Intent); err != nil { - return fmt.Errorf("dataexchange: governed stream disclosure intent binding: %w", err) - } - } - if err := stream.Decision.Validate(); err != nil { - return fmt.Errorf("dataexchange: invalid governed stream decision: %w", err) - } - return nil -} - -func (stream GovernedStreamInit) InitFrame() *Frame { - return &Frame{Type: TypeFileStream, Payload: append([]byte(nil), stream.InitPayload...)} -} - -func (stream GovernedStreamInit) TransferID() ([transferIDLen]byte, error) { - _, id, _, ok := decodeStreamFrame(stream.InitFrame()) - if !ok { - return [transferIDLen]byte{}, fmt.Errorf("dataexchange: invalid governed stream transfer ID") - } - return id, nil -} - -// DeclaredBytes returns the signed stream length from its verified INIT. It is -// used by receiver-local admission controls before any stream chunks are -// accepted. -func (stream GovernedStreamInit) DeclaredBytes() (uint64, error) { - if err := stream.Validate(); err != nil { - return 0, err - } - _, _, body, ok := decodeStreamFrame(stream.InitFrame()) - if !ok { - return 0, fmt.Errorf("dataexchange: invalid governed stream INIT") - } - size, _, _, _, valid := decodeInit(body) - if !valid { - return 0, fmt.Errorf("dataexchange: invalid governed stream metadata") - } - return size, nil -} - -func EncodeGovernedStreamInit(stream GovernedStreamInit) (*Frame, error) { - if err := stream.Validate(); err != nil { - return nil, err - } - body, err := json.Marshal(stream) - if err != nil { - return nil, fmt.Errorf("dataexchange: encode governed stream: %w", err) - } - if uint64(len(body)) > uint64(MaxFrameSize) { - return nil, fmt.Errorf("dataexchange: governed stream envelope exceeds maximum frame size") - } - return &Frame{Type: TypeGovernedFileStream, Payload: body}, nil -} - -func DecodeGovernedStreamInit(frame *Frame) (GovernedStreamInit, error) { - if frame == nil || frame.Type != TypeGovernedFileStream || uint64(len(frame.Payload)) > uint64(MaxFrameSize) { - return GovernedStreamInit{}, fmt.Errorf("dataexchange: invalid governed stream envelope") - } - decoder := json.NewDecoder(bytes.NewReader(frame.Payload)) - decoder.DisallowUnknownFields() - var stream GovernedStreamInit - if err := decoder.Decode(&stream); err != nil { - return GovernedStreamInit{}, fmt.Errorf("dataexchange: decode governed stream envelope: %w", err) - } - var trailing any - if err := decoder.Decode(&trailing); err != io.EOF { - return GovernedStreamInit{}, fmt.Errorf("dataexchange: trailing governed stream envelope data") - } - if err := stream.Validate(); err != nil { - return GovernedStreamInit{}, err - } - return stream, nil -} - -// GovernedStreamPayloadHash binds the exact bytes of the file-stream INIT -// frame. It therefore covers the filename, size, full content hash, chunk -// size, and transfer ID without having to duplicate the stream wire grammar. -func GovernedStreamPayloadHash(initPayload []byte) string { - var length [4]byte - binary.BigEndian.PutUint32(length[:], uint32(len(initPayload))) - buffer := make([]byte, 0, len("pilot-dataexchange-governed-stream-v1\x00")+len(length)+len(initPayload)) - buffer = append(buffer, []byte("pilot-dataexchange-governed-stream-v1\x00")...) - buffer = append(buffer, length[:]...) - buffer = append(buffer, initPayload...) - return decision.HashPayload(buffer) -} - -// GovernedStreamDisclosureMetadata returns the canonical values a hosted -// federation disclosure must bind for an already-built stream INIT. -func GovernedStreamDisclosureMetadata(initPayload []byte) (transferID string, declaredBytes uint64, contentHash string, filename string, err error) { - _, id, body, ok := decodeStreamFrame(&Frame{Type: TypeFileStream, Payload: initPayload}) - if !ok || len(initPayload) == 0 || initPayload[0] != streamKindInit { - err = fmt.Errorf("dataexchange: invalid file-stream INIT") - return - } - size, hash, _, name, valid := decodeInit(body) - if !valid || !validGovernedFilename(name) || !bytes.Equal(id[:], hash[:transferIDLen]) { - err = fmt.Errorf("dataexchange: invalid file-stream disclosure metadata") - return - } - return hex.EncodeToString(id[:]), size, hex.EncodeToString(hash[:]), name, nil -} - -func (verifier DecisionFrameVerifier) VerifyGovernedStreamInit(ctx context.Context, remote coreapi.Addr, stream GovernedStreamInit) error { - if verifier.Enforcer == nil || verifier.Resource == nil { - return fmt.Errorf("dataexchange: decision stream verifier is not initialized") - } - if err := stream.Validate(); err != nil { - return err - } - if verifier.RequireDisclosure && stream.Disclosure == nil { - return fmt.Errorf("dataexchange: governed stream disclosure is required") - } - init := stream.InitFrame() - _, _, body, _ := decodeStreamFrame(init) - size, _, _, name, _ := decodeInit(body) - resourceFrame := &Frame{Type: TypeFileStream, Filename: name} - resource := verifier.Resource(remote, resourceFrame) - if resource == "" || stream.Intent.Resource != resource { - return fmt.Errorf("dataexchange: governed stream intent resource binding mismatch") - } - var verifyErr error - if stream.Disclosure != nil { - verifyErr = verifier.Enforcer.VerifyDisclosure(ctx, stream.Intent, stream.Decision, *stream.Disclosure) - } else { - verifyErr = verifier.Enforcer.Verify(ctx, stream.Intent, stream.Decision) - } - if verifyErr != nil { - return fmt.Errorf("dataexchange: verify governed stream decision: %w", verifyErr) - } - switch stream.Decision.Outcome { - case decision.Allow: - return nil - case decision.Constrain: - return enforceStreamConstraints(stream.Decision.Constraints, remote, name, size) - default: - return fmt.Errorf("dataexchange: governed stream decision outcome %q cannot permit delivery", stream.Decision.Outcome) - } -} - -func enforceStreamConstraints(constraints []decision.Constraint, remote coreapi.Addr, filename string, size uint64) error { - attributes := map[string]string{ - "sender": remote.String(), "frame_type": TypeName(TypeFileStream), "bytes": strconv.FormatUint(size, 10), "filename": filename, - } - for _, constraint := range constraints { - actual, found := attributes[constraint.Key] - if !found { - return fmt.Errorf("dataexchange: constraint %q has no enforceable stream attribute", constraint.Key) - } - switch constraint.Operator { - case "eq": - if actual != constraint.Value { - return fmt.Errorf("dataexchange: stream constraint %s rejected", constraint.Key) - } - case "one_of": - matched := false - for _, allowed := range strings.Split(constraint.Value, ",") { - if actual == strings.TrimSpace(allowed) { - matched = true - break - } - } - if !matched { - return fmt.Errorf("dataexchange: stream constraint %s rejected", constraint.Key) - } - case "max", "min": - value, valueErr := strconv.ParseUint(actual, 10, 64) - limit, limitErr := strconv.ParseUint(constraint.Value, 10, 64) - if valueErr != nil || limitErr != nil || (constraint.Operator == "max" && value > limit) || (constraint.Operator == "min" && value < limit) { - return fmt.Errorf("dataexchange: numeric stream constraint %s rejected", constraint.Key) - } - case "require": - if constraint.Value != "" && actual != constraint.Value { - return fmt.Errorf("dataexchange: required stream constraint %s rejected", constraint.Key) - } - default: - return fmt.Errorf("dataexchange: constraint operator %q is not enforceable for a stream", constraint.Operator) - } - } - return nil -} - -var _ GovernedStreamVerifier = DecisionFrameVerifier{} diff --git a/governed_test.go b/governed_test.go deleted file mode 100644 index a2cef03..0000000 --- a/governed_test.go +++ /dev/null @@ -1,686 +0,0 @@ -// SPDX-License-Identifier: AGPL-3.0-or-later - -//go:build !no_dataexchange -// +build !no_dataexchange - -package dataexchange - -import ( - "bytes" - "context" - "crypto/ed25519" - "crypto/rand" - "crypto/sha256" - "encoding/json" - "errors" - "fmt" - "io" - "os" - "path/filepath" - "strings" - "testing" - "time" - - "github.com/pilot-protocol/common/coreapi" - "github.com/pilot-protocol/common/decision" -) - -type governedTestTrust struct { - intentKey ed25519.PublicKey - decisionKey ed25519.PublicKey -} - -type noWriteStream struct{ writes bytes.Buffer } - -func (stream *noWriteStream) Read([]byte) (int, error) { - return 0, errors.New("read should not occur before authorization") -} -func (stream *noWriteStream) Write(value []byte) (int, error) { - return stream.writes.Write(value) -} -func (stream *noWriteStream) Close() error { return nil } - -func (trust governedTestTrust) IntentKey(context.Context, string, string, string) (ed25519.PublicKey, error) { - return trust.intentKey, nil -} - -func (trust governedTestTrust) DecisionKey(context.Context, string, string) (ed25519.PublicKey, error) { - return trust.decisionKey, nil -} - -func (governedTestTrust) MinimumState(context.Context, string) (uint64, uint64, error) { - return 7, 3, nil -} - -type governedTestCeiling struct{} - -func (governedTestCeiling) Check(context.Context, decision.Intent, decision.Decision) error { - return nil -} - -func (governedTestCeiling) CheckDisclosure(context.Context, decision.Intent, decision.Decision, decision.DisclosureBinding) error { - return nil -} - -type governedReceiptRecorder struct { - calls int - intent decision.Intent - decision decision.Decision - err error -} - -type contentInspectorFunc func(context.Context, decision.Intent, *decision.DisclosureBinding, string, string, io.Reader) error - -func (inspect contentInspectorFunc) InspectDisclosureContent(ctx context.Context, intent decision.Intent, disclosure *decision.DisclosureBinding, contentType, filename string, content io.Reader) error { - return inspect(ctx, intent, disclosure, contentType, filename, content) -} - -func TestGovernedContentInspectionIsLocalAndFailClosed(t *testing.T) { - frame := &Frame{Type: TypeFile, Filename: "invoice.pdf", Payload: []byte("classified invoice")} - governed, _ := newGovernedTestFrame(t, frame, decision.Allow, nil) - disclosure := decision.DisclosureBinding{Version: decision.DisclosureBindingVersion, ContentType: "application/pdf", Filename: frame.Filename} - governed.Disclosure = &disclosure - var observed []byte - service := NewService(ServiceConfig{GovernedContentInspector: contentInspectorFunc(func(_ context.Context, intent decision.Intent, got *decision.DisclosureBinding, contentType, filename string, content io.Reader) error { - if intent.ID != governed.Intent.ID || got == nil || contentType != "application/pdf" || filename != frame.Filename { - t.Fatalf("inspection metadata intent=%+v disclosure=%+v content_type=%q filename=%q", intent, got, contentType, filename) - } - var err error - observed, err = io.ReadAll(content) - return err - })}) - if err := service.inspectGovernedFrame(context.Background(), governed); err != nil || string(observed) != string(frame.Payload) { - t.Fatalf("inspection err=%v payload=%q", err, observed) - } - service.cfg.GovernedContentInspector = contentInspectorFunc(func(context.Context, decision.Intent, *decision.DisclosureBinding, string, string, io.Reader) error { - return errors.New("detector unavailable") - }) - if err := service.inspectGovernedFrame(context.Background(), governed); err == nil || err.Error() != "governed content inspection rejected" { - t.Fatalf("inspection failure leaked or was accepted: %v", err) - } - if err := (&Service{cfg: ServiceConfig{RequireGoverned: true, RequireGovernedContentInspection: true}}).validateGovernedConfig(); err == nil || !strings.Contains(err.Error(), "local inspector") { - t.Fatalf("required inspection without local hook err=%v", err) - } -} - -func TestGovernedTransferQuotaChargesOnlyVerifiedAgentIdentity(t *testing.T) { - limiter, err := decision.NewTransferQuotaLimiter(decision.TransferQuotaConfig{Window: time.Minute, MaxBytes: 5, MaxSenders: 2}) - if err != nil { - t.Fatal(err) - } - service := NewService(ServiceConfig{RequireGoverned: true, GovernedTransferQuota: limiter}) - if err := service.validateGovernedConfig(); err != nil { - t.Fatal(err) - } - if err := service.admitGovernedTransfer(decision.Intent{AgentID: "sender-a"}, 5); err != nil { - t.Fatal(err) - } - if err := service.admitGovernedTransfer(decision.Intent{AgentID: "sender-a"}, 1); err == nil || err.Error() != "governed transfer quota rejected" { - t.Fatalf("quota error=%v", err) - } - if err := (&Service{cfg: ServiceConfig{GovernedTransferQuota: limiter}}).validateGovernedConfig(); err == nil || !strings.Contains(err.Error(), "requires a governed receiver") { - t.Fatalf("legacy quota configuration error=%v", err) - } -} - -func (recorder *governedReceiptRecorder) RecordGovernedReceipt(_ context.Context, intent decision.Intent, result decision.Decision) error { - recorder.calls++ - recorder.intent, recorder.decision = intent, result - return recorder.err -} - -type legacyGovernedReceiptRecorder struct{} - -func (legacyGovernedReceiptRecorder) RecordGovernedReceipt(context.Context, decision.Intent, decision.Decision) error { - return nil -} - -type disclosureGovernedReceiptRecorder struct { - governedReceiptRecorder - disclosure decision.DisclosureBinding -} - -func (recorder *disclosureGovernedReceiptRecorder) RecordGovernedDisclosureReceipt(_ context.Context, intent decision.Intent, result decision.Decision, disclosure decision.DisclosureBinding) error { - recorder.calls++ - recorder.intent, recorder.decision, recorder.disclosure = intent, result, disclosure - return recorder.err -} - -func TestDisclosureReceiptRecorderRequiresV2Evidence(t *testing.T) { - disclosure := decision.DisclosureBinding{Version: decision.DisclosureBindingVersion} - if err := recordGovernedReceipt(context.Background(), legacyGovernedReceiptRecorder{}, decision.Intent{ID: "intent"}, decision.Decision{ID: "decision"}, &disclosure); err == nil || !strings.Contains(err.Error(), "does not support disclosure") { - t.Fatalf("legacy disclosure recorder err=%v", err) - } - recorder := &disclosureGovernedReceiptRecorder{} - if err := recordGovernedReceipt(context.Background(), recorder, decision.Intent{ID: "intent"}, decision.Decision{ID: "decision"}, &disclosure); err != nil { - t.Fatal(err) - } - if recorder.calls != 1 || recorder.disclosure.Version != decision.DisclosureBindingVersion { - t.Fatalf("disclosure recorder=%+v", recorder) - } -} - -func newGovernedTestFrame(t *testing.T, frame *Frame, outcome decision.Outcome, constraints []decision.Constraint) (GovernedFrame, DecisionFrameVerifier) { - t.Helper() - intentPublic, intentPrivate, err := ed25519.GenerateKey(rand.Reader) - if err != nil { - t.Fatalf("generate intent key: %v", err) - } - decisionPublic, decisionPrivate, err := ed25519.GenerateKey(rand.Reader) - if err != nil { - t.Fatalf("generate decision key: %v", err) - } - now := time.Now().UTC().Truncate(time.Second) - nonce, err := decision.NewNonce() - if err != nil { - t.Fatalf("nonce: %v", err) - } - intent := decision.Intent{ - Version: decision.SchemaVersion, ID: "governed-intent", TenantID: "tenant-a", AgentID: "sender-a", - Action: governedAction(frame.Type), Resource: "agent:receiver/inbox", PayloadHash: GovernedPayloadHash(frame.Type, frame.Filename, frame.Payload), - Risk: decision.RiskMedium, IssuedAt: now.Unix(), ExpiresAt: now.Add(2 * time.Minute).Unix(), Nonce: nonce, KeyID: "sender-key", - } - if err := intent.Sign(intentPrivate); err != nil { - t.Fatalf("sign intent: %v", err) - } - intentHash, err := intent.Hash() - if err != nil { - t.Fatalf("hash intent: %v", err) - } - result := decision.Decision{ - Version: decision.SchemaVersion, ID: "governed-decision", IntentHash: intentHash, TenantID: intent.TenantID, AgentID: intent.AgentID, - Outcome: outcome, Constraints: constraints, PolicyRevision: 7, RevocationEpoch: 3, ProviderID: "authority-a", - IssuedAt: now.Unix(), ExpiresAt: now.Add(90 * time.Second).Unix(), KeyID: "authority-key", - } - if err := result.Sign(decisionPrivate); err != nil { - t.Fatalf("sign decision: %v", err) - } - governed, err := NewGovernedFrame(frame, intent, result) - if err != nil { - t.Fatalf("new governed frame: %v", err) - } - verifier := DecisionFrameVerifier{ - Enforcer: &decision.Enforcer{ - Trust: governedTestTrust{intentKey: intentPublic, decisionKey: decisionPublic}, Ceiling: governedTestCeiling{}, Now: func() time.Time { return now }, - }, - Resource: func(_ coreapi.Addr, _ *Frame) string { return "agent:receiver/inbox" }, - } - return governed, verifier -} - -func newGovernedStreamForTest(t *testing.T, name string, payload []byte, outcome decision.Outcome, constraints []decision.Constraint) (GovernedStreamInit, DecisionFrameVerifier) { - t.Helper() - intentPublic, intentPrivate, err := ed25519.GenerateKey(rand.Reader) - if err != nil { - t.Fatal(err) - } - decisionPublic, decisionPrivate, err := ed25519.GenerateKey(rand.Reader) - if err != nil { - t.Fatal(err) - } - now := time.Now().UTC().Truncate(time.Second) - fullHash := sha256.Sum256(payload) - initPayload, err := BuildStreamInitPayload(name, int64(len(payload)), fullHash) - if err != nil { - t.Fatal(err) - } - nonce, err := decision.NewNonce() - if err != nil { - t.Fatal(err) - } - intent := decision.Intent{ - Version: decision.SchemaVersion, ID: "governed-stream-intent", TenantID: "tenant-a", AgentID: "sender-a", - Action: "file.share", Resource: "agent:receiver/inbox", PayloadHash: GovernedStreamPayloadHash(initPayload), - Risk: decision.RiskMedium, IssuedAt: now.Unix(), ExpiresAt: now.Add(2 * time.Minute).Unix(), Nonce: nonce, KeyID: "sender-key", - } - if err := intent.Sign(intentPrivate); err != nil { - t.Fatal(err) - } - intentHash, err := intent.Hash() - if err != nil { - t.Fatal(err) - } - result := decision.Decision{ - Version: decision.SchemaVersion, ID: "governed-stream-decision", IntentHash: intentHash, TenantID: intent.TenantID, AgentID: intent.AgentID, - Outcome: outcome, Constraints: constraints, PolicyRevision: 7, RevocationEpoch: 3, ProviderID: "authority-a", - IssuedAt: now.Unix(), ExpiresAt: now.Add(90 * time.Second).Unix(), KeyID: "authority-key", - } - if err := result.Sign(decisionPrivate); err != nil { - t.Fatal(err) - } - stream, err := NewGovernedStreamInit(&Frame{Type: TypeFileStream, Payload: initPayload}, intent, result) - if err != nil { - t.Fatal(err) - } - verifier := DecisionFrameVerifier{ - Enforcer: &decision.Enforcer{ - Trust: governedTestTrust{intentKey: intentPublic, decisionKey: decisionPublic}, Ceiling: governedTestCeiling{}, Now: func() time.Time { return now }, - }, - Resource: func(_ coreapi.Addr, _ *Frame) string { return "agent:receiver/inbox" }, - } - return stream, verifier -} - -func TestGovernedStreamAuthorizerReceivesExactInitBeforeAnyWrite(t *testing.T) { - payload := []byte("sensitive export") - stream := &noWriteStream{} - var initPayload []byte - denied := errors.New("policy denied") - _, err := streamSendWithInit(stream, "export.txt", bytes.NewReader(payload), int64(len(payload)), time.Second, func(id [transferIDLen]byte, size uint64, hash [32]byte, chunkSize uint32, name string) (*Frame, error) { - init := encodeInit(id, size, hash, chunkSize, name) - initPayload = append([]byte(nil), init.Payload...) - return nil, denied - }) - if !errors.Is(err, denied) { - t.Fatalf("err=%v, want denied authorization", err) - } - if len(stream.writes.Bytes()) != 0 { - t.Fatalf("authorization rejection wrote %d transport bytes", stream.writes.Len()) - } - fullHash := sha256.Sum256(payload) - want, err := BuildStreamInitPayload("export.txt", int64(len(payload)), fullHash) - if err != nil { - t.Fatal(err) - } - if !bytes.Equal(initPayload, want) { - t.Fatalf("authorizer init does not match governed payload binding") - } -} - -func TestGovernedFrameRoundTripBindsPayloadAndDecision(t *testing.T) { - frame := &Frame{Type: TypeFile, Filename: "report.pdf", Payload: []byte("approved report")} - governed, _ := newGovernedTestFrame(t, frame, decision.Allow, nil) - envelope, err := EncodeGovernedFrame(governed) - if err != nil { - t.Fatalf("encode governed frame: %v", err) - } - decoded, err := DecodeGovernedFrame(envelope) - if err != nil { - t.Fatalf("decode governed frame: %v", err) - } - if got := decoded.DataFrame(); got.Type != frame.Type || got.Filename != frame.Filename || string(got.Payload) != string(frame.Payload) { - t.Fatalf("decoded data frame = %#v, want %#v", got, frame) - } - - decoded.Payload = []byte("substituted report") - tamperedBody, err := json.Marshal(decoded) - if err != nil { - t.Fatalf("marshal tampered frame: %v", err) - } - if _, err := DecodeGovernedFrame(&Frame{Type: TypeGoverned, Payload: tamperedBody}); err == nil || !strings.Contains(err.Error(), "payload binding") { - t.Fatalf("tampered payload error = %v, want payload-binding failure", err) - } - - invalidName := governed - invalidName.Filename = string([]byte{0xff}) - if err := invalidName.Validate(); err == nil || !strings.Contains(err.Error(), "invalid governed filename") { - t.Fatalf("invalid filename error = %v, want UTF-8 validation failure", err) - } -} - -func TestGovernedFrameDisclosureBindingAndRequiredProfile(t *testing.T) { - frame := &Frame{Type: TypeFile, Filename: "report.json", Payload: []byte(`{"amount":42}`)} - governed, verifier := newGovernedTestFrame(t, frame, decision.Allow, nil) - strict := verifier - strict.RequireDisclosure = true - if err := strict.VerifyGovernedFrame(context.Background(), coreapi.Addr{}, governed); err == nil || !strings.Contains(err.Error(), "disclosure is required") { - t.Fatalf("missing disclosure error=%v", err) - } - - binding := decision.DisclosureBinding{ - Version: decision.DisclosureBindingVersion, ContentHash: decision.HashPayload(frame.Payload), DeclaredBytes: uint64(len(frame.Payload)), - ContentType: "application/json", Labels: []string{"finance", "pii"}, Recipient: "agent:finance", Purpose: "invoice-payment", Residency: "eu-west-1", Filename: frame.Filename, - } - hash, err := binding.Hash() - if err != nil { - t.Fatal(err) - } - // Constructing a fresh signed Intent/Decision is covered by the common - // disclosure tests; this transport test verifies the frame-level mismatch - // and required-profile checks before receiver-side signature resolution. - governed.Intent.PayloadHash = hash - governed.Intent.Audience = binding.Recipient - governed.Intent.Purpose = binding.Purpose - governed.Intent.Signature = "transport-test-signature" - governed.Decision.Signature = "transport-test-signature" - governed, err = NewGovernedFrameWithDisclosure(frame, governed.Intent, governed.Decision, binding) - if err != nil { - t.Fatalf("valid disclosure envelope: %v", err) - } - tampered := governed - tampered.Disclosure = &binding - tampered.Disclosure.Residency = "us-east-1" - if err := tampered.Validate(); err == nil || !strings.Contains(err.Error(), "disclosure intent binding") { - t.Fatalf("disclosure mutation error=%v", err) - } -} - -func TestDecisionFrameVerifierBindsReceiverAndEnforcesConstraints(t *testing.T) { - frame := &Frame{Type: TypeText, Payload: []byte("hello")} - governed, verifier := newGovernedTestFrame(t, frame, decision.Constrain, []decision.Constraint{ - {Key: "frame_type", Operator: "eq", Value: "TEXT"}, - {Key: "bytes", Operator: "max", Value: "5"}, - }) - if err := verifier.VerifyGovernedFrame(context.Background(), coreapi.Addr{}, governed); err != nil { - t.Fatalf("verify governed frame: %v", err) - } - - wrongDestination := verifier - wrongDestination.Resource = func(_ coreapi.Addr, _ *Frame) string { return "agent:other/inbox" } - if err := wrongDestination.VerifyGovernedFrame(context.Background(), coreapi.Addr{}, governed); err == nil || !strings.Contains(err.Error(), "resource binding") { - t.Fatalf("wrong destination error = %v, want resource-binding failure", err) - } - - tooLarge, constrainedVerifier := newGovernedTestFrame(t, &Frame{Type: TypeText, Payload: []byte("too large")}, decision.Constrain, []decision.Constraint{{Key: "bytes", Operator: "max", Value: "3"}}) - if err := constrainedVerifier.VerifyGovernedFrame(context.Background(), coreapi.Addr{}, tooLarge); err == nil || !strings.Contains(err.Error(), "numeric constraint") { - t.Fatalf("oversize payload error = %v, want constraint failure", err) - } -} - -func TestGovernedStreamInitBindsExactMetadataAndConstraints(t *testing.T) { - stream, verifier := newGovernedStreamForTest(t, "large.bin", []byte("governed streaming payload"), decision.Constrain, []decision.Constraint{{Key: "bytes", Operator: "max", Value: "32"}}) - envelope, err := EncodeGovernedStreamInit(stream) - if err != nil { - t.Fatal(err) - } - decoded, err := DecodeGovernedStreamInit(envelope) - if err != nil { - t.Fatal(err) - } - if err := verifier.VerifyGovernedStreamInit(context.Background(), coreapi.Addr{}, decoded); err != nil { - t.Fatalf("verify governed stream: %v", err) - } - tampered := decoded - tampered.InitPayload[len(tampered.InitPayload)-1] ^= 0x01 - body, err := json.Marshal(tampered) - if err != nil { - t.Fatal(err) - } - if _, err := DecodeGovernedStreamInit(&Frame{Type: TypeGovernedFileStream, Payload: body}); err == nil || !strings.Contains(err.Error(), "binding") { - t.Fatalf("tampered stream error=%v", err) - } - tooLarge, constrainedVerifier := newGovernedStreamForTest(t, "large.bin", bytes.Repeat([]byte("x"), 33), decision.Constrain, []decision.Constraint{{Key: "bytes", Operator: "max", Value: "32"}}) - if err := constrainedVerifier.VerifyGovernedStreamInit(context.Background(), coreapi.Addr{}, tooLarge); err == nil || !strings.Contains(err.Error(), "numeric stream constraint") { - t.Fatalf("oversize stream constraint error=%v", err) - } -} - -func TestGovernedStreamDisclosureBindingAndRequiredProfile(t *testing.T) { - stream, verifier := newGovernedStreamForTest(t, "export.txt", []byte("sensitive export"), decision.Allow, nil) - verifier.RequireDisclosure = true - if err := verifier.VerifyGovernedStreamInit(context.Background(), coreapi.Addr{}, stream); err == nil || !strings.Contains(err.Error(), "disclosure is required") { - t.Fatalf("stream without required disclosure err=%v", err) - } - - payload := []byte("sensitive export") - fullHash := sha256.Sum256(payload) - initPayload, err := BuildStreamInitPayload("export.txt", int64(len(payload)), fullHash) - if err != nil { - t.Fatal(err) - } - disclosure := decision.DisclosureBinding{ - Version: decision.DisclosureBindingVersion, ContentHash: decision.HashPayload(payload), DeclaredBytes: uint64(len(payload)), - ContentType: "text/plain", Labels: []string{"confidential", "pii"}, Recipient: "agent:receiver", - Purpose: "customer-support", Residency: "eu-west-1", Filename: "export.txt", TransferID: fmt.Sprintf("%x", fullHash[:transferIDLen]), - } - disclosureHash, err := disclosure.Hash() - if err != nil { - t.Fatal(err) - } - intent := decision.Intent{ - Version: decision.SchemaVersion, ID: "governed-stream-disclosure-intent", TenantID: "tenant-a", AgentID: "sender-a", - Action: "file.share", Resource: "agent:receiver/inbox", Audience: disclosure.Recipient, Purpose: disclosure.Purpose, - PayloadHash: disclosureHash, Risk: decision.RiskHigh, IssuedAt: 1785500000, ExpiresAt: 1785500060, - Nonce: strings.Repeat("a", 32), KeyID: "sender-key", Signature: "test-signature", - } - intentHash, err := intent.Hash() - if err != nil { - t.Fatal(err) - } - result := decision.Decision{ - Version: decision.SchemaVersion, ID: "governed-stream-disclosure-decision", IntentHash: intentHash, TenantID: intent.TenantID, AgentID: intent.AgentID, - Outcome: decision.Allow, PolicyRevision: 7, RevocationEpoch: 3, ProviderID: "authority-a", IssuedAt: 1785500000, ExpiresAt: 1785500060, - KeyID: "authority-key", Signature: "test-signature", - } - governed, err := NewGovernedStreamInitWithDisclosure(&Frame{Type: TypeFileStream, Payload: initPayload}, intent, result, disclosure) - if err != nil { - t.Fatal(err) - } - encoded, err := EncodeGovernedStreamInit(governed) - if err != nil { - t.Fatal(err) - } - decoded, err := DecodeGovernedStreamInit(encoded) - if err != nil || decoded.Disclosure == nil || decoded.Disclosure.TransferID != disclosure.TransferID { - t.Fatalf("decode disclosure stream err=%v decoded=%+v", err, decoded) - } - decoded.Disclosure.Residency = "us-east-1" - if err := decoded.Validate(); err == nil || !strings.Contains(err.Error(), "binding") { - t.Fatalf("residency mutation accepted: %v", err) - } -} - -func TestRequireGovernedServicePersistsOnlyVerifiedFrames(t *testing.T) { - tmp := t.TempDir() - governed, verifier := newGovernedTestFrame(t, &Frame{Type: TypeText, Payload: []byte("approved")}, decision.Allow, nil) - w, r, wait := makeServiceConn(t, ServiceConfig{InboxDir: tmp, RequireGoverned: true, GovernedVerifier: verifier}) - defer wait() - - envelope, err := EncodeGovernedFrame(governed) - if err != nil { - t.Fatalf("encode governed frame: %v", err) - } - if err := WriteFrame(w, envelope); err != nil { - t.Fatalf("write governed frame: %v", err) - } - ack, err := ReadFrame(r) - if err != nil { - t.Fatalf("read governed ack: %v", err) - } - if !strings.Contains(string(ack.Payload), "ACK TEXT 8 bytes") { - t.Fatalf("governed ack = %q", ack.Payload) - } - entries, err := os.ReadDir(tmp) - if err != nil || len(entries) != 1 { - t.Fatalf("inbox after governed delivery: entries=%d err=%v", len(entries), err) - } - - if err := WriteFrame(w, &Frame{Type: TypeText, Payload: []byte("unsigned")}); err != nil { - t.Fatalf("write unsigned frame: %v", err) - } - ack, err = ReadFrame(r) - if err != nil { - t.Fatalf("read unsigned ack: %v", err) - } - if !strings.Contains(string(ack.Payload), "ERR TEXT save failed: unsigned legacy frame rejected") { - t.Fatalf("unsigned ack = %q", ack.Payload) - } - entries, err = os.ReadDir(tmp) - if err != nil || len(entries) != 1 { - t.Fatalf("unsigned delivery changed inbox: entries=%d err=%v", len(entries), err) - } -} - -func TestGovernedReceiptIsRequiredBeforeDeliveryAcknowledgement(t *testing.T) { - governed, verifier := newGovernedTestFrame(t, &Frame{Type: TypeText, Payload: []byte("approved")}, decision.Allow, nil) - envelope, err := EncodeGovernedFrame(governed) - if err != nil { - t.Fatal(err) - } - recorder := &governedReceiptRecorder{} - directory := t.TempDir() - w, r, wait := makeServiceConn(t, ServiceConfig{ - InboxDir: directory, RequireGoverned: true, GovernedVerifier: verifier, - RequireGovernedReceipts: true, GovernedReceiptRecorder: recorder, - }) - defer wait() - if err := WriteFrame(w, envelope); err != nil { - t.Fatal(err) - } - ack, err := ReadFrame(r) - if err != nil || !strings.Contains(string(ack.Payload), "ACK TEXT 8 bytes") { - t.Fatalf("receipt-backed acknowledgement=%q err=%v", ack.Payload, err) - } - if recorder.calls != 1 || recorder.intent.ID != governed.Intent.ID || recorder.decision.ID != governed.Decision.ID { - t.Fatalf("receipt recorder=%+v", recorder) - } - entries, err := os.ReadDir(directory) - if err != nil || len(entries) != 1 { - t.Fatalf("receipt-backed delivery entries=%d err=%v", len(entries), err) - } - - failingDirectory := t.TempDir() - failing := &governedReceiptRecorder{err: os.ErrPermission} - w, r, wait = makeServiceConn(t, ServiceConfig{ - InboxDir: failingDirectory, RequireGoverned: true, GovernedVerifier: verifier, - RequireGovernedReceipts: true, GovernedReceiptRecorder: failing, - }) - defer wait() - if err := WriteFrame(w, envelope); err != nil { - t.Fatal(err) - } - ack, err = ReadFrame(r) - if err != nil || !strings.Contains(string(ack.Payload), "record governed delivery receipt") { - t.Fatalf("receipt failure acknowledgement=%q err=%v", ack.Payload, err) - } - entries, err = os.ReadDir(failingDirectory) - if err != nil || len(entries) != 0 { - t.Fatalf("unreceipted delivery remained on disk: entries=%d err=%v", len(entries), err) - } -} - -func TestGovernedStreamRequiresReceiptBeforeComplete(t *testing.T) { - payload := []byte("a governed resumable file") - stream, verifier := newGovernedStreamForTest(t, "report.bin", payload, decision.Allow, nil) - envelope, err := EncodeGovernedStreamInit(stream) - if err != nil { - t.Fatal(err) - } - id, err := stream.TransferID() - if err != nil { - t.Fatal(err) - } - recorder := &governedReceiptRecorder{} - directory := t.TempDir() - w, r, wait := makeServiceConn(t, ServiceConfig{ - ReceivedDir: directory, RequireGoverned: true, GovernedVerifier: verifier, GovernedStreamVerifier: verifier, - RequireGovernedReceipts: true, GovernedReceiptRecorder: recorder, - }) - defer wait() - if err := WriteFrame(w, envelope); err != nil { - t.Fatal(err) - } - initAck, err := ReadFrame(r) - if err != nil { - t.Fatal(err) - } - kind, gotID, _, ok := decodeStreamFrame(initAck) - if !ok || kind != streamKindInitAck || gotID != id { - t.Fatalf("governed stream init response=%#v", initAck) - } - if err := WriteFrame(w, encodeChunk(id, 0, payload)); err != nil { - t.Fatal(err) - } - if _, err := ReadFrame(r); err != nil { - t.Fatal(err) - } - if err := WriteFrame(w, encodeStreamFrame(streamKindDone, id, nil)); err != nil { - t.Fatal(err) - } - complete, err := ReadFrame(r) - if err != nil { - t.Fatal(err) - } - kind, _, body, ok := decodeStreamFrame(complete) - completeOK, message := decodeComplete(body) - if !ok || kind != streamKindComplete || !completeOK || message != "" { - t.Fatalf("governed stream completion=%#v ok=%v message=%q", complete, completeOK, message) - } - if recorder.calls != 1 || recorder.intent.ID != stream.Intent.ID || recorder.decision.ID != stream.Decision.ID { - t.Fatalf("stream receipt recorder=%+v", recorder) - } - entries, err := os.ReadDir(directory) - if err != nil { - t.Fatal(err) - } - var saved string - for _, entry := range entries { - if !entry.IsDir() { - saved = entry.Name() - } - } - contents, err := os.ReadFile(filepath.Join(directory, saved)) - if err != nil || string(contents) != string(payload) { - t.Fatalf("saved governed stream contents=%q err=%v", contents, err) - } - - failingDirectory := t.TempDir() - w, r, wait = makeServiceConn(t, ServiceConfig{ - ReceivedDir: failingDirectory, RequireGoverned: true, GovernedVerifier: verifier, GovernedStreamVerifier: verifier, - RequireGovernedReceipts: true, GovernedReceiptRecorder: &governedReceiptRecorder{err: os.ErrPermission}, - }) - defer wait() - if err := WriteFrame(w, envelope); err != nil { - t.Fatal(err) - } - if _, err := ReadFrame(r); err != nil { - t.Fatal(err) - } - if err := WriteFrame(w, encodeChunk(id, 0, payload)); err != nil { - t.Fatal(err) - } - if _, err := ReadFrame(r); err != nil { - t.Fatal(err) - } - if err := WriteFrame(w, encodeStreamFrame(streamKindDone, id, nil)); err != nil { - t.Fatal(err) - } - complete, err = ReadFrame(r) - if err != nil { - t.Fatal(err) - } - kind, _, body, ok = decodeStreamFrame(complete) - completeOK, message = decodeComplete(body) - if !ok || kind != streamKindComplete || completeOK || !strings.Contains(message, "record governed stream receipt") { - t.Fatalf("receipt-failed stream completion=%#v ok=%v message=%q", complete, completeOK, message) - } - entries, err = os.ReadDir(failingDirectory) - if err != nil { - t.Fatal(err) - } - for _, entry := range entries { - if !entry.IsDir() { - t.Fatalf("unreceipted stream file remained: %s", entry.Name()) - } - } - - rawDirectory := t.TempDir() - w, r, wait = makeServiceConn(t, ServiceConfig{ - ReceivedDir: rawDirectory, RequireGoverned: true, GovernedVerifier: verifier, GovernedStreamVerifier: verifier, - RequireGovernedReceipts: true, GovernedReceiptRecorder: &governedReceiptRecorder{}, - }) - defer wait() - if err := WriteFrame(w, stream.InitFrame()); err != nil { - t.Fatal(err) - } - rejected, err := ReadFrame(r) - if err != nil { - t.Fatal(err) - } - kind, _, body, ok = decodeStreamFrame(rejected) - completeOK, message = decodeComplete(body) - if !ok || kind != streamKindComplete || completeOK || !strings.Contains(message, "unsigned stream frame") { - t.Fatalf("raw stream rejection=%#v ok=%v message=%q", rejected, completeOK, message) - } -} - -func TestGovernedReceiptRequirementFailsStartupWithoutRecorder(t *testing.T) { - service := NewService(ServiceConfig{RequireGoverned: true, RequireGovernedReceipts: true}) - if err := service.Start(context.Background(), coreapi.Deps{}); err == nil || !strings.Contains(err.Error(), "receipt recorder") { - t.Fatalf("start error=%v, want missing receipt recorder", err) - } -} - -var _ decision.TrustStore = governedTestTrust{} -var _ decision.AuthorityCeiling = governedTestCeiling{} -var _ GovernedReceiptRecorder = (*governedReceiptRecorder)(nil) diff --git a/retention.go b/retention.go deleted file mode 100644 index 4eee828..0000000 --- a/retention.go +++ /dev/null @@ -1,295 +0,0 @@ -// SPDX-License-Identifier: AGPL-3.0-or-later - -package dataexchange - -import ( - "crypto/sha256" - "encoding/hex" - "encoding/json" - "fmt" - "io" - "log/slog" - "os" - "path/filepath" - "strings" - "time" - - "github.com/pilot-protocol/common/decision" -) - -const ( - retentionEntryVersion uint16 = 1 - defaultRetentionSweep = time.Minute - minRetentionDuration = time.Second - maxRetentionDuration = 10 * 365 * 24 * time.Hour - maxRetentionPolicies = 32 -) - -// GovernedRetentionPolicy maps one V2 signed retention class to a local -// retention duration. The class is selected by signed policy and bound to the -// Intent; the duration is attachment-controlled, so a sender cannot extend it. -type GovernedRetentionPolicy struct { - Class string - RetainFor time.Duration -} - -type governedRetentionEntry struct { - Version uint16 `json:"version"` - Root string `json:"root"` - RelativePath string `json:"relative_path"` - RetentionClass string `json:"retention_class"` - DisclosureHash string `json:"disclosure_hash"` - ExpiresAt int64 `json:"expires_at"` -} - -type governedRetentionManager struct { - stateDir string - roots map[string]string - policies map[string]time.Duration - now func() time.Time -} - -type retentionTicket struct { - entryPath string -} - -func newGovernedRetentionManager(stateDir string, roots map[string]string, policies []GovernedRetentionPolicy, now func() time.Time) (*governedRetentionManager, error) { - if len(policies) == 0 || len(policies) > maxRetentionPolicies { - return nil, fmt.Errorf("dataexchange: retention needs 1-%d policies", maxRetentionPolicies) - } - if strings.TrimSpace(stateDir) == "" { - return nil, fmt.Errorf("dataexchange: retention state directory is required") - } - if len(roots) == 0 { - return nil, fmt.Errorf("dataexchange: retention roots are required") - } - cleanRoots := make(map[string]string, len(roots)) - for name, root := range roots { - if name == "" || strings.TrimSpace(root) == "" || !filepath.IsAbs(root) { - return nil, fmt.Errorf("dataexchange: invalid retention root") - } - cleanRoots[name] = filepath.Clean(root) - } - policyMap := make(map[string]time.Duration, len(policies)) - for _, policy := range policies { - if !validRetentionClass(policy.Class) || policy.RetainFor < minRetentionDuration || policy.RetainFor > maxRetentionDuration { - return nil, fmt.Errorf("dataexchange: invalid retention policy") - } - if _, exists := policyMap[policy.Class]; exists { - return nil, fmt.Errorf("dataexchange: duplicate retention class %q", policy.Class) - } - policyMap[policy.Class] = policy.RetainFor - } - if now == nil { - now = time.Now - } - if err := os.MkdirAll(stateDir, 0700); err != nil { - return nil, fmt.Errorf("dataexchange: create retention state directory: %w", err) - } - return &governedRetentionManager{stateDir: filepath.Clean(stateDir), roots: cleanRoots, policies: policyMap, now: now}, nil -} - -func (manager *governedRetentionManager) prepare(disclosure *decision.DisclosureBinding, path string) (retentionTicket, error) { - if manager == nil { - return retentionTicket{}, nil - } - if disclosure == nil || disclosure.Version != decision.DisclosureBindingRetentionVersion || disclosure.RetentionClass == "" { - return retentionTicket{}, fmt.Errorf("dataexchange: governed retention requires a V2 disclosure retention class") - } - retention, exists := manager.policies[disclosure.RetentionClass] - if !exists { - return retentionTicket{}, fmt.Errorf("dataexchange: disclosure retention class is not configured") - } - rootName, relativePath, err := manager.relativePath(path) - if err != nil { - return retentionTicket{}, err - } - disclosureHash, err := disclosure.Hash() - if err != nil { - return retentionTicket{}, err - } - entry := governedRetentionEntry{ - Version: retentionEntryVersion, Root: rootName, RelativePath: relativePath, - RetentionClass: disclosure.RetentionClass, DisclosureHash: disclosureHash, - ExpiresAt: manager.now().UTC().Add(retention).Unix(), - } - encoded, err := json.Marshal(entry) - if err != nil { - return retentionTicket{}, fmt.Errorf("dataexchange: encode retention entry: %w", err) - } - sum := sha256.Sum256([]byte(rootName + "\x00" + relativePath)) - entryPath := filepath.Join(manager.stateDir, hex.EncodeToString(sum[:])+".json") - if err := writeRetentionEntry(entryPath, encoded); err != nil { - return retentionTicket{}, err - } - return retentionTicket{entryPath: entryPath}, nil -} - -func (ticket retentionTicket) rollback() error { - if ticket.entryPath == "" { - return nil - } - if err := os.Remove(ticket.entryPath); err != nil && !os.IsNotExist(err) { - return fmt.Errorf("dataexchange: remove retention entry: %w", err) - } - return nil -} - -// Sweep deletes expired governed content and only then removes its durable -// journal entry. A malformed journal causes an error instead of silently -// disabling a retention obligation after restart. -func (manager *governedRetentionManager) Sweep() error { - if manager == nil { - return nil - } - entries, err := os.ReadDir(manager.stateDir) - if err != nil { - return fmt.Errorf("dataexchange: read retention state: %w", err) - } - now := manager.now().UTC().Unix() - for _, file := range entries { - if file.IsDir() || !strings.HasSuffix(file.Name(), ".json") { - continue - } - entryPath := filepath.Join(manager.stateDir, file.Name()) - entry, err := readRetentionEntry(entryPath) - if err != nil { - return err - } - if err := manager.validateEntry(entry); err != nil { - return err - } - if entry.ExpiresAt > now { - continue - } - path, err := manager.pathFor(entry.Root, entry.RelativePath) - if err != nil { - return err - } - if err := os.Remove(path); err != nil && !os.IsNotExist(err) { - return fmt.Errorf("dataexchange: expire retained content: %w", err) - } - if err := os.Remove(entryPath); err != nil && !os.IsNotExist(err) { - return fmt.Errorf("dataexchange: remove expired retention entry: %w", err) - } - } - return nil -} - -func (manager *governedRetentionManager) run(stop <-chan struct{}, interval time.Duration) { - if interval <= 0 { - interval = defaultRetentionSweep - } - ticker := time.NewTicker(interval) - defer ticker.Stop() - for { - select { - case <-stop: - return - case <-ticker.C: - if err := manager.Sweep(); err != nil { - // Keep retrying. Startup already verifies the journal, and a - // transient removal error must not kill future retention work. - slog.Error("governed retention sweep failed", "error", err) - continue - } - } - } -} - -func (manager *governedRetentionManager) relativePath(path string) (string, string, error) { - cleanPath := filepath.Clean(path) - for name, root := range manager.roots { - relativePath, err := filepath.Rel(root, cleanPath) - if err == nil && relativePath != "." && relativePath != ".." && !strings.HasPrefix(relativePath, ".."+string(filepath.Separator)) && !filepath.IsAbs(relativePath) { - return name, relativePath, nil - } - } - return "", "", fmt.Errorf("dataexchange: retention path is outside managed roots") -} - -func (manager *governedRetentionManager) pathFor(rootName, relativePath string) (string, error) { - root, exists := manager.roots[rootName] - if !exists || relativePath == "" || filepath.IsAbs(relativePath) { - return "", fmt.Errorf("dataexchange: invalid retention entry path") - } - path := filepath.Join(root, relativePath) - _, checkedRelative, err := manager.relativePath(path) - if err != nil || checkedRelative != filepath.Clean(relativePath) { - return "", fmt.Errorf("dataexchange: invalid retention entry path") - } - return path, nil -} - -func (manager *governedRetentionManager) validateEntry(entry governedRetentionEntry) error { - if entry.Version != retentionEntryVersion || !validRetentionClass(entry.RetentionClass) || len(entry.DisclosureHash) != 64 || entry.ExpiresAt <= 0 { - return fmt.Errorf("dataexchange: invalid retention entry") - } - for _, character := range entry.DisclosureHash { - if !(character >= '0' && character <= '9' || character >= 'a' && character <= 'f') { - return fmt.Errorf("dataexchange: invalid retention entry") - } - } - _, err := manager.pathFor(entry.Root, entry.RelativePath) - return err -} - -func writeRetentionEntry(path string, data []byte) error { - temporary, err := os.CreateTemp(filepath.Dir(path), ".retention-*") - if err != nil { - return fmt.Errorf("dataexchange: create retention entry: %w", err) - } - temporaryPath := temporary.Name() - defer os.Remove(temporaryPath) - if err := temporary.Chmod(0600); err != nil { - _ = temporary.Close() - return fmt.Errorf("dataexchange: protect retention entry: %w", err) - } - if _, err := temporary.Write(data); err != nil { - _ = temporary.Close() - return fmt.Errorf("dataexchange: write retention entry: %w", err) - } - if err := temporary.Sync(); err != nil { - _ = temporary.Close() - return fmt.Errorf("dataexchange: sync retention entry: %w", err) - } - if err := temporary.Close(); err != nil { - return fmt.Errorf("dataexchange: close retention entry: %w", err) - } - if err := os.Rename(temporaryPath, path); err != nil { - return fmt.Errorf("dataexchange: publish retention entry: %w", err) - } - return nil -} - -func readRetentionEntry(path string) (governedRetentionEntry, error) { - file, err := os.Open(path) - if err != nil { - return governedRetentionEntry{}, fmt.Errorf("dataexchange: read retention entry: %w", err) - } - defer file.Close() - decoder := json.NewDecoder(io.LimitReader(file, 4<<10)) - decoder.DisallowUnknownFields() - var entry governedRetentionEntry - if err := decoder.Decode(&entry); err != nil { - return governedRetentionEntry{}, fmt.Errorf("dataexchange: decode retention entry: %w", err) - } - var trailing any - if err := decoder.Decode(&trailing); err != io.EOF { - return governedRetentionEntry{}, fmt.Errorf("dataexchange: invalid retention entry") - } - return entry, nil -} - -func validRetentionClass(value string) bool { - if len(value) == 0 || len(value) > 64 || value[0] == '-' || value[len(value)-1] == '-' { - return false - } - for index, character := range value { - if (character >= 'a' && character <= 'z') || (character >= '0' && character <= '9') || (character == '-' && index > 0 && index+1 < len(value)) { - continue - } - return false - } - return true -} diff --git a/retention_test.go b/retention_test.go deleted file mode 100644 index e04f9d3..0000000 --- a/retention_test.go +++ /dev/null @@ -1,171 +0,0 @@ -// SPDX-License-Identifier: AGPL-3.0-or-later - -//go:build !no_dataexchange -// +build !no_dataexchange - -package dataexchange - -import ( - "crypto/sha256" - "os" - "path/filepath" - "testing" - "time" - - "github.com/pilot-protocol/common/decision" - "github.com/pilot-protocol/common/protocol" -) - -func TestGovernedRetentionManagerDurablyExpiresOnlyManagedContent(t *testing.T) { - root := t.TempDir() - inbox := filepath.Join(root, "inbox") - received := filepath.Join(root, "received") - state := filepath.Join(root, "retention") - if err := os.MkdirAll(inbox, 0700); err != nil { - t.Fatal(err) - } - if err := os.MkdirAll(received, 0700); err != nil { - t.Fatal(err) - } - path := filepath.Join(inbox, "message.json") - if err := os.WriteFile(path, []byte("classified"), 0600); err != nil { - t.Fatal(err) - } - now := time.Unix(1785500000, 0).UTC() - manager, err := newGovernedRetentionManager(state, map[string]string{"inbox": inbox, "received": received}, []GovernedRetentionPolicy{{Class: "finance-7y", RetainFor: time.Second}}, func() time.Time { return now }) - if err != nil { - t.Fatal(err) - } - disclosure := decision.DisclosureBinding{ - Version: decision.DisclosureBindingRetentionVersion, ContentHash: decision.HashPayload([]byte("classified")), DeclaredBytes: 10, - ContentType: "application/json", Labels: []string{"finance"}, Recipient: "agent:finance", Purpose: "inbox-delivery", Residency: "eu-west-1", RetentionClass: "finance-7y", - } - ticket, err := manager.prepare(&disclosure, path) - if err != nil || ticket.entryPath == "" { - t.Fatalf("prepare ticket=%+v err=%v", ticket, err) - } - if err := manager.Sweep(); err != nil { - t.Fatal(err) - } - if _, err := os.Stat(path); err != nil { - t.Fatalf("content deleted before expiry: %v", err) - } - now = now.Add(time.Second) - if err := manager.Sweep(); err != nil { - t.Fatal(err) - } - if _, err := os.Stat(path); !os.IsNotExist(err) { - t.Fatalf("expired content remained: %v", err) - } - if _, err := os.Stat(ticket.entryPath); !os.IsNotExist(err) { - t.Fatalf("expired journal remained: %v", err) - } -} - -func TestGovernedRetentionManagerRejectsUnknownClassAndEscape(t *testing.T) { - root := t.TempDir() - inbox := filepath.Join(root, "inbox") - if err := os.MkdirAll(inbox, 0700); err != nil { - t.Fatal(err) - } - manager, err := newGovernedRetentionManager(filepath.Join(root, "retention"), map[string]string{"inbox": inbox}, []GovernedRetentionPolicy{{Class: "finance-7y", RetainFor: time.Hour}}, time.Now) - if err != nil { - t.Fatal(err) - } - disclosure := &decision.DisclosureBinding{Version: decision.DisclosureBindingRetentionVersion, RetentionClass: "finance-30d"} - if _, err := manager.prepare(disclosure, filepath.Join(inbox, "message.json")); err == nil { - t.Fatal("unknown retention class was accepted") - } - disclosure.RetentionClass = "finance-7y" - if _, err := manager.prepare(disclosure, filepath.Join(root, "outside.json")); err == nil { - t.Fatal("retention path escape was accepted") - } -} - -func TestServicePersistsAndExpiresGovernedRetention(t *testing.T) { - root := t.TempDir() - inbox := filepath.Join(root, "inbox") - received := filepath.Join(root, "received") - state := filepath.Join(root, "retention") - now := time.Unix(1785500000, 0).UTC() - service := NewService(ServiceConfig{ - InboxDir: inbox, ReceivedDir: received, RequireGoverned: true, - GovernedRetentionPolicies: []GovernedRetentionPolicy{{Class: "finance-7y", RetainFor: time.Second}}, - RetentionStateDir: state, - }) - if err := service.initializeGovernedRetention(); err != nil { - t.Fatal(err) - } - service.retention.now = func() time.Time { return now } - disclosure := &decision.DisclosureBinding{ - Version: decision.DisclosureBindingRetentionVersion, ContentHash: decision.HashPayload([]byte("classified")), DeclaredBytes: 10, - ContentType: "text/plain", Labels: []string{"finance"}, Recipient: "agent:finance", Purpose: "inbox-delivery", Residency: "eu-west-1", RetentionClass: "finance-7y", - } - if err := service.requireGovernedRetention(disclosure); err != nil { - t.Fatal(err) - } - delivery, err := service.prepareInboxMessage(&Frame{Type: TypeText, Payload: []byte("classified")}, protocol.Addr{Node: 1}, disclosure) - if err != nil { - t.Fatal(err) - } - delivery.commit() - files, err := os.ReadDir(inbox) - if err != nil || len(files) != 1 { - t.Fatalf("retained inbox files=%v err=%v", files, err) - } - now = now.Add(time.Second) - if err := service.retention.Sweep(); err != nil { - t.Fatal(err) - } - files, err = os.ReadDir(inbox) - if err != nil || len(files) != 0 { - t.Fatalf("expired inbox files=%v err=%v", files, err) - } -} - -func TestStreamRetentionPreparationRunsBeforeFinalRename(t *testing.T) { - dir := t.TempDir() - payload := []byte("classified") - hash := sha256.Sum256(payload) - var id [transferIDLen]byte - copy(id[:], hash[:transferIDLen]) - prepared, committed := false, false - receiver := NewStreamReceiverWithQuotaAndPrepareAndCommit( - dir, - func(string) string { return "final.txt" }, - nil, - func(_ [transferIDLen]byte, _ string, path string, _ int64) error { - if _, err := os.Stat(path); !os.IsNotExist(err) { - t.Fatalf("final path was visible before preparation: %v", err) - } - prepared = true - return nil - }, - func(_ [transferIDLen]byte, _ string, path string, _ int64) error { - if !prepared { - t.Fatal("commit ran before preparation") - } - if _, err := os.Stat(path); err != nil { - t.Fatalf("final path was not visible at commit: %v", err) - } - committed = true - return nil - }, - 0, - ) - if response := receiver.HandleFrame(encodeInit(id, uint64(len(payload)), hash, uint32(StreamChunkSize), "source.txt")); response == nil { - t.Fatal("INIT did not produce a response") - } - if response := receiver.HandleFrame(encodeChunk(id, 0, payload)); response == nil { - t.Fatal("CHUNK did not produce a response") - } - response := receiver.HandleFrame(encodeStreamFrame(streamKindDone, id, nil)) - if response == nil { - t.Fatal("DONE did not produce completion") - } - _, _, body, ok := decodeStreamFrame(response) - complete, _ := decodeComplete(body) - if !ok || !complete || !prepared || !committed { - t.Fatalf("stream lifecycle response=%+v prepared=%v committed=%v", response, prepared, committed) - } -} diff --git a/service.go b/service.go index 84cfd0a..d505794 100644 --- a/service.go +++ b/service.go @@ -6,7 +6,6 @@ package dataexchange import ( - "bytes" "context" "encoding/base64" "encoding/json" @@ -19,7 +18,6 @@ import ( "unicode/utf8" "github.com/pilot-protocol/common/coreapi" - "github.com/pilot-protocol/common/decision" "github.com/pilot-protocol/common/protocol" ) @@ -72,31 +70,6 @@ type ServiceConfig struct { // instead of pinning the goroutine and its buffers indefinitely. // Zero ⇒ DefaultIdleTimeout; a negative value disables the deadline. IdleTimeout time.Duration - // RequireGoverned rejects legacy frames unless they carry a signed - // governed envelope verified by GovernedVerifier. Enable this only after - // sender rollout; the default preserves compatibility with older peers. - RequireGoverned bool - GovernedVerifier GovernedFrameVerifier - GovernedStreamVerifier GovernedStreamVerifier - RequireGovernedReceipts bool - GovernedReceiptRecorder GovernedReceiptRecorder - // GovernedContentInspector runs only at this receiver, after signed - // governed verification and before a message/file is released. It receives - // a reader, not an exported payload copy, so central authority services do - // not need application plaintext for DLP. - GovernedContentInspector decision.DisclosureContentInspector - RequireGovernedContentInspection bool - // GovernedTransferQuota bounds admitted transfers for each verified - // Intent.AgentID. It never uses an untrusted peer address as the subject. - // Quota is charged at governed-frame verification (or stream INIT) and does - // not silently apply to legacy traffic. - GovernedTransferQuota *decision.TransferQuotaLimiter - // GovernedRetentionPolicies maps signed V2 disclosure retention classes to - // durable local expiry work. When non-empty, governed deliveries without a - // configured V2 retention class are rejected before persistence. - GovernedRetentionPolicies []GovernedRetentionPolicy - RetentionStateDir string - RetentionSweepInterval time.Duration } // Defensible defaults applied when the corresponding ServiceConfig field @@ -167,15 +140,13 @@ const inboxEvictCheckEvery = 64 // Service is the L11 plugin adapter. Daemon (L7) holds it only as // coreapi.Service; cmd/daemon/main.go (L12) constructs it. type Service struct { - cfg ServiceConfig - listener coreapi.Listener - deps coreapi.Deps - cancel context.CancelFunc - done chan struct{} - seq atomic.Uint64 - retention *governedRetentionManager - replay *governedReplayGuard - inbox inboxBudget + cfg ServiceConfig + listener coreapi.Listener + deps coreapi.Deps + cancel context.CancelFunc + done chan struct{} + seq atomic.Uint64 + inbox inboxBudget // dedupeByID remembers stored frames that carried a MessageID; // dedupeByContent (nil unless DedupeContentWindow > 0) remembers frames // without one. See dedupe.go. @@ -186,17 +157,15 @@ type Service struct { afterInboxWrite func(path string) } -// persistedDelivery separates a disk write from its visible notification so a -// governed receipt can be durably appended before the receiver acknowledges -// or publishes the delivery event. A failed receipt causes the staged file to -// be removed rather than leaving an unreceipted enterprise side effect. +// persistedDelivery separates a disk write from its visible notification: +// commit logs and publishes the delivery, rollback removes the staged file. type persistedDelivery struct { rollback func() error commit func() } func NewService(cfg ServiceConfig) *Service { - s := &Service{cfg: cfg, replay: newGovernedReplayGuard()} + s := &Service{cfg: cfg} if window := s.effectiveDedupeWindow(); window > 0 { s.dedupeByID = newDeliveryDedupe(window, dedupeMaxEntries) } @@ -212,13 +181,7 @@ func (s *Service) Name() string { return "dataexchange" } func (s *Service) Order() int { return 110 } func (s *Service) Start(ctx context.Context, deps coreapi.Deps) error { - if err := s.validateGovernedConfig(); err != nil { - return err - } s.deps = deps - if err := s.initializeGovernedRetention(); err != nil { - return err - } ln, err := deps.Streams.Listen(protocol.PortDataExchange) if err != nil { return fmt.Errorf("dataexchange: listen on port %d: %w", protocol.PortDataExchange, err) @@ -229,77 +192,10 @@ func (s *Service) Start(ctx context.Context, deps coreapi.Deps) error { s.cancel = cancel s.done = make(chan struct{}) go s.acceptLoop(runCtx) - if s.retention != nil { - go s.retention.run(runCtx.Done(), s.cfg.RetentionSweepInterval) - } slog.Info("dataexchange service listening", "port", protocol.PortDataExchange) return nil } -func (s *Service) validateGovernedConfig() error { - if s.cfg.RequireGovernedReceipts && (!s.cfg.RequireGoverned || s.cfg.GovernedReceiptRecorder == nil) { - return fmt.Errorf("dataexchange: governed receipts require a governed receiver and receipt recorder") - } - if s.cfg.RequireGovernedContentInspection && (!s.cfg.RequireGoverned || s.cfg.GovernedContentInspector == nil) { - return fmt.Errorf("dataexchange: required content inspection needs a governed receiver and local inspector") - } - if s.cfg.GovernedTransferQuota != nil && !s.cfg.RequireGoverned { - return fmt.Errorf("dataexchange: governed transfer quota requires a governed receiver") - } - if len(s.cfg.GovernedRetentionPolicies) > 0 && !s.cfg.RequireGoverned { - return fmt.Errorf("dataexchange: governed retention requires a governed receiver") - } - return nil -} - -func (s *Service) initializeGovernedRetention() error { - if len(s.cfg.GovernedRetentionPolicies) == 0 { - s.retention = nil - return nil - } - inbox, err := s.inboxDir() - if err != nil { - return fmt.Errorf("dataexchange: retention inbox directory: %w", err) - } - received, err := s.receivedDir() - if err != nil { - return fmt.Errorf("dataexchange: retention received directory: %w", err) - } - stateDir := s.cfg.RetentionStateDir - if stateDir == "" { - stateDir = filepath.Join(filepath.Dir(inbox), "retention") - } - manager, err := newGovernedRetentionManager(stateDir, map[string]string{"inbox": inbox, "received": received}, s.cfg.GovernedRetentionPolicies, nil) - if err != nil { - return err - } - if err := manager.Sweep(); err != nil { - return err - } - s.retention = manager - return nil -} - -func (s *Service) requireGovernedRetention(disclosure *decision.DisclosureBinding) error { - if s.retention == nil { - return nil - } - if disclosure == nil || disclosure.Version != decision.DisclosureBindingRetentionVersion || disclosure.RetentionClass == "" { - return fmt.Errorf("dataexchange: governed retention requires a V2 disclosure retention class") - } - if _, exists := s.retention.policies[disclosure.RetentionClass]; !exists { - return fmt.Errorf("dataexchange: disclosure retention class is not configured") - } - return nil -} - -func (s *Service) prepareGovernedRetention(disclosure *decision.DisclosureBinding, path string) (retentionTicket, error) { - if s.retention == nil { - return retentionTicket{}, nil - } - return s.retention.prepare(disclosure, path) -} - func (s *Service) Stop(ctx context.Context) error { if s.cancel != nil { s.cancel() @@ -352,17 +248,12 @@ func (s *Service) handleConn(ctx context.Context, conn coreapi.Stream) { // Final filenames and the file.received event match saveReceivedFile so // the two transfer paths are indistinguishable to consumers. var sr *StreamReceiver - governedStreams := make(map[[transferIDLen]byte]GovernedStreamInit) - streamRetention := make(map[[transferIDLen]byte]retentionTicket) // claim is the dedupe claim for the frame being processed. If handling // unwinds early (a panic), releasing it as "not stored" lets any // identical delivery waiting on it proceed instead of hanging. var claim dedupeClaim defer func() { claim.finish(false, "") - for _, ticket := range streamRetention { - _ = ticket.rollback() - } if sr != nil { sr.Close() } @@ -382,55 +273,6 @@ func (s *Service) handleConn(ctx context.Context, conn coreapi.Stream) { }) } } - streamOnPrepare := func(id [transferIDLen]byte, _ string, path string, _ int64) error { - governed, governedTransfer := governedStreams[id] - if s.cfg.RequireGoverned && !governedTransfer { - return fmt.Errorf("stream completion is not bound to a governed INIT") - } - if !governedTransfer { - return nil - } - ticket, err := s.prepareGovernedRetention(governed.Disclosure, path) - if err != nil { - return err - } - streamRetention[id] = ticket - return nil - } - streamOnCommit := func(id [transferIDLen]byte, name, path string, _ int64) error { - governed, governedTransfer := governedStreams[id] - if s.cfg.RequireGoverned && !governedTransfer { - return fmt.Errorf("stream completion is not bound to a governed INIT") - } - if !governedTransfer { - return nil - } - defer delete(governedStreams, id) - retention := streamRetention[id] - defer delete(streamRetention, id) - committed := false - defer func() { - if !committed { - _ = retention.rollback() - } - }() - if err := s.inspectGovernedStreamFile(ctx, governed, path, name); err != nil { - return err - } - if s.cfg.GovernedReceiptRecorder == nil { - if s.cfg.RequireGovernedReceipts { - return fmt.Errorf("governed receipt recorder is not configured") - } - committed = true - return nil - } - if err := recordGovernedReceipt(ctx, s.cfg.GovernedReceiptRecorder, governed.Intent, governed.Decision, governed.Disclosure); err != nil { - return fmt.Errorf("record governed stream receipt: %w", err) - } - committed = true - return nil - } - idle := s.effectiveIdleTimeout() dl, canDeadline := conn.(readDeadliner) for { @@ -452,8 +294,7 @@ func (s *Service) handleConn(ctx context.Context, conn coreapi.Stream) { "bytes", len(frame.Payload), "remote", conn.RemoteAddr()}, correlationAttrs(frame)...)...) - // Correlation metadata rides outside any governed envelope, so keep - // it to re-attach after the envelope is unwrapped below. + // Correlation metadata, re-attached to a trace frame's inner message. messageID, replyTo := frame.MessageID, frame.ReplyTo // Duplicate suppression: an identical delivery of a frame that was @@ -482,176 +323,71 @@ func (s *Service) handleConn(ctx context.Context, conn coreapi.Stream) { var ( saveErr error ackFrame *Frame - governed *GovernedFrame - streamed *GovernedStreamInit - delivery persistedDelivery ) - if frame.Type == TypeGoverned { - decoded, governErr := DecodeGovernedFrame(frame) - if governErr != nil { - saveErr = governErr - } else if s.cfg.GovernedVerifier == nil { - saveErr = fmt.Errorf("governed frame received but no verifier is configured") - } else if governErr = s.cfg.GovernedVerifier.VerifyGovernedFrame(ctx, conn.RemoteAddr(), decoded); governErr != nil { - saveErr = governErr - } else { - governed = &decoded - frame = governed.DataFrame() - frame.MessageID, frame.ReplyTo = messageID, replyTo - } - } else if frame.Type == TypeGovernedFileStream { - decoded, governErr := DecodeGovernedStreamInit(frame) - if governErr != nil { - saveErr = governErr - } else if s.cfg.GovernedStreamVerifier == nil { - saveErr = fmt.Errorf("governed stream received but no verifier is configured") - } else if governErr = s.cfg.GovernedStreamVerifier.VerifyGovernedStreamInit(ctx, conn.RemoteAddr(), decoded); governErr != nil { - saveErr = governErr - } else { - streamed = &decoded - frame = streamed.InitFrame() - } - } else if s.cfg.RequireGoverned && frame.Type != TypeFileStream { - saveErr = fmt.Errorf("unsigned legacy frame rejected by governed receiver") - } - if saveErr == nil { - if governed != nil { - saveErr = s.admitGovernedTransfer(governed.Intent, uint64(len(governed.Payload))) - } else if streamed != nil { - declaredBytes, declaredErr := streamed.DeclaredBytes() - if declaredErr != nil { - saveErr = declaredErr - } else { - saveErr = s.admitGovernedTransfer(streamed.Intent, declaredBytes) - } - } - } - if saveErr == nil { - if governed != nil { - saveErr = s.requireGovernedRetention(governed.Disclosure) - } else if streamed != nil { - saveErr = s.requireGovernedRetention(streamed.Disclosure) - } - } - if saveErr == nil { - if governed != nil && frame.Type != TypeFileStream { - saveErr = s.inspectGovernedFrame(ctx, *governed) - } - } - if saveErr == nil { - switch frame.Type { - case TypeFileStream: - // Chunked/resumable transfer. The receiver emits its own - // control responses (INIT-ACK / ACK / COMPLETE), so skip the - // generic per-frame ACK below. - if sr == nil { - dir, derr := s.receivedDir() - if derr != nil { - _ = WriteFrame(conn, &Frame{Type: TypeText, Payload: []byte("ERR received dir: " + derr.Error())}) - return - } - if mderr := os.MkdirAll(dir, 0700); mderr != nil { - _ = WriteFrame(conn, &Frame{Type: TypeText, Payload: []byte("ERR mkdir: " + mderr.Error())}) - return - } - sr = NewStreamReceiverWithQuotaAndPrepareAndCommit(dir, streamNameSuffix, streamOnSaved, streamOnPrepare, streamOnCommit, s.effectiveReceivedMaxBytes()) - } - kind, id, _, validStreamFrame := decodeStreamFrame(frame) - if !validStreamFrame { - _ = WriteFrame(conn, encodeComplete(id, false, "malformed stream frame")) - continue - } - if s.cfg.RequireGoverned { - if streamed == nil { - if kind == streamKindInit || !streamBound(governedStreams, id) { - _ = WriteFrame(conn, encodeComplete(id, false, "unsigned stream frame rejected by governed receiver")) - continue - } - } else if kind != streamKindInit { - _ = WriteFrame(conn, encodeComplete(id, false, "governed stream envelope must carry INIT")) - continue - } - } - if resp := sr.HandleFrame(frame); resp != nil { - responseKind, _, responseBody, responseValid := decodeStreamFrame(resp) - if streamed != nil && responseValid && responseKind == streamKindInitAck { - governedStreams[id] = *streamed - } - if responseValid && responseKind == streamKindComplete { - completeOK, _ := decodeComplete(responseBody) - if completeOK { - delete(governedStreams, id) - } - } - if werr := WriteFrame(conn, resp); werr != nil { - return - } + switch frame.Type { + case TypeFileStream: + // Chunked/resumable transfer. The receiver emits its own + // control responses (INIT-ACK / ACK / COMPLETE), so skip the + // generic per-frame ACK below. + if sr == nil { + dir, derr := s.receivedDir() + if derr != nil { + _ = WriteFrame(conn, &Frame{Type: TypeText, Payload: []byte("ERR received dir: " + derr.Error())}) + return } - if kind == streamKindAbort { - delete(governedStreams, id) + if mderr := os.MkdirAll(dir, 0700); mderr != nil { + _ = WriteFrame(conn, &Frame{Type: TypeText, Payload: []byte("ERR mkdir: " + mderr.Error())}) + return } + sr = NewStreamReceiverWithQuota(dir, streamNameSuffix, streamOnSaved, s.effectiveReceivedMaxBytes()) + } + _, id, _, validStreamFrame := decodeStreamFrame(frame) + if !validStreamFrame { + _ = WriteFrame(conn, encodeComplete(id, false, "malformed stream frame")) continue - case TypeFile: - if frame.Filename != "" { - if governed != nil { - delivery, saveErr = s.prepareReceivedFile(frame, governed.Disclosure) - } else { - saveErr = s.saveReceivedFile(frame) - } - } - case TypeText, TypeJSON, TypeBinary: - if governed != nil { - delivery, saveErr = s.prepareInboxMessage(frame, conn.RemoteAddr(), governed.Disclosure) - } else { - saveErr = s.saveInboxMessage(frame, conn.RemoteAddr()) - } - case TypeTrace: - tf, tferr := ReadTracePayload(frame) - if tferr != nil { - ackFrame = &Frame{ - Type: TypeText, - Payload: []byte(fmt.Sprintf("ERR trace parse: %v", tferr)), - } - } else { - innerFrame := &Frame{Type: tf.InnerType, Payload: tf.Payload, MessageID: messageID, ReplyTo: replyTo} - innerSaveErr := s.saveInboxMessage(innerFrame, conn.RemoteAddr()) - inboxWrittenAtNs := time.Now().UnixNano() - innerAck := fmt.Sprintf("ACK %s %d bytes", TypeName(tf.InnerType), len(tf.Payload)) - if innerSaveErr != nil { - innerAck = fmt.Sprintf("ERR %s save failed: %v", TypeName(tf.InnerType), innerSaveErr) - } - ackSentAtNs := time.Now().UnixNano() - timingJSON, _ := json.Marshal(map[string]interface{}{ - "sent_at_ns": tf.SentAtNs, - "received_at_ns": frameReceivedAtNs, - "inbox_written_at_ns": inboxWrittenAtNs, - "ack_sent_at_ns": ackSentAtNs, - "inner_ack": innerAck, - }) - ackFrame = &Frame{Type: TypeJSON, Payload: timingJSON} + } + if resp := sr.HandleFrame(frame); resp != nil { + if werr := WriteFrame(conn, resp); werr != nil { + return } - default: - saveErr = fmt.Errorf("unsupported frame type %d", frame.Type) } - } - if saveErr == nil && governed != nil && delivery.commit != nil { - if s.cfg.GovernedReceiptRecorder == nil { - if s.cfg.RequireGovernedReceipts { - saveErr = fmt.Errorf("governed receipt recorder is not configured") + continue + case TypeFile: + if frame.Filename != "" { + saveErr = s.saveReceivedFile(frame) + } + case TypeText, TypeJSON, TypeBinary: + saveErr = s.saveInboxMessage(frame, conn.RemoteAddr()) + case TypeTrace: + tf, tferr := ReadTracePayload(frame) + if tferr != nil { + ackFrame = &Frame{ + Type: TypeText, + Payload: []byte(fmt.Sprintf("ERR trace parse: %v", tferr)), } - } else if receiptErr := recordGovernedReceipt(ctx, s.cfg.GovernedReceiptRecorder, governed.Intent, governed.Decision, governed.Disclosure); receiptErr != nil { - if delivery.rollback != nil { - if rollbackErr := delivery.rollback(); rollbackErr != nil { - receiptErr = fmt.Errorf("%w; remove unreceipted delivery: %v", receiptErr, rollbackErr) - } + } else { + innerFrame := &Frame{Type: tf.InnerType, Payload: tf.Payload, MessageID: messageID, ReplyTo: replyTo} + innerSaveErr := s.saveInboxMessage(innerFrame, conn.RemoteAddr()) + inboxWrittenAtNs := time.Now().UnixNano() + innerAck := fmt.Sprintf("ACK %s %d bytes", TypeName(tf.InnerType), len(tf.Payload)) + if innerSaveErr != nil { + innerAck = fmt.Sprintf("ERR %s save failed: %v", TypeName(tf.InnerType), innerSaveErr) } - saveErr = fmt.Errorf("record governed delivery receipt: %w", receiptErr) + ackSentAtNs := time.Now().UnixNano() + timingJSON, _ := json.Marshal(map[string]interface{}{ + "sent_at_ns": tf.SentAtNs, + "received_at_ns": frameReceivedAtNs, + "inbox_written_at_ns": inboxWrittenAtNs, + "ack_sent_at_ns": ackSentAtNs, + "inner_ack": innerAck, + }) + ackFrame = &Frame{Type: TypeJSON, Payload: timingJSON} } + default: + // Includes the reserved TypeGoverned / TypeGovernedFileStream: + // refused like any type this receiver does not handle. + saveErr = fmt.Errorf("unsupported frame type %d", frame.Type) } - if saveErr == nil && delivery.commit != nil { - delivery.commit() - } - if ackFrame == nil { ackMsg := fmt.Sprintf("ACK %s %d bytes", TypeName(frame.Type), len(frame.Payload)) if saveErr != nil { @@ -673,74 +409,6 @@ func (s *Service) handleConn(ctx context.Context, conn coreapi.Stream) { } } -func (s *Service) admitGovernedTransfer(intent decision.Intent, bytes uint64) error { - if s.replay != nil { - if err := s.replay.admit(intent); err != nil { - slog.Warn("governed transfer replay rejected", "agent_id", intent.AgentID, "intent_id", intent.ID, "error", err) - return err - } - } - if s.cfg.GovernedTransferQuota == nil { - return nil - } - if err := s.cfg.GovernedTransferQuota.Allow(intent.AgentID, bytes); err != nil { - slog.Warn("governed transfer quota rejected", "agent_id", intent.AgentID, "bytes", bytes, "error", err) - return fmt.Errorf("governed transfer quota rejected") - } - return nil -} - -func streamBound(streams map[[transferIDLen]byte]GovernedStreamInit, id [transferIDLen]byte) bool { - _, exists := streams[id] - return exists -} - -func (s *Service) inspectGovernedFrame(ctx context.Context, governed GovernedFrame) error { - if s.cfg.GovernedContentInspector == nil { - return nil - } - contentType := contentTypeForFrame(governed.Type, governed.Disclosure) - if err := s.cfg.GovernedContentInspector.InspectDisclosureContent(ctx, governed.Intent, governed.Disclosure, contentType, governed.Filename, bytes.NewReader(governed.Payload)); err != nil { - slog.Warn("governed content inspection rejected", "action", governed.Intent.Action, "resource", governed.Intent.Resource, "error", err) - return fmt.Errorf("governed content inspection rejected") - } - return nil -} - -func (s *Service) inspectGovernedStreamFile(ctx context.Context, governed GovernedStreamInit, path string, filename string) error { - if s.cfg.GovernedContentInspector == nil { - return nil - } - file, err := os.Open(path) - if err != nil { - return fmt.Errorf("governed content inspection rejected") - } - defer file.Close() - contentType := contentTypeForFrame(TypeFile, governed.Disclosure) - // Stream the verified on-disk body directly. A scanner that cannot process - // the whole file must return an error; truncating the reader would create a - // bypass for sensitive content placed after an arbitrary byte boundary. - if err := s.cfg.GovernedContentInspector.InspectDisclosureContent(ctx, governed.Intent, governed.Disclosure, contentType, filename, file); err != nil { - slog.Warn("governed stream content inspection rejected", "action", governed.Intent.Action, "resource", governed.Intent.Resource, "error", err) - return fmt.Errorf("governed content inspection rejected") - } - return nil -} - -func contentTypeForFrame(frameType uint32, disclosure *decision.DisclosureBinding) string { - if disclosure != nil { - return disclosure.ContentType - } - switch frameType { - case TypeText: - return "text/plain" - case TypeJSON: - return "application/json" - default: - return "application/octet-stream" - } -} - // receivedDir returns the configured received-file directory or the // default ~/.pilot/received. func (s *Service) receivedDir() (string, error) { @@ -768,7 +436,7 @@ func (s *Service) inboxDir() (string, error) { } func (s *Service) saveReceivedFile(frame *Frame) error { - delivery, err := s.prepareReceivedFile(frame, nil) + delivery, err := s.prepareReceivedFile(frame) if err != nil { return err } @@ -776,7 +444,7 @@ func (s *Service) saveReceivedFile(frame *Frame) error { return nil } -func (s *Service) prepareReceivedFile(frame *Frame, disclosure *decision.DisclosureBinding) (persistedDelivery, error) { +func (s *Service) prepareReceivedFile(frame *Frame) (persistedDelivery, error) { dir, err := s.receivedDir() if err != nil { slog.Warn("save received file: cannot determine dir", "err", err) @@ -815,25 +483,19 @@ func (s *Service) prepareReceivedFile(frame *Frame, disclosure *decision.Disclos base := safeName[:len(safeName)-len(ext)] destName := fmt.Sprintf("%s-%s-%06d%s", base, ts, seq, ext) destPath := filepath.Join(dir, destName) - retention, err := s.prepareGovernedRetention(disclosure, destPath) - if err != nil { - return persistedDelivery{}, err - } if err := os.WriteFile(destPath, frame.Payload, 0600); err != nil { _ = os.Remove(destPath) - _ = retention.rollback() slog.Warn("save received file: write failed", "path", destPath, "err", err) return persistedDelivery{}, fmt.Errorf("write: %w", err) } return persistedDelivery{ rollback: func() error { removeErr := os.Remove(destPath) - retentionErr := retention.rollback() if removeErr != nil && !os.IsNotExist(removeErr) { return removeErr } - return retentionErr + return nil }, commit: func() { slog.Info("file saved", append([]any{"path", destPath, "bytes", len(frame.Payload)}, correlationAttrs(frame)...)...) @@ -849,7 +511,7 @@ func (s *Service) prepareReceivedFile(frame *Frame, disclosure *decision.Disclos } func (s *Service) saveInboxMessage(frame *Frame, from protocol.Addr) error { - delivery, err := s.prepareInboxMessage(frame, from, nil) + delivery, err := s.prepareInboxMessage(frame, from) if err != nil { return err } @@ -857,7 +519,7 @@ func (s *Service) saveInboxMessage(frame *Frame, from protocol.Addr) error { return nil } -func (s *Service) prepareInboxMessage(frame *Frame, from protocol.Addr, disclosure *decision.DisclosureBinding) (persistedDelivery, error) { +func (s *Service) prepareInboxMessage(frame *Frame, from protocol.Addr) (persistedDelivery, error) { dir, err := s.inboxDir() if err != nil { return persistedDelivery{}, err @@ -930,12 +592,7 @@ func (s *Service) prepareInboxMessage(frame *Frame, from protocol.Addr, disclosu } }() - retention, err := s.prepareGovernedRetention(disclosure, destPath) - if err != nil { - return persistedDelivery{}, err - } if err := os.WriteFile(destPath, data, 0600); err != nil { - _ = retention.rollback() return persistedDelivery{}, fmt.Errorf("write: %w", err) } written = true @@ -948,11 +605,10 @@ func (s *Service) prepareInboxMessage(frame *Frame, from protocol.Addr, disclosu if removeErr == nil && maxBytes > 0 { s.inboxFileRemoved(size) } - retentionErr := retention.rollback() if removeErr != nil && !os.IsNotExist(removeErr) { return removeErr } - return retentionErr + return nil }, commit: func() { slog.Info("inbox message saved", append([]any{"path", destPath, "type", TypeName(frame.Type), "bytes", len(frame.Payload)}, correlationAttrs(frame)...)...) diff --git a/service_disabled.go b/service_disabled.go index 19b37d1..9d0df13 100644 --- a/service_disabled.go +++ b/service_disabled.go @@ -15,7 +15,6 @@ import ( "time" "github.com/pilot-protocol/common/coreapi" - "github.com/pilot-protocol/common/decision" ) // ServiceConfig mirrors the real ServiceConfig so cmd/daemon's @@ -23,26 +22,15 @@ import ( // compiles unchanged when the plugin is disabled. Field set kept in sync // with the real ServiceConfig in service.go. type ServiceConfig struct { - ReceivedDir string - InboxDir string - IncludeBase64 bool - InboxMaxFiles int - InboxMaxBytes int64 - DedupeWindow time.Duration - DedupeContentWindow time.Duration - ReceivedMaxBytes int64 - IdleTimeout time.Duration - RequireGoverned bool - GovernedVerifier GovernedFrameVerifier - GovernedStreamVerifier GovernedStreamVerifier - RequireGovernedReceipts bool - GovernedReceiptRecorder GovernedReceiptRecorder - GovernedContentInspector decision.DisclosureContentInspector - RequireGovernedContentInspection bool - GovernedTransferQuota *decision.TransferQuotaLimiter - GovernedRetentionPolicies []GovernedRetentionPolicy - RetentionStateDir string - RetentionSweepInterval time.Duration + ReceivedDir string + InboxDir string + IncludeBase64 bool + InboxMaxFiles int + InboxMaxBytes int64 + DedupeWindow time.Duration + DedupeContentWindow time.Duration + ReceivedMaxBytes int64 + IdleTimeout time.Duration } // Service is a no-op replacement for the real plugin Service. diff --git a/zz_correlation_test.go b/zz_correlation_test.go index 5a363e3..578cadc 100644 --- a/zz_correlation_test.go +++ b/zz_correlation_test.go @@ -22,7 +22,6 @@ import ( "time" "github.com/pilot-protocol/common/coreapi" - "github.com/pilot-protocol/common/decision" "github.com/pilot-protocol/common/protocol" ) @@ -375,37 +374,6 @@ func TestService_FailedDeliveryIsNotRemembered(t *testing.T) { } } -func TestService_TaggedGovernedFrameKeepsCorrelation(t *testing.T) { - t.Parallel() - tmp := t.TempDir() - governed, verifier := newGovernedTestFrame(t, &Frame{Type: TypeText, Payload: []byte("approved")}, decision.Allow, nil) - svc := NewService(ServiceConfig{InboxDir: tmp, RequireGoverned: true, GovernedVerifier: verifier}) - conn := openServiceConn(t, svc, peerA) - - envelope, err := EncodeGovernedFrame(governed) - if err != nil { - t.Fatal(err) - } - envelope.MessageID, envelope.ReplyTo = "gov-1", "req-7" - res := mustSend(t, conn, envelope) - if !res.Tagged || string(res.Ack.Payload) != "ACK TEXT 8 bytes" { - t.Fatalf("result=%+v ack=%q", res, res.Ack.Payload) - } - recs := inboxRecords(t, tmp) - if len(recs) != 1 || recs[0]["message_id"] != "gov-1" || recs[0]["reply_to"] != "req-7" || recs[0]["data"] != "approved" { - t.Fatalf("records = %v", recs) - } - // An identical re-delivery is acknowledged as a duplicate instead of - // tripping the governed replay guard with an error. - res = mustSend(t, conn, envelope) - if !res.Duplicate || string(res.Ack.Payload) != "ACK TEXT 8 bytes (duplicate)" { - t.Fatalf("re-delivery result=%+v ack=%q", res, res.Ack.Payload) - } - if n := len(inboxRecords(t, tmp)); n != 1 { - t.Fatalf("inbox records = %d, want 1", n) - } -} - func TestService_TaggedTraceFrameKeepsCorrelation(t *testing.T) { t.Parallel() tmp := t.TempDir() @@ -464,8 +432,9 @@ const ( // genRejectUnknown is v0.2.2 and later, before TypeTagged: an unknown // type is answered "ERR UNKNOWN() save failed: unsupported frame type ". genRejectUnknown - // genRejectUnknownGoverned is genRejectUnknown with RequireGoverned, - // which rejects every non-governed type before looking at it. + // genRejectUnknownGoverned is genRejectUnknown with RequireGoverned (a + // receiver gate those releases had, removed since), which rejects every + // non-governed type before looking at it. genRejectUnknownGoverned ) @@ -624,18 +593,17 @@ func TestSend_CurrentReceiverKeepsTagsForEveryType(t *testing.T) { } // TestSend_NewReceiverRejectionIsNotRetried: a current receiver that rejects -// a tagged frame (here: unsigned under RequireGoverned) must not trigger the -// old-receiver fallback. +// a tagged frame (here: the message is larger than the inbox byte cap) must +// not trigger the old-receiver fallback. func TestSend_NewReceiverRejectionIsNotRetried(t *testing.T) { t.Parallel() tmp := t.TempDir() - _, verifier := newGovernedTestFrame(t, &Frame{Type: TypeText, Payload: []byte("x")}, decision.Allow, nil) - conn := openServiceConn(t, NewService(ServiceConfig{InboxDir: tmp, RequireGoverned: true, GovernedVerifier: verifier}), peerA) - res, err := sendAndAwaitAck(conn, &Frame{Type: TypeText, Payload: []byte("unsigned"), MessageID: "m-1"}) + conn := openServiceConn(t, NewService(ServiceConfig{InboxDir: tmp, InboxMaxBytes: 1}), peerA) + res, err := sendAndAwaitAck(conn, &Frame{Type: TypeText, Payload: []byte("too large"), MessageID: "m-1"}) if !errors.Is(err, ErrRejected) || res == nil || !res.Tagged { t.Fatalf("res=%+v err=%v", res, err) } - if !strings.HasPrefix(string(res.Ack.Payload), "ERR TEXT save failed: unsigned legacy frame rejected") { + if !strings.HasPrefix(string(res.Ack.Payload), "ERR TEXT save failed: inbox byte budget exceeded") { t.Fatalf("ack = %q", res.Ack.Payload) } if n := len(inboxRecords(t, tmp)); n != 0 { diff --git a/zz_coverage_test.go b/zz_coverage_test.go index affa4f8..08ade71 100644 --- a/zz_coverage_test.go +++ b/zz_coverage_test.go @@ -418,8 +418,9 @@ func TestHandleConn_SaveError(t *testing.T) { } // TestHandleConn_UnknownType ensures an unsupported wire type is rejected. -// Silently ACKing it would allow a sender to mistake an unverified governed -// envelope (or a future control frame) for a successfully delivered message. +// Silently ACKing it would allow a sender to mistake a frame this receiver +// does not implement (a reserved governed envelope, or a future control +// frame) for a successfully delivered message. func TestHandleConn_UnknownType(t *testing.T) { t.Parallel() tmp := t.TempDir() diff --git a/zz_governed_replay_test.go b/zz_governed_replay_test.go deleted file mode 100644 index 7b4890c..0000000 --- a/zz_governed_replay_test.go +++ /dev/null @@ -1,39 +0,0 @@ -// SPDX-License-Identifier: AGPL-3.0-or-later - -package dataexchange - -import ( - "testing" - "time" - - "github.com/pilot-protocol/common/decision" -) - -// TestGovernedReplayGuard pins SECURITY_REVIEW_v1.14 finding M5: a verified -// governed intent may be delivered at most once; a replay within the intent's -// validity window is rejected, and the cache drains after expiry. -func TestGovernedReplayGuard(t *testing.T) { - base := time.Unix(1_700_000_000, 0) - clock := base - g := newGovernedReplayGuard() - g.now = func() time.Time { return clock } - - intent := decision.Intent{ID: "int-1", TenantID: "t", AgentID: "a", ExpiresAt: base.Add(5 * time.Minute).Unix()} - - if err := g.admit(intent); err != nil { - t.Fatalf("first delivery rejected: %v", err) - } - if err := g.admit(intent); err == nil { - t.Fatal("replay within TTL was ACCEPTED") - } - // A distinct intent from the same agent is fine (fresh nonce/id). - if err := g.admit(decision.Intent{ID: "int-2", TenantID: "t", AgentID: "a", ExpiresAt: clock.Add(5 * time.Minute).Unix()}); err != nil { - t.Fatalf("distinct intent rejected: %v", err) - } - // After the intent expires the entry is pruned; the same id no longer - // collides (and upstream freshness would reject the stale intent anyway). - clock = base.Add(6 * time.Minute) - if err := g.admit(decision.Intent{ID: "int-1", TenantID: "t", AgentID: "a", ExpiresAt: clock.Add(5 * time.Minute).Unix()}); err != nil { - t.Fatalf("post-expiry re-use rejected: %v", err) - } -} diff --git a/zz_governed_reserved_test.go b/zz_governed_reserved_test.go new file mode 100644 index 0000000..ec51571 --- /dev/null +++ b/zz_governed_reserved_test.go @@ -0,0 +1,107 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +//go:build !no_dataexchange +// +build !no_dataexchange + +package dataexchange + +import ( + "errors" + "os" + "path/filepath" + "testing" + + "github.com/pilot-protocol/common/coreapi" +) + +// TestService_ReservedGovernedTypesAreRefused: frame types 8 and 9 are no +// longer implemented but stay reserved, because senders built before the +// removal can still emit them. The receiver must answer each with the same +// error reply as any unsupported type, store and publish nothing, not +// remember the frame as a delivery, and keep serving the connection. +func TestService_ReservedGovernedTypesAreRefused(t *testing.T) { + t.Parallel() + if TypeGoverned != 8 || TypeGovernedFileStream != 9 { + t.Fatalf("reserved types renumbered: TypeGoverned=%d TypeGovernedFileStream=%d", TypeGoverned, TypeGovernedFileStream) + } + // The shape of an envelope as an older sender encodes it. + envelope := []byte(`{"version":1,"type":1,"payload":"YXBwcm92ZWQ=","intent":{},"decision":{}}`) + for _, tc := range []struct { + name string + frame *Frame + wantAck string + }{ + {"governed", &Frame{Type: TypeGoverned, Payload: envelope}, "ERR GOVERNED save failed: unsupported frame type 8"}, + {"governed file", &Frame{Type: TypeGoverned, Filename: "report.txt", Payload: envelope}, "ERR GOVERNED save failed: unsupported frame type 8"}, + {"governed tagged", &Frame{Type: TypeGoverned, Payload: envelope, MessageID: "gov-1", ReplyTo: "req-7"}, "ERR GOVERNED save failed: unsupported frame type 8"}, + {"governed garbage", &Frame{Type: TypeGoverned, Payload: []byte{0xff, 0x00, 0x7b}}, "ERR GOVERNED save failed: unsupported frame type 8"}, + {"governed filestream", &Frame{Type: TypeGovernedFileStream, Payload: envelope}, "ERR GOVERNED_FILESTREAM save failed: unsupported frame type 9"}, + {"governed filestream tagged", &Frame{Type: TypeGovernedFileStream, Payload: envelope, MessageID: "gov-2"}, "ERR GOVERNED_FILESTREAM save failed: unsupported frame type 9"}, + {"governed filestream garbage", &Frame{Type: TypeGovernedFileStream, Payload: []byte{0xff, 0x00, 0x7b}}, "ERR GOVERNED_FILESTREAM save failed: unsupported frame type 9"}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + tmp := t.TempDir() + inbox, received := filepath.Join(tmp, "inbox"), filepath.Join(tmp, "received") + events := newCapturingEvents() + svc := NewService(ServiceConfig{InboxDir: inbox, ReceivedDir: received, DedupeContentWindow: DefaultDedupeWindow}) + svc.deps = coreapi.Deps{Events: events} + conn := openServiceConn(t, svc, peerA) + + // Twice: the second attempt must be refused again rather than + // acknowledged as a duplicate of a stored delivery. + for attempt := 1; attempt <= 2; attempt++ { + res, err := sendAndAwaitAck(conn, tc.frame) + if !errors.Is(err, ErrRejected) || res == nil { + t.Fatalf("attempt %d: res=%+v err=%v, want ErrRejected", attempt, res, err) + } + if res.Ack.Type != TypeText || string(res.Ack.Payload) != tc.wantAck { + t.Fatalf("attempt %d: ack = %d %q, want %q", attempt, res.Ack.Type, res.Ack.Payload, tc.wantAck) + } + if tagged := tc.frame.MessageID != ""; res.Tagged != tagged { + t.Fatalf("attempt %d: Tagged = %v, want %v (a refusal must not trigger the untagged re-send)", attempt, res.Tagged, tagged) + } + if res.Duplicate { + t.Fatalf("attempt %d: refused frame reported as a duplicate", attempt) + } + } + for _, dir := range []string{inbox, received} { + if entries, err := os.ReadDir(dir); err == nil && len(entries) != 0 { + t.Fatalf("%s holds %d entries after a refused frame, want none", dir, len(entries)) + } else if err != nil && !os.IsNotExist(err) { + t.Fatal(err) + } + } + events.mu.Lock() + published := len(events.published) + events.mu.Unlock() + if published != 0 { + t.Fatalf("a refused frame published %d events, want 0", published) + } + + // The connection is still in service for ordinary frames. + res := mustSend(t, conn, &Frame{Type: TypeText, Payload: []byte("after")}) + if string(res.Ack.Payload) != "ACK TEXT 5 bytes" { + t.Fatalf("ack after refusal = %q", res.Ack.Payload) + } + if recs := inboxRecords(t, inbox); len(recs) != 1 || recs[0]["data"] != "after" { + t.Fatalf("inbox after refusal = %v", recs) + } + }) + } +} + +// TestReservedGovernedTypesKeepTheirNames pins the wire numbers and names so +// they are not handed to a new frame type, and checks they are not treated +// as storable deliveries by the duplicate-suppression layer. +func TestReservedGovernedTypesKeepTheirNames(t *testing.T) { + t.Parallel() + for typ, name := range map[uint32]string{8: "GOVERNED", 9: "GOVERNED_FILESTREAM"} { + if got := TypeName(typ); got != name { + t.Errorf("TypeName(%d) = %q, want %q", typ, got, name) + } + if dedupeEligible(typ) { + t.Errorf("reserved type %d is dedupe-eligible", typ) + } + } +} diff --git a/zz_inbox_budget_test.go b/zz_inbox_budget_test.go index 8543b41..fd5f3f1 100644 --- a/zz_inbox_budget_test.go +++ b/zz_inbox_budget_test.go @@ -280,13 +280,13 @@ func TestInboxByteCap_ConcurrentWritersHoldCap(t *testing.T) { } } -// TestInboxByteCap_RollbackReleasesBytes: a governed delivery that is -// rolled back (receipt failure) must hand its bytes back to the budget. +// TestInboxByteCap_RollbackReleasesBytes: a staged delivery that is rolled +// back must hand its bytes back to the budget. func TestInboxByteCap_RollbackReleasesBytes(t *testing.T) { t.Parallel() tmp := t.TempDir() s := NewService(ServiceConfig{InboxDir: tmp, InboxMaxBytes: 1 << 20}) - delivery, err := s.prepareInboxMessage(&Frame{Type: TypeText, Payload: []byte("to be rolled back")}, protocol.Addr{Node: 3}, nil) + delivery, err := s.prepareInboxMessage(&Frame{Type: TypeText, Payload: []byte("to be rolled back")}, protocol.Addr{Node: 3}) if err != nil { t.Fatalf("prepare: %v", err) }