diff --git a/cmd/flex-auth/main.go b/cmd/flex-auth/main.go index 24ba555..49d078a 100644 --- a/cmd/flex-auth/main.go +++ b/cmd/flex-auth/main.go @@ -18,6 +18,7 @@ import ( "github.com/netkingdom/flex-auth/internal/audit" "github.com/netkingdom/flex-auth/internal/callerauth" decisioncore "github.com/netkingdom/flex-auth/internal/decision" + "github.com/netkingdom/flex-auth/internal/emission" "github.com/netkingdom/flex-auth/internal/policy" "github.com/netkingdom/flex-auth/internal/registry" "github.com/netkingdom/flex-auth/pkg/api" @@ -326,6 +327,8 @@ func runServe(args []string, stdout, stderr io.Writer) int { registryPath := fs.String("registry", "", "registry snapshot JSON file") policyPath := fs.String("policy", "", "policy package Markdown file") logPath := fs.String("log", "", "optional JSONL decision log path") + outboxDir := fs.String("outbox-dir", "", "durable emission outbox directory (FLEX-WP-0031); exclusive with --log") + emissionSource := fs.String("emission-source", "", "audit-core source name for emitted decision events, e.g. flex-auth.tenant-engine") callerAuthMode := fs.String("caller-auth-mode", "disabled", "disabled, warn, or enforce") callerAudience := fs.String("caller-audience", "flex-auth", "required caller token audience") callerKubernetesURL := fs.String("caller-kubernetes-url", "https://kubernetes.default.svc", "Kubernetes API base URL for TokenReview") @@ -341,10 +344,24 @@ func runServe(args []string, stdout, stderr io.Writer) int { return 64 } + if *outboxDir != "" && (*logPath != "" || *emissionSource == "") { + fmt.Fprintln(stderr, "serve --outbox-dir requires --emission-source and excludes --log") + return 64 + } + engine, err := buildEngine(context.Background(), *registryPath, *policyPath, *logPath) if err != nil { return fail(stderr, err) } + var outbox *emission.Outbox + if *outboxDir != "" { + outbox, err = emission.Open(*outboxDir, *emissionSource) + if err != nil { + return fail(stderr, err) + } + defer outbox.Close() + engine.SetDecisionLog(outbox) + } authenticator, err := buildCallerAuthenticator( callerauth.Mode(*callerAuthMode), @@ -360,6 +377,9 @@ func runServe(args []string, stdout, stderr io.Writer) int { } mux := newServeMuxWithCallerAuth(engine, authenticator) + if outbox != nil { + mountEmission(mux, outbox) + } fmt.Fprintf(stderr, "flex-auth serving on http://%s\n", *addr) if err := http.ListenAndServe(*addr, mux); err != nil { @@ -520,8 +540,25 @@ func writeStatus(w io.Writer, status string, extra map[string]any) int { return 0 } +// mountEmission exposes the local half of reconciliation: committed and +// released-uncommitted counts per class. Counts only, never records. +func mountEmission(mux *http.ServeMux, outbox *emission.Outbox) { + mux.HandleFunc("/v1/emission", func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + writeHTTP(w, outbox.Status(), nil) + }) +} + func writeHTTP(w http.ResponseWriter, value any, err error) { w.Header().Set("content-type", "application/json") + if errors.Is(err, decisioncore.ErrRecordNotCommitted) { + // The consumer's declared stance applies (FLEX-DEC-2026-018). + http.Error(w, err.Error(), http.StatusServiceUnavailable) + return + } if err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return diff --git a/decisions/decisions.md b/decisions/decisions.md index e222bef..34ae42a 100644 --- a/decisions/decisions.md +++ b/decisions/decisions.md @@ -2070,5 +2070,13 @@ not one per resource. again, so a restart during an outbox outage loses them. This is the same residual as the stated bound, not a new one. +**Batch residual (added 2026-09-23, `FLEX-WP-0031-T03`).** A batch has one +HTTP status. If a batch holds a permissive decision whose record did not +commit, the whole batch is withheld (503), and its restrictions are withheld +with it. This is the one place the rule above yields. It is accepted because no +live consumer calls `/v1/batch_check`: user-engine's `batch_check` loops over +`/v1/check`. It must be revisited before any consumer adopts the batch +endpoint. + **Bound.** Unchanged from `cadence.yaml`: none of this detects a compromised flex-auth suppressing a record together with its own count. diff --git a/internal/decision/emission_atomicity_test.go b/internal/decision/emission_atomicity_test.go new file mode 100644 index 0000000..e5080a1 --- /dev/null +++ b/internal/decision/emission_atomicity_test.go @@ -0,0 +1,105 @@ +package decision_test + +import ( + "context" + "errors" + "testing" + + "github.com/netkingdom/flex-auth/internal/decision" + "github.com/netkingdom/flex-auth/pkg/api" +) + +// failingRecorder is an outbox whose commit always fails. +type failingRecorder struct { + appends int + batches int + uncommitted []api.DecisionEffect +} + +func (r *failingRecorder) Append(api.DecisionEnvelope) error { + r.appends++ + return errors.New("disk full") +} + +func (r *failingRecorder) AppendBatch([]api.DecisionEnvelope) error { + r.batches++ + return errors.New("disk full") +} + +func (r *failingRecorder) NoteReleasedUncommitted(effect api.DecisionEffect) { + r.uncommitted = append(r.uncommitted, effect) +} + +var ( + allowedRead = api.CheckRequest{ + Subject: api.SubjectRef{ID: "user:alice"}, + Action: "read", + Resource: api.ResourceRef{ID: "document:internal-note", System: "markitect-tool"}, + } + deniedRead = api.CheckRequest{ + Subject: api.SubjectRef{ID: "user:alice"}, + Action: "read", + Resource: api.ResourceRef{ID: "document:missing", Type: "document", System: "markitect-tool"}, + } +) + +// FLEX-DEC-2026-018: withholding a deny turns it into an error an open-stance +// consumer reads as proceed, so a restriction is released and counted. +func TestUncommittedDenyIsReleasedAndCounted(t *testing.T) { + engine := newTestEngine(t) + recorder := &failingRecorder{} + engine.SetDecisionLog(recorder) + + got, err := engine.Check(context.Background(), deniedRead) + if err != nil { + t.Fatalf("Check: %v; a restriction must never be withheld", err) + } + if got.Effect != api.DecisionEffectDeny { + t.Fatalf("effect = %q; want deny", got.Effect) + } + if len(recorder.uncommitted) != 1 || recorder.uncommitted[0] != api.DecisionEffectDeny { + t.Fatalf("released_uncommitted = %v; want [deny]", recorder.uncommitted) + } +} + +// FLEX-DEC-2026-018: authority is never granted without its evidence. +func TestUncommittedAllowIsWithheld(t *testing.T) { + engine := newTestEngine(t) + recorder := &failingRecorder{} + engine.SetDecisionLog(recorder) + + got, err := engine.Check(context.Background(), allowedRead) + if !errors.Is(err, decision.ErrRecordNotCommitted) { + t.Fatalf("err = %v, decision = %+v; want ErrRecordNotCommitted", err, got) + } + if len(recorder.uncommitted) != 0 { + t.Fatalf("a withheld allow was counted as released: %v", recorder.uncommitted) + } +} + +func TestBatchCommitsOnceAndWithholdsOnAnyUncommittedAllow(t *testing.T) { + engine := newTestEngine(t) + recorder := &failingRecorder{} + engine.SetDecisionLog(recorder) + + _, err := engine.BatchCheck(context.Background(), api.BatchCheckRequest{ + Subject: allowedRead.Subject, + Action: "read", + Resources: []api.ResourceRef{allowedRead.Resource, deniedRead.Resource}, + }) + if !errors.Is(err, decision.ErrRecordNotCommitted) { + t.Fatalf("err = %v; want ErrRecordNotCommitted for a batch holding an allow", err) + } + if recorder.batches != 1 || recorder.appends != 0 { + t.Fatalf("batches=%d appends=%d; want one batch commit", recorder.batches, recorder.appends) + } + + decisions, err := engine.BatchCheck(context.Background(), api.BatchCheckRequest{ + Subject: deniedRead.Subject, + Action: "read", + Resources: []api.ResourceRef{deniedRead.Resource}, + }) + if err != nil || len(decisions) != 1 { + t.Fatalf("restriction-only batch: decisions=%v err=%v; want it released", decisions, err) + } +} diff --git a/internal/decision/engine.go b/internal/decision/engine.go index e826030..9c75411 100644 --- a/internal/decision/engine.go +++ b/internal/decision/engine.go @@ -6,6 +6,7 @@ import ( "crypto/sha256" "encoding/hex" "encoding/json" + "errors" "fmt" "reflect" "slices" @@ -36,6 +37,22 @@ type DecisionRecorder interface { Append(api.DecisionEnvelope) error } +// BatchRecorder commits several decisions in one durable write. +type BatchRecorder interface { + AppendBatch([]api.DecisionEnvelope) error +} + +// UncommittedNoter counts restrictions released after their record failed to +// commit (FLEX-DEC-2026-018). +type UncommittedNoter interface { + NoteReleasedUncommitted(api.DecisionEffect) +} + +// ErrRecordNotCommitted means a permissive decision was withheld because its +// record did not commit. The service answers 503, and the consumer's declared +// stance applies (FLEX-DEC-2026-018). +var ErrRecordNotCommitted = errors.New("decision withheld: record not committed") + // ListAllowedRequest describes a deterministic list_allowed call. type ListAllowedRequest struct { Subject api.SubjectRef `json:"subject"` @@ -104,8 +121,8 @@ func (e *Engine) now() time.Time { return time.Now().UTC() } -// Check evaluates one subject/action/resource request. -func (e *Engine) Check(ctx context.Context, request api.CheckRequest) (api.DecisionEnvelope, error) { +// decide evaluates one request without recording it. +func (e *Engine) decide(ctx context.Context, request api.CheckRequest) (api.DecisionEnvelope, error) { normalized, facts := e.normalizeRequest(request) expectation, err := e.policy.Evaluate(ctx, normalized) @@ -113,19 +130,29 @@ func (e *Engine) Check(ctx context.Context, request api.CheckRequest) (api.Decis return api.DecisionEnvelope{}, err } - decision := e.envelope(ctx, normalized, request, expectation, facts) - if err := e.recordDecision(decision); err != nil { + return e.envelope(ctx, normalized, request, expectation, facts), nil +} + +// Check evaluates one subject/action/resource request. The decision is +// released only after its record commits, except that a restriction is never +// withheld (FLEX-DEC-2026-018). +func (e *Engine) Check(ctx context.Context, request api.CheckRequest) (api.DecisionEnvelope, error) { + decision, err := e.decide(ctx, request) + if err != nil { + return api.DecisionEnvelope{}, err + } + if err := e.recordDecisions([]api.DecisionEnvelope{decision}); err != nil { return api.DecisionEnvelope{}, err } return decision, nil } // BatchCheck evaluates one subject/action/context tuple against resources in -// request order. +// request order, and commits the whole batch in one durable write. func (e *Engine) BatchCheck(ctx context.Context, request api.BatchCheckRequest) ([]api.DecisionEnvelope, error) { decisions := make([]api.DecisionEnvelope, 0, len(request.Resources)) for _, resource := range request.Resources { - decision, err := e.Check(ctx, api.CheckRequest{ + decision, err := e.decide(ctx, api.CheckRequest{ ID: request.ID, Tenant: request.Tenant, Subject: request.Subject, @@ -139,18 +166,21 @@ func (e *Engine) BatchCheck(ctx context.Context, request api.BatchCheckRequest) } decisions = append(decisions, decision) } + if err := e.recordDecisions(decisions); err != nil { + return nil, err + } return decisions, nil } // ListAllowed evaluates candidate resources and returns only allow decisions. func (e *Engine) ListAllowed(ctx context.Context, request ListAllowedRequest) ([]api.DecisionEnvelope, error) { candidates := e.store.ResourceRefs(request.System, request.ResourceType) - allowed := make([]api.DecisionEnvelope, 0, len(candidates)) + decisions := make([]api.DecisionEnvelope, 0, len(candidates)) for _, resource := range candidates { if !resourceMatchesFilters(resource, request.Filters) { continue } - decision, err := e.Check(ctx, api.CheckRequest{ + decision, err := e.decide(ctx, api.CheckRequest{ Subject: request.Subject, Action: request.Action, Resource: resource, @@ -160,6 +190,13 @@ func (e *Engine) ListAllowed(ctx context.Context, request ListAllowedRequest) ([ if err != nil { return nil, err } + decisions = append(decisions, decision) + } + if err := e.recordDecisions(decisions); err != nil { + return nil, err + } + allowed := make([]api.DecisionEnvelope, 0, len(decisions)) + for _, decision := range decisions { if decision.Effect == api.DecisionEffectAllow { allowed = append(allowed, decision) } @@ -404,12 +441,46 @@ func callerProvenance(ctx context.Context) *api.CallerProvenance { return out } -func (e *Engine) recordDecision(decision api.DecisionEnvelope) error { +// recordDecisions commits decisions and applies FLEX-DEC-2026-018 when the +// commit fails: restrictions are released and counted, and any permissive +// decision withholds the call. In a batch that withholds the batch's +// restrictions too; no live consumer calls batch_check (user-engine loops +// /v1/check), and the residual is stated in the ruling. +func (e *Engine) recordDecisions(decisions []api.DecisionEnvelope) error { e.mu.Lock() - defer e.mu.Unlock() - e.history[decision.ID] = decision - if e.log != nil { - return e.log.Append(decision) + for _, decision := range decisions { + e.history[decision.ID] = decision + } + log := e.log + e.mu.Unlock() + if log == nil || len(decisions) == 0 { + return nil + } + err := appendAll(log, decisions) + if err == nil { + return nil + } + for _, decision := range decisions { + if !decision.Effect.Restricts() { + return fmt.Errorf("%w: %v", ErrRecordNotCommitted, err) + } + } + if noter, ok := log.(UncommittedNoter); ok { + for _, decision := range decisions { + noter.NoteReleasedUncommitted(decision.Effect) + } + } + return nil +} + +func appendAll(log DecisionRecorder, decisions []api.DecisionEnvelope) error { + if batch, ok := log.(BatchRecorder); ok { + return batch.AppendBatch(decisions) + } + for _, decision := range decisions { + if err := log.Append(decision); err != nil { + return err + } } return nil } diff --git a/internal/emission/outbox.go b/internal/emission/outbox.go new file mode 100644 index 0000000..e7f67e3 --- /dev/null +++ b/internal/emission/outbox.go @@ -0,0 +1,268 @@ +// Package emission delivers the decision-record emission guarantee declared in +// cadence.yaml (FLEX-WP-0031). A decision is released only after its event is +// durably committed to a local outbox (FLEX-DEC-2026-018); draining to +// audit-core is asynchronous. +package emission + +import ( + "bytes" + "crypto/rand" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "sync" + "time" + + "github.com/netkingdom/flex-auth/pkg/api" +) + +// ClassPrefix prefixes every decision event class; the suffix is the effect. +const ClassPrefix = "flex-auth.decision." + +// PlatformTenant is sent when a decision names no tenant, because audit-core +// requires one on every event. +const PlatformTenant = "tenant:platform" + +const outboxFile = "outbox.jsonl" + +// Class returns the event class for a decision effect. +func Class(effect api.DecisionEffect) string { + return ClassPrefix + string(effect) +} + +// Event is one audit-core.event envelope: exactly the fields audit-core's +// normalize() requires. Data carries the decision record itself. +type Event struct { + ID string `json:"id"` + Type string `json:"type"` + Source string `json:"source"` + Subject string `json:"subject"` + Tenant string `json:"tenant"` + CorrelationID string `json:"correlation_id"` + OccurredAt string `json:"occurred_at"` + Data any `json:"data"` +} + +// entry is one committed outbox line. +type entry struct { + Seq uint64 `json:"seq"` + CommittedAt string `json:"committed_at"` + Event Event `json:"event"` +} + +// Status is the local half of reconciliation: what this outbox committed, per +// class, and what it released without committing. +type Status struct { + Source string `json:"source"` + Committed map[string]uint64 `json:"committed"` + ReleasedUncommitted map[string]uint64 `json:"released_uncommitted"` + // ReleasedUncommitted is held in memory: a restart during an outbox outage + // loses it (FLEX-DEC-2026-018, consequences). + Means string `json:"means"` +} + +// Outbox is a durable, append-only, fsync-per-commit event log. +type Outbox struct { + source string + path string + clock func() time.Time + + mu sync.Mutex + file *os.File + seq uint64 + size int64 // bytes committed; a failed commit is cut back to it + committed map[string]uint64 + uncommitted map[string]uint64 +} + +// Open opens or creates the outbox in dir and rebuilds committed counts from it. +func Open(dir, source string) (*Outbox, error) { + if source == "" { + return nil, errors.New("emission source is required") + } + if err := os.MkdirAll(dir, 0o750); err != nil { + return nil, fmt.Errorf("create outbox directory: %w", err) + } + o := &Outbox{ + source: source, + path: filepath.Join(dir, outboxFile), + clock: time.Now, + committed: map[string]uint64{}, + uncommitted: map[string]uint64{}, + } + if err := o.load(); err != nil { + return nil, err + } + file, err := os.OpenFile(o.path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o640) + if err != nil { + return nil, fmt.Errorf("open outbox: %w", err) + } + o.file = file + return o, nil +} + +func (o *Outbox) load() error { + data, err := os.ReadFile(o.path) + if errors.Is(err, os.ErrNotExist) { + return nil + } + if err != nil { + return fmt.Errorf("read outbox: %w", err) + } + good := 0 + for line := 1; good < len(data); line++ { + end := bytes.IndexByte(data[good:], '\n') + if end < 0 { + // A torn final line is a write whose sync never returned, so its + // decision was never released as committed. Drop it so the next + // append does not bury it mid-file. + o.size = int64(good) + return os.Truncate(o.path, o.size) + } + var e entry + if err := json.Unmarshal(data[good:good+end], &e); err != nil { + return fmt.Errorf("outbox line %d is corrupt: %w", line, err) + } + o.seq = e.Seq + o.committed[e.Event.Type]++ + good += end + 1 + } + o.size = int64(good) + return nil +} + +// SetClock overrides the commit clock. Tests use it. +func (o *Outbox) SetClock(clock func() time.Time) { + o.mu.Lock() + defer o.mu.Unlock() + o.clock = clock +} + +// Close closes the outbox file. +func (o *Outbox) Close() error { + o.mu.Lock() + defer o.mu.Unlock() + return o.file.Close() +} + +// Append commits one decision; it satisfies decision.DecisionRecorder. +func (o *Outbox) Append(decision api.DecisionEnvelope) error { + return o.AppendBatch([]api.DecisionEnvelope{decision}) +} + +// AppendBatch commits decisions with one sync, so a batch check pays the +// durable-write cost once rather than per resource. +func (o *Outbox) AppendBatch(decisions []api.DecisionEnvelope) error { + o.mu.Lock() + defer o.mu.Unlock() + now := o.clock().UTC().Format(time.RFC3339Nano) + var buf []byte + seq := o.seq + for _, decision := range decisions { + id, err := eventID() + if err != nil { + return err + } + seq++ + line, err := json.Marshal(entry{Seq: seq, CommittedAt: now, Event: o.event(id, now, decision)}) + if err != nil { + return fmt.Errorf("marshal outbox entry: %w", err) + } + buf = append(append(buf, line...), '\n') + } + if _, err := o.file.Write(buf); err != nil { + return o.rollback(fmt.Errorf("write outbox: %w", err)) + } + if err := o.file.Sync(); err != nil { + return o.rollback(fmt.Errorf("sync outbox: %w", err)) + } + o.seq = seq + o.size += int64(len(buf)) + for _, decision := range decisions { + o.committed[Class(decision.Effect)]++ + } + return nil +} + +// rollback cuts a failed commit back to the last committed size, so a partial +// line never sits in front of the next commit. The commit error is returned +// either way; a failed cut is reported alongside it. +func (o *Outbox) rollback(cause error) error { + if err := o.file.Truncate(o.size); err != nil { + return errors.Join(cause, fmt.Errorf("truncate outbox after failed commit: %w", err)) + } + return cause +} + +// NoteReleasedUncommitted counts a restriction released after its commit +// failed. Every non-zero count is a finding. +func (o *Outbox) NoteReleasedUncommitted(effect api.DecisionEffect) { + o.mu.Lock() + defer o.mu.Unlock() + o.uncommitted[Class(effect)]++ +} + +// Status reports committed and released-uncommitted counts per class. +func (o *Outbox) Status() Status { + o.mu.Lock() + defer o.mu.Unlock() + status := Status{ + Source: o.source, + Committed: map[string]uint64{}, + ReleasedUncommitted: map[string]uint64{}, + Means: "committed counts this outbox holds since it was created; " + + "released_uncommitted is decisions released without a record " + + "since this process started, and any non-zero value is a finding", + } + for _, effect := range api.DecisionEffects() { + class := Class(effect) + status.Committed[class] = o.committed[class] + status.ReleasedUncommitted[class] = o.uncommitted[class] + } + return status +} + +func (o *Outbox) event(id, at string, decision api.DecisionEnvelope) Event { + tenant := decision.Resource.Tenant + if decision.Binding != nil && decision.Binding.Tenant != "" { + tenant = decision.Binding.Tenant + } + if tenant == "" { + tenant = PlatformTenant + } + return Event{ + ID: id, + Type: Class(decision.Effect), + Source: o.source, + Subject: resourceSubject(decision.Resource), + Tenant: tenant, + CorrelationID: decision.ID, + OccurredAt: at, + Data: decision, + } +} + +func resourceSubject(resource api.ResourceRef) string { + subject := resource.ID + if resource.Type != "" { + subject = resource.Type + "/" + subject + } + if resource.System != "" { + subject = resource.System + "/" + subject + } + return subject +} + +// eventID is random, not derived from the decision: decision ids are content +// digests, so two identical checks share one, and audit-core's idempotency key +// would merge two real decisions into one. +func eventID() (string, error) { + var b [16]byte + if _, err := rand.Read(b[:]); err != nil { + return "", fmt.Errorf("event id: %w", err) + } + return "flex-auth-evt:" + hex.EncodeToString(b[:]), nil +} diff --git a/internal/emission/outbox_test.go b/internal/emission/outbox_test.go new file mode 100644 index 0000000..e2e7209 --- /dev/null +++ b/internal/emission/outbox_test.go @@ -0,0 +1,180 @@ +package emission_test + +import ( + "os" + "path/filepath" + "sort" + "testing" + + "gopkg.in/yaml.v3" + + "github.com/netkingdom/flex-auth/internal/emission" + "github.com/netkingdom/flex-auth/pkg/api" +) + +func deny(id, tenant string) api.DecisionEnvelope { + return api.DecisionEnvelope{ + ID: id, + Effect: api.DecisionEffectDeny, + Resource: api.ResourceRef{ID: "t1", Type: "tenant", System: "tenant-engine"}, + Binding: &api.DecisionBinding{Tenant: tenant}, + } +} + +func readEvents(t *testing.T, dir string) []map[string]any { + t.Helper() + data, err := os.ReadFile(filepath.Join(dir, "outbox.jsonl")) + if err != nil { + t.Fatal(err) + } + var events []map[string]any + for _, line := range splitLines(data) { + var e struct { + Event map[string]any `json:"event"` + } + if err := yaml.Unmarshal(line, &e); err != nil { + t.Fatal(err) + } + events = append(events, e.Event) + } + return events +} + +func splitLines(data []byte) [][]byte { + var lines [][]byte + start := 0 + for i, b := range data { + if b == '\n' { + lines = append(lines, data[start:i]) + start = i + 1 + } + } + return lines +} + +func TestCommittedEventsCarryAuditCoreRequiredFields(t *testing.T) { + dir := t.TempDir() + outbox, err := emission.Open(dir, "flex-auth.tenant-engine") + if err != nil { + t.Fatal(err) + } + // Identical checks share a content-digest decision id. + if err := outbox.AppendBatch([]api.DecisionEnvelope{deny("decision:aa", ""), deny("decision:aa", "tenant:acme")}); err != nil { + t.Fatal(err) + } + outbox.Close() + + events := readEvents(t, dir) + if len(events) != 2 { + t.Fatalf("events = %d; want 2", len(events)) + } + // audit-core normalize() rejects an event missing any of these. + for _, event := range events { + for _, key := range []string{"id", "type", "source", "subject", "tenant", "correlation_id", "occurred_at", "data"} { + if value, ok := event[key]; !ok || value == "" || value == nil { + t.Fatalf("event missing %q: %v", key, event) + } + } + if event["type"] != "flex-auth.decision.deny" || event["correlation_id"] != "decision:aa" { + t.Fatalf("event = %v", event) + } + } + if events[0]["id"] == events[1]["id"] { + t.Fatal("two decisions share an event id; audit-core idempotency would merge them") + } + if events[0]["tenant"] != emission.PlatformTenant || events[1]["tenant"] != "tenant:acme" { + t.Fatalf("tenants = %v, %v", events[0]["tenant"], events[1]["tenant"]) + } +} + +func TestCountsSurviveRestartAndTornTailIsDropped(t *testing.T) { + dir := t.TempDir() + outbox, err := emission.Open(dir, "flex-auth.test") + if err != nil { + t.Fatal(err) + } + if err := outbox.Append(deny("decision:1", "")); err != nil { + t.Fatal(err) + } + outbox.Close() + + // A crash mid-write leaves a line with no newline. + f, _ := os.OpenFile(filepath.Join(dir, "outbox.jsonl"), os.O_APPEND|os.O_WRONLY, 0) + f.WriteString(`{"seq":2,"event":{"type":"flex-auth.decision.de`) + f.Close() + + outbox, err = emission.Open(dir, "flex-auth.test") + if err != nil { + t.Fatalf("reopen with torn tail: %v", err) + } + if err := outbox.Append(deny("decision:3", "")); err != nil { + t.Fatal(err) + } + outbox.Close() + + outbox, err = emission.Open(dir, "flex-auth.test") + if err != nil { + t.Fatalf("reopen after append past torn tail: %v", err) + } + defer outbox.Close() + if got := outbox.Status().Committed["flex-auth.decision.deny"]; got != 2 { + t.Fatalf("committed deny = %d; want 2 (torn write never counted)", got) + } +} + +func TestReleasedUncommittedIsReportedPerClass(t *testing.T) { + outbox, err := emission.Open(t.TempDir(), "flex-auth.test") + if err != nil { + t.Fatal(err) + } + defer outbox.Close() + outbox.NoteReleasedUncommitted(api.DecisionEffectRedact) + status := outbox.Status() + if status.ReleasedUncommitted["flex-auth.decision.redact"] != 1 { + t.Fatalf("status = %+v", status) + } + if len(status.Committed) != len(api.DecisionEffects()) { + t.Fatalf("status must report every class, zero included: %+v", status.Committed) + } +} + +// cadence.yaml must classify exactly the effect vocabulary, so a new effect +// cannot ship unclassified (FLEX-WP-0031-T03 gate). +func TestCadenceClassifiesEveryEffect(t *testing.T) { + data, err := os.ReadFile(filepath.Join("..", "..", "cadence.yaml")) + if err != nil { + t.Fatal(err) + } + var cadence struct { + Classes map[string]struct { + Action string `yaml:"action"` + EvidenceClass string `yaml:"evidence_class"` + } `yaml:"classes"` + } + if err := yaml.Unmarshal(data, &cadence); err != nil { + t.Fatal(err) + } + var declared, want []string + for name, class := range cadence.Classes { + if class.EvidenceClass != "load-bearing" { + continue + } + declared = append(declared, name) + if class.Action != emission.Class(api.DecisionEffect(name)) { + t.Errorf("class %q action = %q; want %q", name, class.Action, emission.Class(api.DecisionEffect(name))) + } + } + for _, effect := range api.DecisionEffects() { + want = append(want, string(effect)) + } + sort.Strings(declared) + sort.Strings(want) + if len(declared) != len(want) { + t.Fatalf("cadence.yaml load-bearing classes %v; want the effect vocabulary %v", declared, want) + } + for i := range want { + if declared[i] != want[i] { + t.Fatalf("cadence.yaml load-bearing classes %v; want %v", declared, want) + } + } +} diff --git a/pkg/api/canonical.go b/pkg/api/canonical.go index bde56cb..fba023f 100644 --- a/pkg/api/canonical.go +++ b/pkg/api/canonical.go @@ -209,6 +209,30 @@ const ( DecisionEffectNotApplicable DecisionEffect = "not_applicable" ) +// DecisionEffects returns the complete effect vocabulary. Every effect is an +// emission class in cadence.yaml; a test fails if a constant above is missing +// here, so a new effect cannot ship unclassified (FLEX-WP-0031-T03). +func DecisionEffects() []DecisionEffect { + return []DecisionEffect{ + DecisionEffectAllow, + DecisionEffectDeny, + DecisionEffectRedact, + DecisionEffectAuditOnly, + DecisionEffectNotApplicable, + } +} + +// Restricts reports whether an effect withholds authority. A restriction is +// released even when its record fails to commit, because withholding it turns +// it into an error an open-stance consumer reads as proceed (FLEX-DEC-2026-018). +func (e DecisionEffect) Restricts() bool { + switch e { + case DecisionEffectDeny, DecisionEffectRedact, DecisionEffectNotApplicable: + return true + } + return false +} + // DecisionRecordContractV1 is the published decision-record contract identifier. const DecisionRecordContractV1 = "flex-auth.decision-record.v1" diff --git a/pkg/api/effects_test.go b/pkg/api/effects_test.go new file mode 100644 index 0000000..f624480 --- /dev/null +++ b/pkg/api/effects_test.go @@ -0,0 +1,67 @@ +package api + +import ( + "go/ast" + "go/parser" + "go/token" + "testing" +) + +// TestDecisionEffectsListsEveryConstant parses canonical.go so that adding a +// DecisionEffect constant without listing it in DecisionEffects fails here. +func TestDecisionEffectsListsEveryConstant(t *testing.T) { + file, err := parser.ParseFile(token.NewFileSet(), "canonical.go", nil, 0) + if err != nil { + t.Fatal(err) + } + declared := map[string]bool{} + ast.Inspect(file, func(node ast.Node) bool { + spec, ok := node.(*ast.ValueSpec) + if !ok { + return true + } + if ident, ok := spec.Type.(*ast.Ident); !ok || ident.Name != "DecisionEffect" { + return true + } + for _, value := range spec.Values { + if lit, ok := value.(*ast.BasicLit); ok { + declared[lit.Value[1:len(lit.Value)-1]] = true + } + } + return true + }) + listed := map[string]bool{} + for _, effect := range DecisionEffects() { + listed[string(effect)] = true + } + if len(declared) == 0 { + t.Fatal("found no DecisionEffect constants in canonical.go") + } + for effect := range declared { + if !listed[effect] { + t.Errorf("DecisionEffect %q is declared but missing from DecisionEffects()", effect) + } + } + if len(listed) != len(declared) { + t.Errorf("DecisionEffects() lists %d effects, canonical.go declares %d", len(listed), len(declared)) + } +} + +func TestRestrictsSplitsByDirection(t *testing.T) { + want := map[DecisionEffect]bool{ + DecisionEffectAllow: false, + DecisionEffectAuditOnly: false, + DecisionEffectDeny: true, + DecisionEffectRedact: true, + DecisionEffectNotApplicable: true, + } + for _, effect := range DecisionEffects() { + expected, ok := want[effect] + if !ok { + t.Fatalf("effect %q has no stated direction; decide it under FLEX-DEC-2026-018", effect) + } + if effect.Restricts() != expected { + t.Errorf("%q.Restricts() = %v, want %v", effect, effect.Restricts(), expected) + } + } +} diff --git a/workplans/FLEX-WP-0031-decision-record-emission.md b/workplans/FLEX-WP-0031-decision-record-emission.md index 503dacf..edaa77c 100644 --- a/workplans/FLEX-WP-0031-decision-record-emission.md +++ b/workplans/FLEX-WP-0031-decision-record-emission.md @@ -61,7 +61,7 @@ and the node disk is at 84 %. ```task id: FLEX-WP-0031-T02 -status: todo +status: wait priority: high state_hub_task_id: "89661908-ca9a-5e4a-a0a5-62d1a9e02568" ``` @@ -72,22 +72,81 @@ class exactly as `cadence.yaml` publishes them, and a token lane routed via `warden route find`. audit-core has said it accepts the classification as supplied and will not infer it. Gate: sender registered; no secret in any file. -## 3. Transactional outbox, heartbeat and reconciliation counts +Requested 2026-09-23 (hub message `6044ed35`). The request asks for six senders, +one per pin (`flex-auth.`), because each pin can only reconcile its own +count. It also asks for `tenants: ["*"]` with a justification, +`may_read: true` for `GET /v1/reconciliation`, and per-class `heartbeat_classes` +at 86400 s for the four rare classes. It asks audit-core to rule on whether +FLEX-DEC-2026-018's failure-path exception counts as a `completeness_trade`, +which `senders.py` forbids for a load-bearing source. warden route has no +catalog lane for audit-core sender tokens, and audit-core was asked to name one. +Waiting on audit-core. + +## 3. Durable outbox and the release rule ```task id: FLEX-WP-0031-T03 -status: todo +status: done priority: high state_hub_task_id: "a59603d5-8954-5b05-b1a0-b143c82e439b" ``` -Emit one event per decision into a local outbox, drained to `audit-core` -`POST /v1/events`; a daily `flex-auth.decision.heartbeat`; committed counts per -class exposed for `GET /v1/reconciliation`; lag bound per `cadence.yaml`. -Gate: `cadence.yaml` validates under the net-kingdom emission-cadence profile -checker with the inventory supplied as `--rare-load-bearing`/`--load-bearing` -assertions; tests assert the declared classes equal the `DecisionEffect` -vocabulary so a new effect cannot ship unclassified. +Split on 2026-09-23 under the per-task budget. T05 and T06 carry the rest of +the original scope, and the original gate is divided between the three. + +Done 2026-09-23: + +- `internal/emission`: a durable outbox with one fsync per commit. A batch + commits once. Each event carries exactly the eight fields audit-core's + `normalize()` requires. +- Event ids are random and the decision id travels as `correlation_id`. + Decision ids are content digests, so two identical checks share one, and + idempotency would merge them. +- A torn tail is cut on open, and a failed commit is cut back to the last + committed size. Committed counts per class are rebuilt on restart. +- The engine applies `FLEX-DEC-2026-018` in `Check`, `BatchCheck` and + `ListAllowed`. A withheld decision answers 503. +- `serve --outbox-dir --emission-source` enables it, and `GET /v1/emission` + reports committed and `released_uncommitted` counts per class. +- `api.DecisionEffects()` lists the effect vocabulary. A source-parsing test + fails on any constant it misses, and a second test fails unless the + load-bearing classes in `cadence.yaml` equal that vocabulary. +- `make test` (race) passes, and a local serve smoke test committed and counted + one deny and one allow. Nothing is deployed and no chart changed. + +## 5. Heartbeat and drain to audit-core + +```task +id: FLEX-WP-0031-T05 +status: todo +priority: high +``` + +- Emit a daily `audit-core.heartbeat` event per rare class, with `data.class` + set to the class, following audit-core's `stream_findings` shape. +- Correct `cadence.yaml`, which still names a single + `flex-auth.decision.heartbeat` class. +- Drain committed events to `POST /v1/events` with `Idempotency-Key` set to the + event id, keep a drain cursor, and expose outbox depth and age against the + `lag_bound`. +- The drain stays disabled until T02 admits the senders. + +## 6. Reconciliation, profile check, and storage + +```task +id: FLEX-WP-0031-T06 +status: todo +priority: high +``` + +- Compare committed counts per class and window with audit-core's + `GET /v1/reconciliation`. Divergence is a finding, and undrained events count + as lag, not divergence. +- Validate `cadence.yaml` with the net-kingdom emission-cadence profile checker, + supplying the inventory as `--rare-load-bearing`/`--load-bearing`. +- Add an optional PVC to the chart for the outbox, sized against the node disk + (84 % on 2026-09-23). Rolling it out to any pin is a production change and + needs the founder's go-ahead. ## 4. Close G2