diff --git a/charts/access-engine/templates/deployment.yaml b/charts/access-engine/templates/deployment.yaml index c9dd6d6..744c8b2 100644 --- a/charts/access-engine/templates/deployment.yaml +++ b/charts/access-engine/templates/deployment.yaml @@ -68,6 +68,16 @@ spec: - {{ required "envelopeSigning.kid must be independently admitted" .Values.envelopeSigning.kid | quote }} {{- end }} {{- if .Values.outbox.enabled }} + {{- $limit := .Values.outbox.maxBytes }} + {{- $numericKind := has (kindOf $limit) (list "int" "int64" "int32" "uint" "uint64" "float64" "float32") }} + {{- $asFloat := float64 $limit }} + {{- $validNumber := and $numericKind (eq $asFloat (floor $asFloat)) }} + {{- $validString := and (kindIs "string" $limit) (regexMatch "^[1-9][0-9]*$" (toString $limit)) }} + {{- if or (not (or $validNumber $validString)) (lt $asFloat 1.0) (gt $asFloat 67108864.0) }} + {{- fail "outbox.maxBytes must be an exact integer between1 and67108864" }} + {{- end }} + - --outbox-max-bytes + - {{ printf "%.0f" $asFloat | quote }} - --outbox-dir - {{ .Values.outbox.mountPath | quote }} - --emission-source diff --git a/charts/access-engine/values.yaml b/charts/access-engine/values.yaml index 4f967b4..af3fbc9 100644 --- a/charts/access-engine/values.yaml +++ b/charts/access-engine/values.yaml @@ -15,6 +15,8 @@ outbox: source: "" mountPath: /var/lib/access-engine/outbox size: 1Gi + # Source ceiling; PVC requested capacity alone is not a filesystem quota. + maxBytes: 67108864 storageClassName: "" existingClaim: "" diff --git a/charts/flex-auth/templates/deployment.yaml b/charts/flex-auth/templates/deployment.yaml index 4889a30..62f84ba 100644 --- a/charts/flex-auth/templates/deployment.yaml +++ b/charts/flex-auth/templates/deployment.yaml @@ -67,6 +67,16 @@ spec: - {{ required "envelopeSigning.kid must be independently admitted" .Values.envelopeSigning.kid | quote }} {{- end }} {{- if .Values.outbox.enabled }} + {{- $limit := .Values.outbox.maxBytes }} + {{- $numericKind := has (kindOf $limit) (list "int" "int64" "int32" "uint" "uint64" "float64" "float32") }} + {{- $asFloat := float64 $limit }} + {{- $validNumber := and $numericKind (eq $asFloat (floor $asFloat)) }} + {{- $validString := and (kindIs "string" $limit) (regexMatch "^[1-9][0-9]*$" (toString $limit)) }} + {{- if or (not (or $validNumber $validString)) (lt $asFloat 1.0) (gt $asFloat 67108864.0) }} + {{- fail "outbox.maxBytes must be an exact integer between1 and67108864" }} + {{- end }} + - --outbox-max-bytes + - {{ printf "%.0f" $asFloat | quote }} - --outbox-dir - {{ .Values.outbox.mountPath | quote }} - --emission-source diff --git a/charts/flex-auth/values.yaml b/charts/flex-auth/values.yaml index 2b780ed..97a262e 100644 --- a/charts/flex-auth/values.yaml +++ b/charts/flex-auth/values.yaml @@ -15,6 +15,8 @@ outbox: source: "" mountPath: /var/lib/flex-auth/outbox size: 1Gi + # Source ceiling; PVC requested capacity alone is not a filesystem quota. + maxBytes: 67108864 storageClassName: "" existingClaim: "" diff --git a/cmd/access-engine/main.go b/cmd/access-engine/main.go index 5da62ba..71dbd9d 100644 --- a/cmd/access-engine/main.go +++ b/cmd/access-engine/main.go @@ -334,6 +334,7 @@ func runServe(args []string, stdout, stderr io.Writer) int { signingKeyFile := fs.String("envelope-signing-key-file", "", "owner-mounted base64url Ed25519 seed file") signingKID := fs.String("envelope-signing-kid", "", "independently admitted envelope signing key id") approvalSourceFile := fs.String("approval-source-file", "", "owner fixed Approval HTTPS reader configuration file") + outboxMaxBytes := fs.Int64("outbox-max-bytes", emission.MaxOutboxBytes, "bounded outbox payload bytes, maximum67108864") 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") auditURL := fs.String("audit-url", "", "admitted audit-core /v1/events endpoint; empty disables drain and heartbeats") @@ -378,7 +379,7 @@ func runServe(args []string, stdout, stderr io.Writer) int { } var outbox *emission.Outbox if *outboxDir != "" { - outbox, err = emission.Open(*outboxDir, *emissionSource) + outbox, err = emission.OpenBounded(*outboxDir, *emissionSource, *outboxMaxBytes) if err != nil { return fail(stderr, err) } diff --git a/cmd/access-engine/outbox_capacity_test.go b/cmd/access-engine/outbox_capacity_test.go new file mode 100644 index 0000000..2f60f8d --- /dev/null +++ b/cmd/access-engine/outbox_capacity_test.go @@ -0,0 +1,58 @@ +package main + +import ( + "bytes" + "context" + "github.com/netkingdom/access-engine/internal/emission" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "testing" +) + +func TestActualHTTPAllEffectsWithheldAtOutboxCapacity(t *testing.T) { + engine, err := buildEngine(context.Background(), opsPath("registry_snapshot.json"), opsPath("policy_package.md"), "") + if err != nil { + t.Fatal(err) + } + outbox, err := emission.OpenBounded(t.TempDir(), "synthetic-capacity", 1) + if err != nil { + t.Fatal(err) + } + defer outbox.Close() + engine.SetDecisionLog(outbox) + server := httptest.NewServer(newServeMux(engine)) + defer server.Close() + for _, name := range []string{"check_request_allow_adm.json", "check_request_deny_ttl_above_max.json"} { + raw, err := os.ReadFile(opsPath(name)) + if err != nil { + t.Fatal(err) + } + response, err := http.Post(server.URL+"/v1/check", "application/json", bytes.NewReader(raw)) + if err != nil { + t.Fatal(err) + } + body, _ := io.ReadAll(response.Body) + response.Body.Close() + if response.StatusCode != http.StatusServiceUnavailable || bytes.Contains(body, []byte(`"effect"`)) { + t.Fatalf("capacity emitted decision: %d", response.StatusCode) + } + } + status := outbox.Status() + if status.CommittedBytes != 0 || status.Pending != 0 { + t.Fatal("failed record claimed committed bytes") + } + for _, count := range status.ReleasedUncommitted { + if count != 0 { + t.Fatal("withheld effect counted released") + } + } + // An invalid configured cap fails production serve startup before HTTP. + var output, errors bytes.Buffer + code := run([]string{"serve", "--registry", opsPath("registry_snapshot.json"), "--policy", opsPath("policy_package.md"), "--outbox-dir", filepath.Join(t.TempDir(), "outbox"), "--emission-source", "synthetic-capacity", "--outbox-max-bytes", "67108865"}, &output, &errors) + if code == 0 { + t.Fatal("source ceiling bypassed at CLI") + } +} diff --git a/docs/decision-emission-operations.md b/docs/decision-emission-operations.md index fe6d5d6..632102c 100644 --- a/docs/decision-emission-operations.md +++ b/docs/decision-emission-operations.md @@ -68,3 +68,56 @@ accepted_at, and local windows would use commit time. The current ingest receipt contains no acceptance timestamp. Comparing these directly would manufacture divergence after delayed delivery. Requirements and native acceptance gates are in `blocked-task-dependencies-2026-10-04.md`. + +## Bounded single writer for native signer admission + +`Open` now requires the admitted Linux filesystem locking semantics. It obtains +an exclusive nonblocking OS `flock` on owner/current-group regular `writer.lock` before +reading the cursor or inspecting/recovering the journal. The FD remains open for +the outbox lifetime; another process refuses before journal mutation. Close waits +for an in-flight drain/cursor commit, closes the journal and releases the lock; +process death also releases it. The lock file is retained, never unlinked. It is created 0600; writable-PVC +fsGroup reconciliation may widen it to 0660 for the same admitted UID/GID, which +is supported. The lock, journal and cursor refuse foreign ownership/group, other-access, +hardlinks, symlinks and nonregular files. The lock is nonsecret; private seed permissions remain separately strict. This +protects competing admitted producers, not hostile same-UID unlink/rename or +privileged host tampering. Native PVC admission must prove actual filesystem +locking semantics; RWO and Recreate with one replica alone are insufficient. + +`serve --outbox-max-bytes` and chart `outbox.maxBytes` select a positive integer +at most **67108864** (64MiB), also the default and source hard ceiling. Each +newline-terminated record and each atomic aggregate batch is capped at **2MiB**. +The cap is checked before any write; a rejected batch cannot partially commit. +`GET /v1/emission` reports max/committed/remaining bytes and byte-cap exhaustion +and `writer_fenced` without payloads. The requested PVC capacity is not a filesystem quota. The +initial retained Secrets pin proposes a new 1Gi local-path claim; no existing +claim was observed. Reclaim/retention, permissions, volume identity and native +filesystem semantics require separate owner admission. + +Startup refuses files above the selected ceiling and streams records with a +2MiB line bound, rather than loading the whole log. The cursor is bounded to +4KiB. A newline-terminated corrupt/oversized entry is preserved and refused; +only the existing bounded unterminated final write repair truncates uncommitted +bytes. Initial journal directory entries and successful repair truncation are +synced. Startup failure releases the OS lock. Existing oversized committed +state is never automatically deleted, compacted, rotated or accepted by raising +a runtime cap above the hard maximum. + +Write/fsync errors withhold all effects under current GH-DEC-2026-024, including +DENY. A partial append is truncated back to the committed prefix and synced; +if that rollback cannot be confirmed, the writer is fenced for its lifetime +until Close/restart and independent recovery. Capacity refusal touches no prior +bytes. Actual `/dev/full` ENOSPC and `/dev/null` fsync failure tests exercise +closed failure/fencing. A filled bounded journal remains full across restart; +acknowledged entries are still retained. At 75% capacity request an exact reviewed +export/retention action, and at 90% stop initiating protected work. These are +operator thresholds, not an invented automated retention worker or audit-core +admission. No underlying disk-full condition is repaired by retrying a protected +operation or deleting evidence. + +The retained lifecycle signer rollout must use a new immutable CI candidate +containing reviewed hardening. Native admission includes actual competing-process +exclusion, capacity/failure/restart and response/original-record checks. Local +source tests, Helm keep annotation and a volume request do not prove node-loss +recovery, receiver custody or complete emission adoption. FLEX-WP-0031's remaining +native/G2 obligations stay open. diff --git a/internal/emission/delivery.go b/internal/emission/delivery.go index e138642..b7e4bb8 100644 --- a/internal/emission/delivery.go +++ b/internal/emission/delivery.go @@ -83,13 +83,25 @@ type drainCursor struct { } func (o *Outbox) loadCursor() error { - b, err := os.ReadFile(filepath.Join(filepath.Dir(o.path), "drain.json")) + file, err := openOwnedFile(filepath.Join(filepath.Dir(o.path), "drain.json"), os.O_RDONLY, 0) if errors.Is(err, os.ErrNotExist) { return nil } if err != nil { return fmt.Errorf("read drain cursor: %w", err) } + defer file.Close() + info, err := file.Stat() + if err != nil || !info.Mode().IsRegular() { + return errors.New("drain cursor must be a regular file") + } + b, err := io.ReadAll(io.LimitReader(file, 4097)) + if err != nil { + return err + } + if len(b) > 4096 { + return errors.New("drain cursor exceeds 4096 bytes") + } if err := json.Unmarshal(b, &o.cursor); err != nil { return fmt.Errorf("decode drain cursor: %w", err) } @@ -101,13 +113,13 @@ func (o *Outbox) loadCursor() error { // nextEntry reads only the committed prefix; mu protects the size snapshot. func (o *Outbox) nextEntry() (entry, int64, error) { - f, err := os.Open(o.path) + f, err := openOwnedFile(o.path, os.O_RDONLY, 0) if err != nil { return entry{}, 0, err } defer f.Close() r := bufio.NewReader(io.NewSectionReader(f, o.cursor.Offset, o.size-o.cursor.Offset)) - line, err := r.ReadBytes('\n') + line, err := boundedLine(r) if err != nil { return entry{}, 0, err } @@ -183,6 +195,10 @@ func (o *Outbox) Drain(ctx context.Context, config DeliveryConfig, limit int) (i delivered := 0 for delivered < limit { o.mu.Lock() + if o.closed { + o.mu.Unlock() + return delivered, os.ErrClosed + } if o.cursor.Seq == o.seq { o.mu.Unlock() return delivered, nil diff --git a/internal/emission/hardening_linux_test.go b/internal/emission/hardening_linux_test.go new file mode 100644 index 0000000..e781d6b --- /dev/null +++ b/internal/emission/hardening_linux_test.go @@ -0,0 +1,349 @@ +//go:build linux + +package emission + +import ( + "bufio" + "bytes" + "context" + "errors" + "fmt" + "github.com/netkingdom/access-engine/pkg/api" + "io" + "os" + "os/exec" + "path/filepath" + "strings" + "syscall" + "testing" + "time" +) + +func boundedFixture() api.DecisionEnvelope { + return api.DecisionEnvelope{ID: "decision:bounded", Effect: api.DecisionEffectDeny, Resource: api.ResourceRef{ID: "synthetic", Type: "tenant", System: "tenant-engine"}} +} +func fixedOutbox(t *testing.T, dir string, max int64) *Outbox { + t.Helper() + o, err := OpenBounded(dir, "flex-auth.test", max) + if err != nil { + t.Fatal(err) + } + o.SetClock(func() time.Time { return time.Date(2026, 10, 16, 12, 0, 0, 0, time.UTC) }) + return o +} +func TestOutboxLockChild(t *testing.T) { + dir := os.Getenv("ACCESS_OUTBOX_LOCK_CHILD") + if dir == "" { + return + } + o, err := Open(dir, "flex-auth.test") + if err != nil { + fmt.Println("REFUSED") + os.Exit(23) + } + fmt.Println("READY") + var one [1]byte + os.Stdin.Read(one[:]) + o.Close() + os.Exit(0) +} +func startLockChild(t *testing.T, dir string) (*exec.Cmd, io.WriteCloser, *bufio.Reader) { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + t.Cleanup(cancel) + child := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestOutboxLockChild$") + child.Env = []string{"ACCESS_OUTBOX_LOCK_CHILD=" + dir} + input, err := child.StdinPipe() + if err != nil { + t.Fatal(err) + } + output, err := child.StdoutPipe() + if err != nil { + t.Fatal(err) + } + if err := child.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { child.Process.Kill() }) + return child, input, bufio.NewReader(output) +} +func TestActualProcessWriterExclusionAndCrashRelease(t *testing.T) { + dir := t.TempDir() + holder, input, output := startLockChild(t, dir) + ready, err := output.ReadString('\n') + if err != nil || ready != "READY\n" { + t.Fatal("holder failed") + } + before, _ := os.ReadFile(filepath.Join(dir, outboxFile)) + if _, err := Open(dir, "flex-auth.test"); err == nil { + t.Fatal("second process writer opened") + } + after, _ := os.ReadFile(filepath.Join(dir, outboxFile)) + if !bytes.Equal(before, after) { + t.Fatal("contender mutated evidence") + } + other := fixedOutbox(t, t.TempDir(), MaxOutboxBytes) + other.Close() + contender, _, result := startLockChild(t, dir) + text, _ := result.ReadString('\n') + if contender.Wait() == nil || text != "REFUSED\n" { + t.Fatal("OS contender did not refuse") + } + holder.Process.Kill() + holder.Wait() + input.Close() + reopened := fixedOutbox(t, dir, MaxOutboxBytes) + if err := reopened.Append(boundedFixture()); err != nil { + t.Fatal(err) + } + reopened.Close() + clean, cleanInput, cleanOutput := startLockChild(t, dir) + if line, _ := cleanOutput.ReadString('\n'); line != "READY\n" { + t.Fatal("close did not release lock") + } + cleanInput.Close() + if clean.Wait() != nil { + t.Fatal("clean child close failed") + } + final := fixedOutbox(t, dir, MaxOutboxBytes) + defer final.Close() + if final.Status().Committed[Class(api.DecisionEffectDeny)] != 1 { + t.Fatal("crash/close recovery changed committed prefix") + } + if _, err := os.Stat(filepath.Join(dir, "writer.lock")); err != nil { + t.Fatal("lock file was unlinked") + } +} +func TestExactCapacityBatchEntryAndRestartBoundaries(t *testing.T) { + probe := fixedOutbox(t, t.TempDir(), MaxOutboxBytes) + if err := probe.Append(boundedFixture()); err != nil { + t.Fatal(err) + } + size := probe.Status().CommittedBytes + probe.Close() + dir := t.TempDir() + o := fixedOutbox(t, dir, size) + if err := o.Append(boundedFixture()); err != nil { + t.Fatal(err) + } + before, _ := os.ReadFile(o.path) + if err := o.Append(boundedFixture()); !errors.Is(err, ErrCapacity) { + t.Fatal("over-cap append accepted") + } + after, _ := os.ReadFile(o.path) + if !bytes.Equal(before, after) || o.seq != 1 || o.cursor.Seq != 0 { + t.Fatal("capacity refusal changed prefix/sequence/cursor") + } + status := o.Status() + if status.RemainingBytes != 0 || !status.CapacityExhausted || status.MaxBytes != size { + t.Fatal("capacity status inaccurate") + } + o.Close() + o = fixedOutbox(t, dir, size) + if o.Status().CommittedBytes != size || o.seq != 1 { + t.Fatal("full outbox restart lost evidence") + } + o.Close() + for _, max := range []int64{size - 1, size*2 - 1} { + batch := fixedOutbox(t, t.TempDir(), max) + count := 1 + if max > size { + count = 2 + } + items := make([]api.DecisionEnvelope, count) + for i := range items { + items[i] = boundedFixture() + } + if err := batch.AppendBatch(items); !errors.Is(err, ErrCapacity) { + t.Fatal("over-cap batch accepted") + } + info, _ := os.Stat(batch.path) + if info.Size() != 0 || batch.seq != 0 { + t.Fatal("batch partially written") + } + batch.Close() + } + huge := fixedOutbox(t, t.TempDir(), MaxOutboxBytes) + defer huge.Close() + entry := boundedFixture() + entry.Diagnostics = map[string]any{"payload": strings.Repeat("x", MaxEntryBytes)} + if !errors.Is(huge.Append(entry), ErrCapacity) { + t.Fatal("oversized entry accepted") + } + info, _ := os.Stat(huge.path) + if info.Size() != 0 { + t.Fatal("oversized entry written") + } + batchMemory := fixedOutbox(t, t.TempDir(), MaxOutboxBytes) + defer batchMemory.Close() + medium := boundedFixture() + medium.Diagnostics = map[string]any{"payload": strings.Repeat("m", MaxBatchBytes/2)} + if !errors.Is(batchMemory.AppendBatch([]api.DecisionEnvelope{medium, medium}), ErrCapacity) { + t.Fatal("aggregate memory bound exceeded") + } + if info, _ := os.Stat(batchMemory.path); info.Size() != 0 { + t.Fatal("oversized batch partially written") + } + for _, limit := range []int64{0, -1, MaxOutboxBytes + 1} { + if _, err := OpenBounded(t.TempDir(), "flex-auth.test", limit); err == nil { + t.Fatal("invalid configured limit") + } + } +} +func TestBoundedStartupPreservesOversizedCommittedEvidence(t *testing.T) { + for _, sample := range []struct { + name string + raw []byte + max int64 + }{{"file", bytes.Repeat([]byte("x"), 1025), 1024}, {"line", append(bytes.Repeat([]byte("x"), MaxEntryBytes+1), '\n'), MaxOutboxBytes}, {"oversized-torn", bytes.Repeat([]byte("x"), MaxEntryBytes+1), MaxOutboxBytes}, {"terminated-corrupt", []byte("{bad}\n"), MaxOutboxBytes}} { + t.Run(sample.name, func(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, outboxFile) + os.WriteFile(path, sample.raw, 0600) + if _, err := OpenBounded(dir, "flex-auth.test", sample.max); err == nil { + t.Fatal("corrupt/oversized evidence accepted") + } + after, _ := os.ReadFile(path) + if !bytes.Equal(after, sample.raw) { + t.Fatal("committed evidence discarded") + } + }) + } + dir := t.TempDir() + cursor := bytes.Repeat([]byte(" "), 4097) + os.WriteFile(filepath.Join(dir, "drain.json"), cursor, 0600) + if _, err := Open(dir, "flex-auth.test"); err == nil { + t.Fatal("unbounded cursor accepted") + } + after, _ := os.ReadFile(filepath.Join(dir, "drain.json")) + if !bytes.Equal(after, cursor) { + t.Fatal("invalid cursor changed") + } +} +func TestActualDiskFullAndFsyncFailureFenceUnconfirmedRollback(t *testing.T) { + for _, device := range []string{"/dev/full", "/dev/null"} { + t.Run(device, func(t *testing.T) { + o := fixedOutbox(t, t.TempDir(), MaxOutboxBytes) + defer o.Close() + if err := o.Append(boundedFixture()); err != nil { + t.Fatal(err) + } + before, _ := os.ReadFile(o.path) + original := o.file + fault, err := os.OpenFile(device, os.O_WRONLY, 0) + if err != nil { + t.Fatal(err) + } + o.file = fault + failure := o.Append(boundedFixture()) + if failure == nil { + t.Fatal("device write/sync failure accepted") + } + if device == "/dev/full" && !errors.Is(failure, syscall.ENOSPC) { + t.Fatal("did not exercise actual ENOSPC") + } + if device == "/dev/null" && !strings.Contains(failure.Error(), "sync outbox") { + t.Fatal("did not exercise actual fsync failure") + } + + if !o.writeFault || !o.Status().WriterFenced { + t.Fatal("unconfirmed rollback did not fence writer") + } + fault.Close() + o.file = original + if err := o.Append(boundedFixture()); err == nil { + t.Fatal("fenced writer resumed") + } + after, _ := os.ReadFile(o.path) + if !bytes.Equal(before, after) || o.seq != 1 { + t.Fatal("disk failure changed committed prefix") + } + }) + } +} + +func TestWriterLockRefusesUnsafeFilesAndStartupFailureReleases(t *testing.T) { + for _, kind := range []string{"symlink", "world-readable", "hardlink"} { + t.Run(kind, func(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "writer.lock") + switch kind { + case "symlink": + target := filepath.Join(t.TempDir(), "target") + os.WriteFile(target, []byte("private unchanged"), 0600) + os.Symlink(target, path) + case "world-readable": + os.WriteFile(path, nil, 0644) + case "hardlink": + target := filepath.Join(t.TempDir(), "target") + os.WriteFile(target, nil, 0600) + os.Link(target, path) + } + if _, err := Open(dir, "flex-auth.test"); err == nil { + t.Fatal("unsafe lock accepted") + } + }) + } + dir := t.TempDir() + os.WriteFile(filepath.Join(dir, outboxFile), []byte("bad\n"), 0600) + if _, err := Open(dir, "flex-auth.test"); err == nil { + t.Fatal("corrupt state accepted") + } + // The failed Open has no live writer and cannot strand an OS lock. Manual + // repair here is a disposable fixture action, never an automatic recovery. + os.WriteFile(filepath.Join(dir, outboxFile), nil, 0600) + o := fixedOutbox(t, dir, MaxOutboxBytes) + o.Close() +} + +func TestPVCFSGroupPermissionWideningRetainsWriterExclusion(t *testing.T) { + dir := t.TempDir() + o := fixedOutbox(t, dir, MaxOutboxBytes) + o.Close() + // Kubernetes writable-volume fsGroup reconciliation may OR group0660 on + // remount. This nonsecret lock must still work for the same admitted group. + if err := os.Chmod(filepath.Join(dir, "writer.lock"), 0660); err != nil { + t.Fatal(err) + } + o = fixedOutbox(t, dir, MaxOutboxBytes) + defer o.Close() + if _, err := Open(dir, "flex-auth.test"); err == nil { + t.Fatal("fsGroup-widened file lost exclusive lock") + } +} + +func TestJournalAndCursorRefuseAliasesAndFIFOs(t *testing.T) { + for _, name := range []string{outboxFile, "drain.json"} { + for _, kind := range []string{"symlink", "hardlink", "fifo"} { + t.Run(name+"/"+kind, func(t *testing.T) { + dir := t.TempDir() + target := filepath.Join(t.TempDir(), "evidence") + original := []byte("retained evidence\n") + if err := os.WriteFile(target, original, 0600); err != nil { + t.Fatal(err) + } + path := filepath.Join(dir, name) + var err error + switch kind { + case "symlink": + err = os.Symlink(target, path) + case "hardlink": + err = os.Link(target, path) + case "fifo": + err = syscall.Mkfifo(path, 0600) + } + if err != nil { + t.Fatal(err) + } + if o, err := Open(dir, "flex-auth.test"); err == nil { + o.Close() + t.Fatal("unsafe evidence path accepted") + } + after, err := os.ReadFile(target) + if err != nil || !bytes.Equal(original, after) { + t.Fatal("refusal changed aliased evidence") + } + }) + } + } +} diff --git a/internal/emission/lock_linux.go b/internal/emission/lock_linux.go new file mode 100644 index 0000000..3dbec8b --- /dev/null +++ b/internal/emission/lock_linux.go @@ -0,0 +1,44 @@ +//go:build linux + +package emission + +import ( + "errors" + "os" + "syscall" +) + +func acquireWriterLock(path string) (*os.File, error) { + file, err := openOwnedFile(path, os.O_CREATE|os.O_RDWR, 0600) + if err != nil { + return nil, err + } + if err := syscall.Flock(int(file.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil { + file.Close() + return nil, errors.New("outbox already has a writer or filesystem lock unavailable") + } + return file, nil +} +func infoSys(info os.FileInfo) (*syscall.Stat_t, bool) { + if info == nil { + return nil, false + } + stat, ok := info.Sys().(*syscall.Stat_t) + return stat, ok +} + +// Shared journal/cursor aliases must not evade directory-based exclusion. +// Nonblocking open also rejects unexpected FIFOs/devices without waiting. +func openOwnedFile(path string, flags int, mode os.FileMode) (*os.File, error) { + file, err := os.OpenFile(path, flags|syscall.O_NOFOLLOW|syscall.O_NONBLOCK, mode) + if err != nil { + return nil, err + } + info, err := file.Stat() + stat, ok := infoSys(info) + if err != nil || !ok || !info.Mode().IsRegular() || info.Mode().Perm()&0007 != 0 || stat.Nlink != 1 || stat.Uid != uint32(os.Geteuid()) || (info.Mode().Perm()&0070 != 0 && stat.Gid != uint32(os.Getegid())) { + file.Close() + return nil, errors.New("outbox file must be a single-link owner/current-group regular file") + } + return file, nil +} diff --git a/internal/emission/lock_other.go b/internal/emission/lock_other.go new file mode 100644 index 0000000..ee992f1 --- /dev/null +++ b/internal/emission/lock_other.go @@ -0,0 +1,16 @@ +//go:build !linux + +package emission + +import ( + "errors" + "os" +) + +func acquireWriterLock(string) (*os.File, error) { + return nil, errors.New("outbox requires admitted Linux filesystem writer exclusion") +} + +func openOwnedFile(string, int, os.FileMode) (*os.File, error) { + return nil, errors.New("outbox requires admitted Linux filesystem writer exclusion") +} diff --git a/internal/emission/outbox.go b/internal/emission/outbox.go index b3ab758..7efd30d 100644 --- a/internal/emission/outbox.go +++ b/internal/emission/outbox.go @@ -5,12 +5,14 @@ package emission import ( + "bufio" "bytes" "crypto/rand" "encoding/hex" "encoding/json" "errors" "fmt" + "io" "os" "path/filepath" "sync" @@ -27,6 +29,11 @@ const ClassPrefix = "flex-auth.decision." const PlatformTenant = "tenant:platform" const outboxFile = "outbox.jsonl" +const MaxOutboxBytes int64 = 64 * 1024 * 1024 +const MaxEntryBytes = 2 * 1024 * 1024 +const MaxBatchBytes = MaxEntryBytes + +var ErrCapacity = errors.New("outbox capacity exceeded; committed evidence retained") // Class returns the event class for a decision effect. func Class(effect api.DecisionEffect) string { @@ -57,6 +64,11 @@ type entry struct { // class, and what it released without committing. type Status struct { Source string `json:"source"` + MaxBytes int64 `json:"max_bytes"` + CommittedBytes int64 `json:"committed_bytes"` + RemainingBytes int64 `json:"remaining_bytes"` + WriterFenced bool `json:"writer_fenced"` + CapacityExhausted bool `json:"capacity_exhausted"` Committed map[string]uint64 `json:"committed"` ReleasedUncommitted map[string]uint64 `json:"released_uncommitted"` // ReleasedUncommitted is held in memory: a restart during an outbox outage @@ -75,6 +87,10 @@ type Outbox struct { mu sync.Mutex file *os.File + writerLock *os.File + maxBytes int64 + closed bool + writeFault bool seq uint64 size int64 // bytes committed; a failed commit is cut back to it committed map[string]uint64 @@ -85,15 +101,35 @@ type Outbox struct { } // Open opens or creates the outbox in dir and rebuilds committed counts from it. -func Open(dir, source string) (*Outbox, error) { +func Open(dir, source string) (*Outbox, error) { return OpenBounded(dir, source, MaxOutboxBytes) } + +// OpenBounded acquires OS writer exclusion before inspecting/recovering evidence. +// The selected limit cannot exceed the reviewed 64MiB source ceiling. +func OpenBounded(dir, source string, maxBytes int64) (*Outbox, error) { + if maxBytes < 1 || maxBytes > MaxOutboxBytes { + return nil, errors.New("outbox byte limit must be between 1 and 67108864") + } + 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) } + writerLock, err := acquireWriterLock(filepath.Join(dir, "writer.lock")) + if err != nil { + return nil, err + } + success := false + defer func() { + if !success { + writerLock.Close() + } + }() o := &Outbox{ source: source, + writerLock: writerLock, + maxBytes: maxBytes, path: filepath.Join(dir, outboxFile), clock: time.Now, committed: map[string]uint64{}, @@ -109,56 +145,102 @@ func Open(dir, source string) (*Outbox, error) { if o.cursor.Seq > o.seq || o.cursor.Offset > o.size { return nil, errors.New("drain cursor is ahead of committed outbox") } - file, err := os.OpenFile(o.path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o640) + file, err := openOwnedFile(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 + if err := syncOutboxDirectory(dir); err != nil { + file.Close() + return nil, err + } + success = true return o, nil } func (o *Outbox) load() error { - data, err := os.ReadFile(o.path) + file, err := openOwnedFile(o.path, os.O_RDONLY, 0) 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) + defer file.Close() + info, err := file.Stat() + if err != nil { + return err + } + if !info.Mode().IsRegular() { + return errors.New("outbox must be a regular file") + } + if info.Size() > o.maxBytes { + return ErrCapacity + } + reader := bufio.NewReaderSize(file, 64*1024) + var good int64 + for line := 1; ; line++ { + raw, err := boundedLine(reader) + if err == io.EOF { + // Only an unterminated final write is uncommitted. Never discard a + // newline-terminated corrupt or oversized record. + o.size = good + if len(raw) > 0 { + if err := os.Truncate(o.path, good); err != nil { + return err + } + if err := file.Sync(); err != nil { + return err + } + } + break + } + if err != nil { + return fmt.Errorf("outbox line %d: %w", line, err) } var e entry - decoder := json.NewDecoder(bytes.NewReader(data[good : good+end])) + decoder := json.NewDecoder(bytes.NewReader(raw)) decoder.UseNumber() if err := decoder.Decode(&e); err != nil { return fmt.Errorf("outbox line %d is corrupt: %w", line, err) } + if _, err := decoder.Token(); err != io.EOF { + return errors.New("trailing outbox entry data") + } if e.Seq != o.seq+1 || e.Event.Source != o.source { return fmt.Errorf("outbox line %d has inconsistent sequence or source", line) } o.seq = e.Seq o.committed[e.Event.Type]++ - good += end + 1 - if e.Seq == o.cursor.Seq && int64(good) != o.cursor.Offset { + good += int64(len(raw)) + if e.Seq == o.cursor.Seq && good != o.cursor.Offset { return errors.New("drain cursor does not match outbox") } o.noteHeartbeat(e.Event) } - o.size = int64(good) if o.cursor.Seq > o.seq { return errors.New("drain cursor is ahead of outbox") } return nil } +// boundedLine bounds startup and delivery allocations independently of log size. +// Its 2MiB limit includes the record newline. +func boundedLine(reader *bufio.Reader) ([]byte, error) { + var line []byte + for { + part, err := reader.ReadSlice('\n') + if len(line)+len(part) > MaxEntryBytes { + return nil, ErrCapacity + } + line = append(line, part...) + if err == bufio.ErrBufferFull { + continue + } + return line, err + } +} + // SetClock overrides the commit clock. Tests use it. func (o *Outbox) SetClock(clock func() time.Time) { o.mu.Lock() @@ -168,9 +250,17 @@ func (o *Outbox) SetClock(clock func() time.Time) { // Close closes the outbox file. func (o *Outbox) Close() error { + // Drain holds drainMu through HTTP and cursor commit. Do not release writer + // exclusion while an old delivery can still mutate the cursor. + o.drainMu.Lock() + defer o.drainMu.Unlock() o.mu.Lock() defer o.mu.Unlock() - return o.file.Close() + if o.closed { + return nil + } + o.closed = true + return errors.Join(o.file.Close(), o.writerLock.Close()) } // Append commits one decision; it satisfies decision.DecisionRecorder. @@ -197,6 +287,12 @@ func (o *Outbox) AppendBatch(decisions []api.DecisionEnvelope) error { // appendEvents requires mu; heartbeats use the same durable commit as decisions. func (o *Outbox) appendEvents(now string, events []Event) error { + if o.closed { + return os.ErrClosed + } + if o.writeFault { + return errors.New("outbox rollback unconfirmed; writer fenced") + } var buf []byte seq := o.seq for _, event := range events { @@ -205,6 +301,9 @@ func (o *Outbox) appendEvents(now string, events []Event) error { if err != nil { return fmt.Errorf("marshal outbox entry: %w", err) } + if len(line)+1 > MaxEntryBytes || len(buf)+len(line)+1 > MaxBatchBytes || int64(len(buf))+int64(len(line))+1 > o.maxBytes-o.size { + return ErrCapacity + } buf = append(append(buf, line...), '\n') } if _, err := o.file.Write(buf); err != nil { @@ -226,12 +325,26 @@ func (o *Outbox) appendEvents(now string, events []Event) error { // 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)) + err := o.file.Truncate(o.size) + if err == nil { + err = o.file.Sync() + } + if err != nil { + o.writeFault = true + return errors.Join(cause, fmt.Errorf("outbox rollback unconfirmed; writer fenced: %w", err)) } return cause } +func syncOutboxDirectory(dir string) error { + file, err := os.Open(dir) + if err != nil { + return err + } + defer file.Close() + return file.Sync() +} + // NoteReleasedUncommitted counts a restriction released after its commit // failed. Every non-zero count is a finding. func (o *Outbox) NoteReleasedUncommitted(effect api.DecisionEffect) { @@ -245,7 +358,8 @@ func (o *Outbox) Status() Status { o.mu.Lock() defer o.mu.Unlock() status := Status{ - Source: o.source, + Source: o.source, + MaxBytes: o.maxBytes, CommittedBytes: o.size, RemainingBytes: o.maxBytes - o.size, CapacityExhausted: o.size == o.maxBytes, WriterFenced: o.writeFault, Committed: map[string]uint64{}, ReleasedUncommitted: map[string]uint64{}, Means: "committed counts this outbox holds since it was created; " + diff --git a/tests/test_outbox_chart.py b/tests/test_outbox_chart.py index 8eff7de..e94f935 100644 --- a/tests/test_outbox_chart.py +++ b/tests/test_outbox_chart.py @@ -6,8 +6,8 @@ from pathlib import Path import yaml -def render(*settings, succeeds=True): - command = ["helm", "template", "flex-auth", "charts/flex-auth", "-f", "values/stage2-canary.yaml"] +def render(*settings, succeeds=True, chart="flex-auth"): + command = ["helm", "template", "flex-auth", "charts/"+chart, "-f", "values/stage2-canary.yaml"] for setting in settings: command += ["--set", setting] result = subprocess.run(command, capture_output=True, text=True, check=False) @@ -33,6 +33,7 @@ for auth in ((), ("callerAuth.mode=enforce", "callerAuth.binding=test=system:ser container = pod["containers"][0] args = container["args"] assert args[args.index("--outbox-dir") + 1] == "/var/lib/flex-auth/outbox" + assert args[args.index("--outbox-max-bytes") + 1] == "67108864" assert args[args.index("--emission-source") + 1] == "flex-auth.test" assert deployment(documents)["spec"]["strategy"]["type"] == "Recreate" pvc = next(d for d in documents if d["kind"] == "PersistentVolumeClaim") @@ -49,6 +50,26 @@ pod = deployment(existing)["spec"]["template"]["spec"] assert any(v.get("persistentVolumeClaim", {}).get("claimName") == "retained-outbox" for v in pod["volumes"]) render(*enabled, "replicaCount=2", succeeds=False) render("outbox.enabled=true", succeeds=False) +for chart in ("flex-auth", "access-engine"): + for value in ("0", "-1", "67108865", "1.0", "true", "null", "1e6", "999999999999999999999999999999"): + render(*enabled, "outbox.maxBytes="+value, succeeds=False, chart=chart) + for value in ("1", "67108864"): + docs=render(*enabled,"outbox.maxBytes="+value,chart=chart) + args=deployment(docs)["spec"]["template"]["spec"]["containers"][0]["args"] + assert args[args.index("--outbox-max-bytes")+1]==value +# Exercise actual YAML numeric decoding, including fractional/boolean values. +for chart in ("flex-auth", "access-engine"): + for value in (0.5, 1.5, 67108864.5, True, False, None, [], {}, 67108865): + with tempfile.TemporaryDirectory() as directory: + overlay = Path(directory) / "bound.yaml" + overlay.write_text(yaml.safe_dump({"outbox":{"maxBytes":value}})) + command = ["helm", "template", "flex-auth", "charts/"+chart, + "-f", "values/stage2-canary.yaml", "-f", str(overlay)] + for setting in enabled: + command += ["--set", setting] + result = subprocess.run(command, capture_output=True, text=True, check=False) + assert result.returncode != 0, f"accepted invalid capacity {value!r}" + assert "maxBytes" in result.stderr, result.stderr print("optional outbox chart checks passed") # Delivery rendering binds an existing admitted credential mount and narrow peers. diff --git a/workplans/FLEX-WP-0024-decision-envelope-authenticity.md b/workplans/FLEX-WP-0024-decision-envelope-authenticity.md index dca8757..bd7a83c 100644 --- a/workplans/FLEX-WP-0024-decision-envelope-authenticity.md +++ b/workplans/FLEX-WP-0024-decision-envelope-authenticity.md @@ -272,3 +272,14 @@ the issued claim; registry drift/refreshed facts cannot silently rebind it. Retained-registry HTTP and production serve proofs and negative joins pass. R3 preserves precise notbefore pending separate Approval-owner timestamp fidelity repair; no rounding waiver, copied registry algorithm or empty-registry workaround. + +2026-10-16 native admission prerequisite: fresh metadata confirms no namespace +PVC exists, and current outbox had neither OS writer exclusion nor a byte/startup +bound. Prepared existing-outbox hardening on exact canonical R3 base2b064ee: +nonblocking Linux flock held through Close/process lifetime before recovery, +64MiB source journal ceiling,2MiB record/atomic-batch bounds, streaming startup, +4KiB cursor, all-effect commit refusal and fenced unconfirmed rollback. Actual +process contention/crash release, capacity/prefix/restart and ENOSPC/fsync failure +proofs are part of the owner package. No new service, live mount, deletion or +custody grant; exact published successor, new PVC/PV retention/managed-package +owner disposition and independent consumer admission remain prerequisites.