diff --git a/module/configurations/local/policy-log.env b/module/configurations/local/policy-log.env index ed2b4b8b5..74467e2d4 100644 --- a/module/configurations/local/policy-log.env +++ b/module/configurations/local/policy-log.env @@ -11,4 +11,8 @@ # against a dedicated bucket whose IAM forbids deleting or replacing anything # under `entries/`. POLICY_LOG_BUCKET= +# Days every entry is held under a LOCKED object retention (nobody can delete or +# replace it before then; the bucket must have object retention enabled). A +# whole number of days, at most ten years. Required with the bucket. +POLICY_LOG_RETENTION_DAYS= POLICY_LOG_WAREHOUSE_ENDPOINT= diff --git a/module/configurations/local/policy-log.secret.env b/module/configurations/local/policy-log.secret.env index e5473960b..5e67db47f 100644 --- a/module/configurations/local/policy-log.secret.env +++ b/module/configurations/local/policy-log.secret.env @@ -3,10 +3,13 @@ # the database could rewrite the evidence that database's narrowings are # witnessed by. # -# A placeholder, never a usable value. Production supplies a unique, -# high-entropy token through Codefly's secret configuration provider, over -# HTTPS, with redirects refused. -POLICY_LOG_WAREHOUSE_TOKEN=local-dev-only-replace-me +# POLICY_LOG_WAREHOUSE_TOKEN is optional, together with +# POLICY_LOG_WAREHOUSE_ENDPOINT: both set configures the warehouse mirror, +# neither runs the witness with no mirror (stated at boot), and one without the +# other is refused. It is not assigned here because this local composition +# configures no mirror, and a shipped secret default may only hold a placeholder. +# Where a mirror is configured, it is a unique, high-entropy token delivered +# through Codefly's secret configuration provider. # Persistent witness identity, provisioned by the composition secret provider. # Canonical standard base64 of PKCS#8 Ed25519 DER; no generated boot fallback. diff --git a/module/services/policy-log/README.md b/module/services/policy-log/README.md index bb0150636..29df348ed 100644 --- a/module/services/policy-log/README.md +++ b/module/services/policy-log/README.md @@ -30,10 +30,36 @@ database restore must not restore the witness. Accounts must have no object write, signing-key read, warehouse writer, or witness administration capability. The witness holds no accounts database credential or dependency. -Delivery must provision a dedicated bucket, workload identity, private signing -key, and warehouse ingestion credential. Bucket IAM must confine the workload +Delivery must provision a dedicated bucket, workload identity and private +signing key, and — only where a warehouse mirror is configured — the warehouse +ingestion credential. Bucket IAM must confine the workload to that bucket and forbid deleting or replacing `entries/` objects; enforce the -immutable prefix with retention and IAM, in addition to application checks. +immutable prefix with retention and IAM, in addition to application checks. The +retention half is this service's at write time: every entry is created under a +**Locked** object retention for `POLICY_LOG_RETENTION_DAYS` (a whole number of +days, at most ten years), so the bucket must have object retention enabled, and +a bucket that cannot hold the lock refuses every append. The head carries no +retention, because it is rewritten on every append. + +Uploading under a retention lock needs `storage.objects.setRetention`. Grant it +**only** under an IAM condition on `resource.name.startsWith(".../objects/entries/")`, +never bucket-wide and never through `roles/storage.objectAdmin`: with it on the +head, a compromised workload could lock `head.json` far into the future and the +log could never advance again. The head's write authority (create and delete, +for the generation-matched replace) is conditioned on `head.json` alone, and +nothing is granted delete under `entries/`. Enabling object retention on a +bucket cannot be undone. + +A bucket without object retention, or an identity without `setRetention`, +still boots — `witness.Open` only reads — and then refuses every append with a +400 or 403 from the store. Those two symptoms name a provisioning defect, not +a witness one. + +An entry written whose head commit was lost (a timeout after the entry landed) +holds its sequence slot under the lock for the whole horizon, and it cannot be +removed out of band. The log is unstuck by retrying that operation's identical +entry — rebuildable from the orphaned object itself — which completes the head +commit; every other operation is answered `ErrPending` until then. The head's write authority is scoped to `head.json`. Bucket lifecycle deletion, rewriting an existing sequence, or restoring the head from a database backup would defeat the trust boundary and are not recovery procedures. Provision @@ -41,8 +67,8 @@ versioning and deletion protection outside the workload's administrative reach. The runtime exposes no delete, reset, unconditional overwrite, ambient signing-key source, insecure warehouse transport, or emulator option. The GCS constructor -refuses a storage emulator setting. The warehouse uses its own mandatory bearer -credential over HTTPS and refuses redirects. +refuses a storage emulator setting. A configured warehouse mirror uses its own +bearer credential, mandatory with its endpoint, over HTTPS and refuses redirects. The listener is private and must stay private. Composition must admit only the host's declared client through workload-authenticated mesh policy, resolve its @@ -201,18 +227,23 @@ and the detail goes to this service's own failure reporter instead. [code/main.go](code/main.go) is the binary. Every construction happens **before** `net.Listen`, in this order, and any failure is fatal: resolve the whole configuration group, read the Codefly-allocated port, parse the delivered signing-key -secret, open the bucket, build the warehouse writer, start the mirror -worker, then `witness.Open`, which verifies the entire committed chain under +secret, open the bucket, build the warehouse writer and start the mirror +worker (or wire `mirror.None` and say so, when no mirror is configured), then +`witness.Open`, which verifies the entire committed chain under that key. A witness that accepted an append it could not sign, store, or verify the history of would be worse than one that is down, because the caller would commit against it. -`POLICY_LOG_BUCKET`, `POLICY_LOG_WAREHOUSE_ENDPOINT` and +`POLICY_LOG_BUCKET`, `POLICY_LOG_RETENTION_DAYS`, `POLICY_LOG_WAREHOUSE_ENDPOINT` and `POLICY_LOG_WAREHOUSE_TOKEN` come from the `policy-log` configuration group and from nowhere else. There is no process-environment fallback, no default bucket and no development bypass, so an ambient value cannot make this service witness somewhere nobody provisioned — and the refusal names the missing keys, never the -delivered values, because one of the three is a credential. The native composition cannot yet start this service with its empty group. +delivered values, because one of them is a credential. The bucket and its entry +retention are always required. The two warehouse settings are optional, but only together: both +delivered starts the mirror, neither starts the witness with no mirror, and one +without the other is refused as a half-provisioned group. The native composition +cannot yet start this service with its empty group. Local execution must use the same signing and verification protocol with durable custody independent of Accounts; missing native storage and bootstrap are owner tooling gaps, not a reason to create a host-private log or fake receipts. @@ -292,6 +323,15 @@ qualification before the transport can claim the full deadline requirement. ## Warehouse mirror +**The mirror is optional** (owner decision 2026-10-18, reversing the 2026-10-17 +call that the witness stays down until a warehouse front exists). The object +store is the authority — every receipt is signed, chained and committed there +before it is returned — so a witness deployed with no warehouse configured +witnesses fully; what it lacks is the projection. It wires `mirror.None`, a +named value rather than a nil, and prints one line at boot naming +`POLICY_LOG_WAREHOUSE_ENDPOINT` and `POLICY_LOG_WAREHOUSE_TOKEN`, so the pod log +says why the warehouse is empty. A warehouse added later backfills from `Read`. + The [warehouse writer](code/mirror/warehouse.go) follows the repository's JSON POST ingestion convention with a dedicated credential. The service offers an entry only after the head commits. One worker consumes a bounded 64-entry diff --git a/module/services/policy-log/code/main.go b/module/services/policy-log/code/main.go index aa3335c2c..d45c0ca3e 100644 --- a/module/services/policy-log/code/main.go +++ b/module/services/policy-log/code/main.go @@ -1,7 +1,9 @@ // policy-log is the independent signed witness for authority changes. This // binary wires the delivered signing-key secret, the object store, the warehouse // mirror and the transport-independent service onto the gRPC endpoint Codefly -// allocated, and refuses to start if any one of them is not provisioned. +// allocated, and refuses to start if the key or the store is not provisioned. +// The mirror is the one optional part: with neither of its settings delivered +// the witness runs without one and says so at boot (owner decision 2026-10-18). // // Nothing here may start a listener it cannot witness with, so every // construction happens BEFORE net.Listen and any failure is fatal. A witness @@ -20,6 +22,7 @@ import ( "strconv" "strings" "syscall" + "time" "policy-log/identity" "policy-log/mirror" @@ -42,13 +45,22 @@ const grpcEndpointName = "grpc" // provisions for this service. It is the ONLY source of these settings. const policyLogConfiguration = "policy-log" -var settingKeys = struct{ bucket, warehouseEndpoint, warehouseToken string }{ +var settingKeys = struct{ bucket, retentionDays, warehouseEndpoint, warehouseToken string }{ bucket: "POLICY_LOG_BUCKET", + retentionDays: "POLICY_LOG_RETENTION_DAYS", warehouseEndpoint: "POLICY_LOG_WAREHOUSE_ENDPOINT", warehouseToken: "POLICY_LOG_WAREHOUSE_TOKEN", } -type settings struct{ bucket, warehouseEndpoint, warehouseToken string } +type settings struct { + bucket string + retention time.Duration + warehouseEndpoint, warehouseToken string +} + +// mirrored reports whether a warehouse mirror was configured. resolveSettings +// guarantees the endpoint and token are delivered together or not at all. +func (s settings) mirrored() bool { return s.warehouseEndpoint != "" } // resolveSettings reads the whole configuration from the delivered group and // from nowhere else. @@ -59,34 +71,97 @@ type settings struct{ bucket, warehouseEndpoint, warehouseToken string } // successfully, which is the one failure mode this service cannot have. Every // missing key is named at once, so a half-provisioned group takes one restart // to diagnose rather than three. -func resolveSettings(value func(group, key string) (string, error)) (settings, error) { - var resolved settings - var missing []string - for _, field := range []struct { - key string - into *string - }{ - {settingKeys.bucket, &resolved.bucket}, - {settingKeys.warehouseEndpoint, &resolved.warehouseEndpoint}, - {settingKeys.warehouseToken, &resolved.warehouseToken}, - } { - // A group that is absent and a key that is empty are the same thing to - // this service: nothing was provisioned. Neither value is reported back, - // because one of the three is a credential. - got, err := value(policyLogConfiguration, field.key) - if err != nil || got == "" { - missing = append(missing, field.key) - continue +// +// The bucket is the witness's authority and is always required. The warehouse +// mirror is a best-effort projection of what the bucket already committed +// (README, "Warehouse mirror"), so it is optional — but only WHOLE: its endpoint +// and its token are delivered together or not at all. One without the other is +// a half-provisioned mirror, and it is refused rather than read as "no mirror", +// because that would silently drop a projection someone meant to configure. +func resolveSettings(value func(group, key string) (string, bool, error)) (settings, error) { + // A value no namespace carries and an empty value are the same thing to this + // service: nothing was provisioned. A value that is there and could not be + // READ is not: it is refused by name, never read as absent — otherwise a + // broken delivery of both mirror settings would boot claiming no mirror was + // configured. No value is ever reported back, because one is a credential. + var unreadable []string + read := func(key string) string { + got, set, err := value(policyLogConfiguration, key) + if err != nil { + unreadable = append(unreadable, key) + return "" } - *field.into = got + if !set { + return "" + } + return got + } + retentionDays := read(settingKeys.retentionDays) + resolved := settings{ + bucket: read(settingKeys.bucket), + warehouseEndpoint: read(settingKeys.warehouseEndpoint), + warehouseToken: read(settingKeys.warehouseToken), + } + if len(unreadable) > 0 { + return settings{}, fmt.Errorf("configuration group %q delivered %s but it could not be read: this service refuses rather than guess what was provisioned", + policyLogConfiguration, strings.Join(unreadable, ", ")) + } + var missing []string + if resolved.bucket == "" { + missing = append(missing, settingKeys.bucket) + } + // The entry retention is the bucket's other half: how long every entry is + // Locked. Required, a whole number of days, and bounded — a Locked + // retention can never be shortened, so a malformed value is refused rather + // than rounded. + // Bounded in DAYS before it is multiplied: a large count would overflow + // time.Duration into a short, accepted retention. + if days, err := strconv.Atoi(retentionDays); err != nil || days <= 0 || + days > int(gcs.MaxRetention/(24*time.Hour)) || strconv.Itoa(days) != retentionDays { + missing = append(missing, settingKeys.retentionDays) + } else { + resolved.retention = time.Duration(days) * 24 * time.Hour + } + switch { + case resolved.warehouseEndpoint != "" && resolved.warehouseToken == "": + missing = append(missing, settingKeys.warehouseToken) + case resolved.warehouseEndpoint == "" && resolved.warehouseToken != "": + missing = append(missing, settingKeys.warehouseEndpoint) } if len(missing) > 0 { - return settings{}, fmt.Errorf("configuration group %q delivered no %s: composition must provision the whole group before this service can witness", + return settings{}, fmt.Errorf("configuration group %q delivered no usable %s: composition must provision the bucket and its entry retention, and the warehouse mirror whole or not at all, before this service can witness", policyLogConfiguration, strings.Join(missing, ", ")) } return resolved, nil } +// noMirrorWarning is the boot line a witness with no warehouse mirror prints, +// so a reader of the pod log learns what the deployment chose rather than +// inferring it from a warehouse that stays empty. +const noMirrorWarning = "policy-log: no warehouse mirror is configured: this witness WITNESSES normally — " + + "the object store is the authority and every receipt is signed, chained and committed there — " + + "but committed entries are not projected to the warehouse until POLICY_LOG_WAREHOUSE_ENDPOINT and " + + "POLICY_LOG_WAREHOUSE_TOKEN are delivered together; Read is the source to backfill it from" + +// startMirror builds the warehouse projection the settings ask for: the +// bounded best-effort worker when one is configured, or mirror.None, stated at +// boot, when none is. +func startMirror(ctx context.Context, resolved settings, warn func(string)) (service.Mirror, error) { + if !resolved.mirrored() { + warn(noMirrorWarning) + return mirror.None{}, nil + } + warehouse, err := mirror.NewWriter(resolved.warehouseEndpoint, resolved.warehouseToken) + if err != nil { + return nil, err + } + // A dropped or failed mirror is reported and repairable from Read. It never + // changes an append's outcome, so it is logged, not fatal. + return mirror.Start(ctx, warehouse, func(err error) { + log.Printf("policy-log: warehouse mirror: %v", err) + }) +} + // resolveSigningKey reads only the declared secret group. The public pin is // independently delivered to clients; this service cannot choose or replace it. func resolveSigningKey(secret func(string, string) (string, error)) (ed25519.PrivateKey, error) { @@ -102,9 +177,7 @@ func resolveSigningKey(secret func(string, string) (string, error)) (ed25519.Pri } func run(ctx context.Context, stop func()) error { - resolved, err := resolveSettings(func(group, key string) (string, error) { - return codefly.For(ctx).WorkspaceValue(group, key) - }) + resolved, err := resolveSettings(codefly.For(ctx).WorkspaceValueIfSet) if err != nil { return err } @@ -120,20 +193,12 @@ func run(ctx context.Context, stop func()) error { if err != nil { return err } - store, err := gcs.New(ctx, resolved.bucket) + store, err := gcs.New(ctx, resolved.bucket, resolved.retention) if err != nil { return err } defer func() { _ = store.Close() }() - warehouse, err := mirror.NewWriter(resolved.warehouseEndpoint, resolved.warehouseToken) - if err != nil { - return err - } - // A dropped or failed mirror is reported and repairable from Read. It never - // changes an append's outcome, so it is logged, not fatal. - worker, err := mirror.Start(ctx, warehouse, func(err error) { - log.Printf("policy-log: warehouse mirror: %v", err) - }) + worker, err := startMirror(ctx, resolved, func(line string) { log.Print(line) }) if err != nil { return err } diff --git a/module/services/policy-log/code/main_test.go b/module/services/policy-log/code/main_test.go index c224aba11..28b6a44ee 100644 --- a/module/services/policy-log/code/main_test.go +++ b/module/services/policy-log/code/main_test.go @@ -2,6 +2,7 @@ package main import ( "bytes" + "context" "crypto/ed25519" "crypto/x509" "encoding/base64" @@ -9,6 +10,9 @@ import ( "os" "strings" "testing" + "time" + + "policy-log/mirror" "gopkg.in/yaml.v3" ) @@ -17,33 +21,117 @@ import ( // cost one restart to diagnose, not one per key, and it must fail rather than // start a listener that cannot witness. func TestResolveSettingsNamesEveryMissingKeyAtOnce(t *testing.T) { - _, err := resolveSettings(func(string, string) (string, error) { return "", nil }) + _, err := resolveSettings(lookup(func(string, string) (string, error) { return "", nil })) if err == nil { t.Fatal("resolveSettings accepted a group that delivered nothing") } - for _, key := range []string{settingKeys.bucket, settingKeys.warehouseEndpoint, settingKeys.warehouseToken} { + for _, key := range []string{settingKeys.bucket, settingKeys.retentionDays} { + if !strings.Contains(err.Error(), key) { + t.Errorf("the refusal does not name the missing %s: %v", key, err) + } + } + // No bucket and a token with no endpoint: both defects in one refusal. + _, err = resolveSettings(lookup(func(_, key string) (string, error) { + if key == settingKeys.warehouseToken { + return "example-warehouse-token", nil + } + return "", nil + })) + if err == nil { + t.Fatal("resolveSettings accepted a group with no bucket and half a mirror") + } + for _, key := range []string{settingKeys.bucket, settingKeys.warehouseEndpoint} { if !strings.Contains(err.Error(), key) { t.Errorf("the refusal does not name the missing %s: %v", key, err) } } } +// TestABucketWithNoMirrorIsAStartableWitness: the owner's 2026-10-18 decision. +// The object store is the authority and the warehouse only mirrors it, so a +// group delivering the bucket and NEITHER mirror setting resolves, with no +// mirror — it is not read as a half-provisioned group. +func TestABucketWithNoMirrorIsAStartableWitness(t *testing.T) { + resolved, err := resolveSettings(lookup(func(_, key string) (string, error) { + switch key { + case settingKeys.bucket: + return "example-witness-bucket", nil + case settingKeys.retentionDays: + return "2557", nil + } + return "", errors.New("not delivered") + })) + if err != nil { + t.Fatalf("a bucket with no mirror was refused: %v", err) + } + if resolved.bucket != "example-witness-bucket" || resolved.mirrored() { + t.Fatalf("resolved %+v, want the bucket and no mirror", resolved) + } +} + +// TestNoMirrorIsStatedByNameAtBoot: a witness with no mirror wires mirror.None +// — a named value, never a nil a check could absorb — and says so in one line +// that names both settings it would need, so the pod log answers "why is the +// warehouse empty" without a code read. +func TestNoMirrorIsStatedByNameAtBoot(t *testing.T) { + var said []string + worker, err := startMirror(context.Background(), settings{bucket: "example-witness-bucket"}, + func(line string) { said = append(said, line) }) + if err != nil { + t.Fatalf("startMirror with no mirror: %v", err) + } + if _, ok := worker.(mirror.None); !ok { + t.Fatalf("startMirror wired %T, want mirror.None", worker) + } + if len(said) != 1 { + t.Fatalf("startMirror said %d lines, want exactly one: %q", len(said), said) + } + for _, want := range []string{"no warehouse mirror is configured", "object store is the authority", + settingKeys.warehouseEndpoint, settingKeys.warehouseToken} { + if !strings.Contains(said[0], want) { + t.Errorf("the boot line does not say %q: %s", want, said[0]) + } + } +} + +// TestAConfiguredMirrorIsBuiltAndStaysSilent: the other branch. A whole mirror +// starts the bounded worker and prints no no-mirror line; a mirror the writer +// refuses (not https) still refuses the boot, as before the decision. +func TestAConfiguredMirrorIsBuiltAndStaysSilent(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + silent := func(line string) { t.Fatalf("a configured mirror printed %q", line) } + worker, err := startMirror(ctx, settings{bucket: "b", warehouseEndpoint: "https://warehouse.example.com/ingest", + warehouseToken: "example-warehouse-token"}, silent) + if err != nil { + t.Fatalf("startMirror with a whole mirror: %v", err) + } + if _, ok := worker.(*mirror.Worker); !ok { + t.Fatalf("startMirror wired %T, want *mirror.Worker", worker) + } + if _, err := startMirror(ctx, settings{bucket: "b", warehouseEndpoint: "http://warehouse.example.com/ingest", + warehouseToken: "example-warehouse-token"}, silent); err == nil { + t.Fatal("startMirror accepted a mirror the writer refuses") + } +} + // TestResolveSettingsRefusesEachMissingKeyIndividually proves no single setting -// has a default. A bucket without a warehouse credential, or a warehouse -// without a bucket, is not a startable configuration. +// has a default. A bucket with half a mirror, or a mirror without a bucket, is +// not a startable configuration. func TestResolveSettingsRefusesEachMissingKeyIndividually(t *testing.T) { delivered := map[string]string{ settingKeys.bucket: "example-witness-bucket", + settingKeys.retentionDays: "2557", settingKeys.warehouseEndpoint: "https://warehouse.example.com/ingest", settingKeys.warehouseToken: "example-warehouse-token", } for withheld := range delivered { - _, err := resolveSettings(func(_, key string) (string, error) { + _, err := resolveSettings(lookup(func(_, key string) (string, error) { if key == withheld { return "", nil } return delivered[key], nil - }) + })) if err == nil { t.Errorf("resolveSettings started with %s missing", withheld) continue @@ -63,9 +151,9 @@ func TestResolveSettingsNeverFallsBackToTheProcessEnvironment(t *testing.T) { t.Setenv(settingKeys.warehouseEndpoint, "https://ambient.example.com/ingest") t.Setenv(settingKeys.warehouseToken, "ambient-token") // The group is absent, exactly as it is on a workstation. - resolved, err := resolveSettings(func(string, string) (string, error) { + resolved, err := resolveSettings(lookup(func(string, string) (string, error) { return "", errors.New("no configuration found for policy-log") - }) + })) if err == nil { t.Fatalf("resolveSettings fell back to the process environment and resolved %+v", resolved) } @@ -81,7 +169,7 @@ func TestResolveSettingsNeverFallsBackToTheProcessEnvironment(t *testing.T) { // has. The refusal names keys, never values. func TestResolveSettingsNeverEchoesADeliveredValue(t *testing.T) { const credential = "s3cr3t-warehouse-token" - _, err := resolveSettings(func(_, key string) (string, error) { + _, err := resolveSettings(lookup(func(_, key string) (string, error) { switch key { case settingKeys.warehouseToken: return credential, nil @@ -89,7 +177,7 @@ func TestResolveSettingsNeverEchoesADeliveredValue(t *testing.T) { return "example-witness-bucket", nil } return "", nil - }) + })) if err == nil { t.Fatal("resolveSettings accepted a group with no warehouse endpoint") } @@ -104,13 +192,15 @@ func TestResolveSettingsNeverEchoesADeliveredValue(t *testing.T) { // group resolves, and each value lands in its own field rather than being // transposed. func TestResolveSettingsAcceptsAWholeGroup(t *testing.T) { - resolved, err := resolveSettings(func(group, key string) (string, error) { + resolved, err := resolveSettings(lookup(func(group, key string) (string, error) { if group != policyLogConfiguration { t.Fatalf("resolveSettings read group %q, want %q", group, policyLogConfiguration) } switch key { case settingKeys.bucket: return "example-witness-bucket", nil + case settingKeys.retentionDays: + return "2557", nil case settingKeys.warehouseEndpoint: return "https://warehouse.example.com/ingest", nil case settingKeys.warehouseToken: @@ -118,11 +208,11 @@ func TestResolveSettingsAcceptsAWholeGroup(t *testing.T) { } t.Fatalf("resolveSettings read an undeclared key %q", key) return "", nil - }) + })) if err != nil { t.Fatalf("resolveSettings: %v", err) } - if resolved.bucket != "example-witness-bucket" || + if resolved.bucket != "example-witness-bucket" || resolved.retention != 2557*24*time.Hour || resolved.warehouseEndpoint != "https://warehouse.example.com/ingest" || resolved.warehouseToken != "example-warehouse-token" { t.Fatalf("resolveSettings transposed a value: %+v", resolved) @@ -242,3 +332,65 @@ func TestInvalidDeliveredSigningSecretNamesItsSetting(t *testing.T) { } } } + +// lookup adapts a (value, error) answer to WorkspaceValueIfSet's three-way +// one: an error is "nothing carries this key" unless it is errUnreadable, +// which stands for a delivered value the carrier could not read. +func lookup(value func(group, key string) (string, error)) func(group, key string) (string, bool, error) { + return func(group, key string) (string, bool, error) { + got, err := value(group, key) + if errors.Is(err, errUnreadable) { + return "", false, err + } + if err != nil || got == "" { + return "", false, nil + } + return got, true, nil + } +} + +var errUnreadable = errors.New("delivered but unreadable") + +// TestAnUnreadableMirrorIsRefusedNotReadAsAbsent: both mirror settings +// delivered and neither readable (a broken file carrier) must refuse the boot +// by name. Read as absent, the witness would start claiming no mirror was +// configured — a provisioned projection gone dark with the wrong cause logged. +func TestAnUnreadableMirrorIsRefusedNotReadAsAbsent(t *testing.T) { + _, err := resolveSettings(lookup(func(_, key string) (string, error) { + switch key { + case settingKeys.bucket: + return "example-witness-bucket", nil + case settingKeys.retentionDays: + return "2557", nil + } + return "", errUnreadable + })) + if err == nil { + t.Fatal("resolveSettings started a witness whose mirror settings could not be read") + } + for _, key := range []string{settingKeys.warehouseEndpoint, settingKeys.warehouseToken} { + if !strings.Contains(err.Error(), key) || !strings.Contains(err.Error(), "could not be read") { + t.Errorf("the refusal does not name the unreadable %s: %v", key, err) + } + } +} + +// TestTheEntryRetentionIsAWholeBoundedNumberOfDays: a Locked retention can +// never be shortened, so anything but a plain positive day count within the +// adapter's bound is refused by name rather than rounded or clamped. +func TestTheEntryRetentionIsAWholeBoundedNumberOfDays(t *testing.T) { + for _, value := range []string{"0", "-1", "7y", "2557.5", " 2557", "02557", "4000", "213504", "9223372036854775807"} { + _, err := resolveSettings(lookup(func(_, key string) (string, error) { + switch key { + case settingKeys.bucket: + return "example-witness-bucket", nil + case settingKeys.retentionDays: + return value, nil + } + return "", nil + })) + if err == nil || !strings.Contains(err.Error(), settingKeys.retentionDays) { + t.Errorf("retention %q was accepted or not named: %v", value, err) + } + } +} diff --git a/module/services/policy-log/code/mirror/warehouse.go b/module/services/policy-log/code/mirror/warehouse.go index edc9cb54e..98c3c61ce 100644 --- a/module/services/policy-log/code/mirror/warehouse.go +++ b/module/services/policy-log/code/mirror/warehouse.go @@ -59,6 +59,15 @@ func (w *Writer) Write(ctx context.Context, record witness.Record) error { return nil } +// None is the mirror of a witness deployed with no warehouse configured. It +// projects nothing: the committed receipt in the object store is the evidence, +// and Read is where a warehouse added later backfills from. It is a named value +// rather than a nil Mirror so that "no mirror" is a decision the wiring states, +// never an accident a nil check absorbs. +type None struct{} + +func (None) Offer(witness.Record) {} + type Sink interface { Write(context.Context, witness.Record) error } diff --git a/module/services/policy-log/code/objectstore/gcs/gcs.go b/module/services/policy-log/code/objectstore/gcs/gcs.go index 1c56f42a0..37398bd2f 100644 --- a/module/services/policy-log/code/objectstore/gcs/gcs.go +++ b/module/services/policy-log/code/objectstore/gcs/gcs.go @@ -9,6 +9,7 @@ import ( "io" "os" "regexp" + "time" "policy-log/objectstore" @@ -21,17 +22,36 @@ var entryName = regexp.MustCompile(`^entries/[0-9]{20}\.json$`) // GCS owns a bucket in the witness trust domain, never a host database or a // host-controlled bucket. It performs one attempt per write, even for 5xx. type GCS struct { - client *storage.Client - bucket *storage.BucketHandle + client *storage.Client + bucket *storage.BucketHandle + retention time.Duration + now func() time.Time } +// MaxRetention bounds the delivered entry retention. A Locked retention cannot +// be shortened or removed by anyone, so a typo of an extra digit would hold the +// evidence — and the bucket — for a century. Ten years covers every declared +// audit horizon with room to spare. +const MaxRetention = 10 * 366 * 24 * time.Hour + // New uses the service's workload identity and the official GCS endpoint. There // is no endpoint or emulator fallback in production. Bucket identity is delivery // data; IAM and the service's own workload identity enforce its trust domain. -func New(ctx context.Context, bucket string) (*GCS, error) { +// +// retention is how long every entry is held under a LOCKED object retention: +// once written, nobody — not this service, not a project owner — can delete or +// replace it before then. That is the "retention" half of the immutable +// prefix; the bucket's IAM, which lets this identity create entries and never +// delete them, is the other. A bucket without object retention enabled refuses +// the write, so a store that cannot hold the lock refuses every append rather +// than accepting evidence it cannot protect. +func New(ctx context.Context, bucket string, retention time.Duration) (*GCS, error) { if bucket == "" { return nil, errors.New("policy-log: GCS bucket is required") } + if retention <= 0 || retention > MaxRetention { + return nil, fmt.Errorf("policy-log: entry retention must be positive and at most %s", MaxRetention) + } if os.Getenv("STORAGE_EMULATOR_HOST") != "" { return nil, errors.New("policy-log: storage emulator is forbidden") } @@ -39,11 +59,12 @@ func New(ctx context.Context, bucket string) (*GCS, error) { if err != nil { return nil, err } - return newGCS(client, bucket), nil + return newGCS(client, bucket, retention), nil } -func newGCS(client *storage.Client, bucket string) *GCS { - return &GCS{client: client, bucket: client.Bucket(bucket).Retryer(storage.WithPolicy(storage.RetryNever))} +func newGCS(client *storage.Client, bucket string, retention time.Duration) *GCS { + return &GCS{client: client, bucket: client.Bucket(bucket).Retryer(storage.WithPolicy(storage.RetryNever)), + retention: retention, now: time.Now} } func (s *GCS) Close() error { return s.client.Close() } @@ -99,6 +120,11 @@ func (s *GCS) Write(ctx context.Context, name string, data []byte, generation in writer.CacheControl = "no-store" writer.CRC32C = crc32.Checksum(data, crc32.MakeTable(crc32.Castagnoli)) writer.SendCRC32C = true + if name != objectstore.HeadObject { + // The head is the one object rewritten on every append; an entry is never + // rewritten, so it is locked for the declared horizon as it is created. + writer.Retention = &storage.ObjectRetention{Mode: "Locked", RetainUntil: s.now().Add(s.retention)} + } if _, err := writer.Write(data); err != nil { _ = writer.CloseWithError(err) return 0, classify(err) diff --git a/module/services/policy-log/code/objectstore/gcs/gcs_test.go b/module/services/policy-log/code/objectstore/gcs/gcs_test.go index 1d48fa7c8..5b5d4e28b 100644 --- a/module/services/policy-log/code/objectstore/gcs/gcs_test.go +++ b/module/services/policy-log/code/objectstore/gcs/gcs_test.go @@ -34,7 +34,9 @@ func fakeGCS(t *testing.T, transport transportFunc) *GCS { t.Fatal(err) } t.Cleanup(func() { _ = client.Close() }) - return newGCS(client, "example-policy-log") + s := newGCS(client, "example-policy-log", 2557*24*time.Hour) + s.now = func() time.Time { return time.Date(2026, 10, 18, 0, 0, 0, 0, time.UTC) } + return s } func TestGCSWritesUseExactGenerationPreconditions(t *testing.T) { @@ -161,7 +163,46 @@ func TestGCSRejectsMissingOversizedAndGenerationlessObjects(t *testing.T) { func TestGCSBootRefusesEmulatorOverride(t *testing.T) { t.Setenv("STORAGE_EMULATOR_HOST", "http://example.invalid") - if s, err := New(context.Background(), "example-policy-log"); s != nil || err == nil { + if s, err := New(context.Background(), "example-policy-log", 24*time.Hour); s != nil || err == nil { t.Fatal("emulator override accepted") } } + +// TestEntriesAreLockedForTheDeclaredRetentionAndTheHeadIsNot: the immutable +// prefix's retention half. Every entry is created under a LOCKED object +// retention for the declared horizon; the head, rewritten on every append, +// carries none — a locked head could never advance. +func TestEntriesAreLockedForTheDeclaredRetentionAndTheHeadIsNot(t *testing.T) { + for _, tc := range []struct { + name string + generation int64 + locked bool + }{{"entries/00000000000000000001.json", 0, true}, {objectstore.HeadObject, 41, false}} { + t.Run(tc.name, func(t *testing.T) { + var body string + s := fakeGCS(t, func(r *http.Request) (*http.Response, error) { + b, _ := io.ReadAll(r.Body) + body = string(b) + return response(200, `{"generation":"42"}`), nil + }) + if _, err := s.Write(context.Background(), tc.name, []byte(`{"seq":1}`), tc.generation); err != nil { + t.Fatal(err) + } + want := `"retention":{"mode":"Locked","retainUntilTime":"2033-10-18T00:00:00Z"}` + if got := strings.Contains(body, want); got != tc.locked { + t.Fatalf("locked=%v, want %v: %s", got, tc.locked, body) + } + if !tc.locked && strings.Contains(body, `"retention"`) { + t.Fatalf("the head carried a retention: %s", body) + } + }) + } +} + +func TestTheRetentionIsRequiredAndBounded(t *testing.T) { + for _, retention := range []time.Duration{0, -time.Hour, MaxRetention + time.Hour} { + if s, err := New(context.Background(), "example-policy-log", retention); s != nil || err == nil { + t.Errorf("retention %s accepted", retention) + } + } +}