amadan
Fix the attestation scope and wire it into the agent behind a switch that is off
The scope (Astra F6, F7; Opus N7). Every route whose path equals or encloses a configured database is excluded, not the first one found: a hibernating route's database can sit inside a unit's tree, and two routes can record one file. Paths are matched as written and as resolved through symlinks, because a unit tenant can link a file in its own state directory to another tenant's database. A running unit with no config file attests nothing. The reader. It requires the unit's command to be the replicator on that one config. It resolves every configured path. It reads the unit and the config's modification time before and after, with the registry read inside that bracket, so the routes and the replicator's state are one observation or none. The call. With CARLOS_DURABILITY_ATTEST=scoped in the agent's environment, every publish tick observes the box and applies the scope before it seals anything. A new exclusion is dated no later than the config's writing and the last tick that verified the old claim, so a day cannot close complete for an account the replicator began touching just before midnight. When the premise fails the claim ends at the last tick that verified it. The switch is unset everywhere, and without it the ledger attests nothing, as before. The call and the reader have had no independent review.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
7 files changed,
+652
−85
internal/host/agent.go+6 −0internal/host/durability.go+9 −0internal/host/durabilityreplicator.go+135 −30internal/host/durabilityreplicator_test.go+129 −30internal/host/durabilityscope.go+151 −22internal/host/durabilityscope_test.go+218 −3internal/ops/litestream.go+4 −0
diff --git a/internal/host/agent.go b/internal/host/agent.go| index 6b1bebf83..cad1267f8 100644 |
| --- a/internal/host/agent.go |
| +++ b/internal/host/agent.go |
| @@ -253,6 +253,12 @@ func RunAgent(args []string) { |
| if durabilityErr != nil { |
| log.Printf("agent: durability: %v", durabilityErr) |
| } |
| + // The scoped replication attestation is off unless this box's |
| + // environment asks for it by name. Unset, the ledger attests nothing. |
| + if scopedAttestationEnabled(os.Getenv) { |
| + durabilityPub.enableScopedAttestation(newReplicatorProbe(), reg.List, dd) |
| + log.Printf("agent: durability: scoped attestation enabled (%s=scoped)", durabilityAttestEnv) |
| + } |
| go durabilityPub.loop() |
| // Fleet box identity (customer-fleets plan, Task 6 fix round — controller |
diff --git a/internal/host/durability.go b/internal/host/durability.go| index 8d3ca3657..90355bb53 100644 |
| --- a/internal/host/durability.go |
| +++ b/internal/host/durability.go |
| @@ -42,6 +42,11 @@ type durabilityPublisher struct { |
| stopped atomic.Bool |
| bindingFailed atomic.Bool |
| registered bool |
| + // attest, when set (enableScopedAttestation), observes the box and |
| + // applies the scoped attestation. It runs before every seal. Nil is |
| + // the default and the production state: nothing is attested. |
| + attest func(context.Context) |
| + scopeVerifiedAt time.Time |
| } |
| func newDurabilityPublisher(dd, box string, now func() time.Time) (*durabilityPublisher, error) { |
| @@ -151,6 +156,10 @@ func (p *durabilityPublisher) publish(ctx context.Context, store release.Store) |
| if p.stopped.Load() { |
| return errors.New("durability_publisher_stopped") |
| } |
| + // The claim is checked before anything is sealed under it. |
| + if p.attest != nil { |
| + p.attest(ctx) |
| + } |
| if err := p.ledger.Seal(); err != nil { |
| return err |
| } |
diff --git a/internal/host/durabilityreplicator.go b/internal/host/durabilityreplicator.go| index 423c026fd..456079619 100644 |
| --- a/internal/host/durabilityreplicator.go |
| +++ b/internal/host/durabilityreplicator.go |
| @@ -4,8 +4,11 @@ import ( |
| "context" |
| "errors" |
| "fmt" |
| + "github.com/carlosframework/platform/internal/registry" |
| + "github.com/carlosframework/platform/internal/usage" |
| "os" |
| "os/exec" |
| + "path/filepath" |
| "strconv" |
| "strings" |
| "time" |
| @@ -29,8 +32,11 @@ type replicatorProbe struct { |
| unit string |
| readFile func(string) ([]byte, error) |
| modTime func(string) (time.Time, error) |
| - // unitState reports whether the unit is active and since when. |
| - unitState func(ctx context.Context, unit string) (active bool, since time.Time, err error) |
| + // unitState reports whether the unit is active and since when, and |
| + // the command it runs. |
| + unitState func(ctx context.Context, unit string) (active bool, since time.Time, argv []string, err error) |
| + // resolve follows symlinks. |
| + resolve func(string) (string, error) |
| } |
| func newReplicatorProbe() replicatorProbe { |
| @@ -46,25 +52,41 @@ func newReplicatorProbe() replicatorProbe { |
| return fi.ModTime(), nil |
| }, |
| unitState: systemdUnitState, |
| + resolve: filepath.EvalSymlinks, |
| } |
| } |
| // read observes the replicator. Anything it cannot establish is an Err, and |
| // an Err attests nothing (durabilityScope). |
| // |
| -// The unit is read before and after the config. If it changed in between |
| -// (a restart landed while this was reading), the config read may belong to |
| -// neither state, so nothing is concluded this time. |
| +// The unit and the config's modification time are read before and after |
| +// the config itself. If either moved (a restart or a rewrite landed while |
| +// this was reading), what was read may belong to neither state, so nothing |
| +// is concluded this time. between, when not nil, runs inside that bracket: |
| +// the caller reads the registry there, so the routes it pairs with this |
| +// state are from the same moment (readDurabilityScope). |
| // |
| // A config that is not, line for line, what the generator writes is an |
| // Err: this code will not speak for a hand-kept config, because it cannot |
| // know it has seen every database in it. A box whose config is still kept |
| // by hand must be cut over to the generated one before it can attest. |
| -func (p replicatorProbe) read(ctx context.Context) replicatorState { |
| - active, since, err := p.unitState(ctx, p.unit) |
| +// |
| +// A running unit must be running the command this code expects, the |
| +// replicator reading that one config file and nothing else. A unit given |
| +// databases on its command line, or another config, replicates what this |
| +// code cannot see. |
| +// |
| +// Every configured path is resolved through symlinks, and both the path as |
| +// written and its target are reported. A path that cannot be resolved is |
| +// an Err. |
| +func (p replicatorProbe) read(ctx context.Context, between func() error) replicatorState { |
| + active, since, argv, err := p.unitState(ctx, p.unit) |
| if err != nil { |
| return replicatorState{Err: fmt.Errorf("unit %s: %w", p.unit, err)} |
| } |
| + if active && !expectedReplicatorCommand(argv, p.configPath) { |
| + return replicatorState{Err: fmt.Errorf("unit %s runs %q, not the replicator on %s alone", p.unit, strings.Join(argv, " "), p.configPath)} |
| + } |
| out := replicatorState{UnitActive: active, UnitStarted: since} |
| raw, err := p.readFile(p.configPath) |
| switch { |
| @@ -75,6 +97,18 @@ func (p replicatorProbe) read(ctx context.Context) replicatorState { |
| if active { |
| return replicatorState{Err: fmt.Errorf("unit %s is running and %s does not exist", p.unit, p.configPath)} |
| } |
| + if between != nil { |
| + if err := between(); err != nil { |
| + return replicatorState{Err: err} |
| + } |
| + } |
| + active2, _, _, err := p.unitState(ctx, p.unit) |
| + if err != nil || active2 { |
| + return replicatorState{Err: fmt.Errorf("unit %s changed state while it was being read", p.unit)} |
| + } |
| + if _, err := p.readFile(p.configPath); !errors.Is(err, os.ErrNotExist) { |
| + return replicatorState{Err: fmt.Errorf("%s appeared while it was being read", p.configPath)} |
| + } |
| return out |
| case err != nil: |
| return replicatorState{Err: fmt.Errorf("read %s: %w", p.configPath, err)} |
| @@ -83,68 +117,139 @@ func (p replicatorProbe) read(ctx context.Context) replicatorState { |
| if err != nil { |
| return replicatorState{Err: fmt.Errorf("stat %s: %w", p.configPath, err)} |
| } |
| - paths, err := ops.GeneratedLitestreamDBs(string(raw)) |
| + written, err := ops.GeneratedLitestreamDBs(string(raw)) |
| if err != nil { |
| return replicatorState{Err: fmt.Errorf("%s: %w", p.configPath, err)} |
| } |
| - // The config again, and the unit again: both must be what they were. |
| + paths := make([]string, 0, len(written)) |
| + for _, path := range written { |
| + paths = append(paths, path) |
| + real, err := p.resolve(path) |
| + switch { |
| + case errors.Is(err, os.ErrNotExist): |
| + // Named and not there: nothing is streamed from it now, and |
| + // the path as written still names its owner. |
| + case err != nil: |
| + return replicatorState{Err: fmt.Errorf("resolve %s: %w", path, err)} |
| + case filepath.Clean(real) != filepath.Clean(path): |
| + paths = append(paths, filepath.Clean(real)) |
| + } |
| + } |
| + if between != nil { |
| + if err := between(); err != nil { |
| + return replicatorState{Err: err} |
| + } |
| + } |
| + // The config, its modification time and the unit again: all must be |
| + // what they were. |
| again, err := p.readFile(p.configPath) |
| if err != nil || string(again) != string(raw) { |
| return replicatorState{Err: fmt.Errorf("%s changed while it was being read", p.configPath)} |
| } |
| - active2, since2, err := p.unitState(ctx, p.unit) |
| + modified2, err := p.modTime(p.configPath) |
| + if err != nil || !modified2.Equal(modified) { |
| + return replicatorState{Err: fmt.Errorf("%s was rewritten while it was being read", p.configPath)} |
| + } |
| + active2, since2, argv2, err := p.unitState(ctx, p.unit) |
| if err != nil { |
| return replicatorState{Err: fmt.Errorf("unit %s: %w", p.unit, err)} |
| } |
| - if active2 != active || !since2.Equal(since) { |
| + if active2 != active || !since2.Equal(since) || strings.Join(argv2, "\x00") != strings.Join(argv, "\x00") { |
| return replicatorState{Err: fmt.Errorf("unit %s changed state while its config was being read", p.unit)} |
| } |
| out.ConfigPresent, out.Paths, out.ConfigModified = true, paths, modified |
| return out |
| } |
| -// systemdUnitState asks systemd whether unit is active and when it became |
| -// so. The timestamp is requested in unix form: the default rendering |
| -// carries a zone abbreviation, which cannot be parsed back reliably. |
| -func systemdUnitState(ctx context.Context, unit string) (bool, time.Time, error) { |
| - out, err := exec.CommandContext(ctx, "systemctl", "show", unit, "--timestamp=unix", "-p", "LoadState", "-p", "ActiveState", "-p", "ActiveEnterTimestamp").Output() |
| +// expectedReplicatorCommand reports whether argv is the box-wide |
| +// replicator as bootstrap installs it: the litestream binary, replicate, |
| +// and this config file, with no database or replica arguments after it. |
| +func expectedReplicatorCommand(argv []string, config string) bool { |
| + return len(argv) == 4 && filepath.Base(argv[0]) == "litestream" && argv[1] == "replicate" && argv[2] == "-config" && argv[3] == config |
| +} |
| + |
| +// readDurabilityScope reads the routes and the replicator's state as one |
| +// observation and computes the scope from them. The registry is read |
| +// inside the probe's bracket: if the unit restarted or the config was |
| +// rewritten around that read, nothing is attested. |
| +func readDurabilityScope(ctx context.Context, p replicatorProbe, listRoutes func() ([]registry.Route, error), dataDir string) (usage.DurabilityAttestation, replicatorState, string) { |
| + var routes []registry.Route |
| + rs := p.read(ctx, func() error { |
| + var err error |
| + routes, err = listRoutes() |
| + if err != nil { |
| + return fmt.Errorf("list routes: %w", err) |
| + } |
| + return nil |
| + }) |
| + scope, reason := durabilityScope(routes, dataDir, rs, p.resolve) |
| + return scope, rs, reason |
| +} |
| + |
| +// systemdUnitState asks systemd whether unit is active, when it became so, |
| +// and what it runs. The timestamp is requested in unix form: the default |
| +// rendering carries a zone abbreviation, which cannot be parsed back |
| +// reliably. |
| +func systemdUnitState(ctx context.Context, unit string) (bool, time.Time, []string, error) { |
| + out, err := exec.CommandContext(ctx, "systemctl", "show", unit, "--timestamp=unix", "-p", "LoadState", "-p", "ActiveState", "-p", "ActiveEnterTimestamp", "-p", "ExecStart").Output() |
| if err != nil { |
| - return false, time.Time{}, err |
| + return false, time.Time{}, nil, err |
| } |
| return parseSystemdUnitState(string(out)) |
| } |
| // parseSystemdUnitState reads `systemctl show --timestamp=unix` output for |
| -// LoadState, ActiveState and ActiveEnterTimestamp. Only two answers are |
| -// accepted: a unit that is active with a start time, and a unit that is |
| -// plainly not running (inactive or failed, or not installed at all). Any |
| -// state in between (activating, deactivating, reloading), a missing |
| -// property or a timestamp in another form is an error. |
| -func parseSystemdUnitState(out string) (bool, time.Time, error) { |
| +// LoadState, ActiveState, ActiveEnterTimestamp and ExecStart. Only two |
| +// answers are accepted: a unit that is active with a start time and one |
| +// command, and a unit that is plainly not running (inactive or failed, or |
| +// not installed at all). Any state in between (activating, deactivating, |
| +// reloading), a missing property, a timestamp in another form, or a unit |
| +// with no command or several is an error. |
| +func parseSystemdUnitState(out string) (bool, time.Time, []string, error) { |
| props := map[string]string{} |
| + execStarts := 0 |
| for _, line := range strings.Split(strings.TrimSpace(out), "\n") { |
| if k, v, ok := strings.Cut(line, "="); ok { |
| + if k == "ExecStart" { |
| + execStarts++ |
| + } |
| props[k] = strings.TrimSpace(v) |
| } |
| } |
| load, state := props["LoadState"], props["ActiveState"] |
| switch { |
| case load == "not-found" && state == "inactive": |
| - return false, time.Time{}, nil |
| + return false, time.Time{}, nil, nil |
| case load != "loaded": |
| - return false, time.Time{}, fmt.Errorf("load state %q", load) |
| + return false, time.Time{}, nil, fmt.Errorf("load state %q", load) |
| case state == "inactive", state == "failed": |
| - return false, time.Time{}, nil |
| + return false, time.Time{}, nil, nil |
| case state != "active": |
| - return false, time.Time{}, fmt.Errorf("active state %q", state) |
| + return false, time.Time{}, nil, fmt.Errorf("active state %q", state) |
| } |
| stamp, ok := strings.CutPrefix(props["ActiveEnterTimestamp"], "@") |
| if !ok { |
| - return false, time.Time{}, fmt.Errorf("start time %q is not in unix form", props["ActiveEnterTimestamp"]) |
| + return false, time.Time{}, nil, fmt.Errorf("start time %q is not in unix form", props["ActiveEnterTimestamp"]) |
| } |
| secs, err := strconv.ParseInt(stamp, 10, 64) |
| if err != nil || secs <= 0 { |
| - return false, time.Time{}, fmt.Errorf("start time %q", props["ActiveEnterTimestamp"]) |
| + return false, time.Time{}, nil, fmt.Errorf("start time %q", props["ActiveEnterTimestamp"]) |
| + } |
| + // ExecStart={ path=/usr/local/bin/litestream ; argv[]=/usr/local/bin/litestream replicate -config /etc/litestream.yml ; ignore_errors=no ; ... } |
| + if execStarts != 1 { |
| + return false, time.Time{}, nil, fmt.Errorf("%d ExecStart commands", execStarts) |
| + } |
| + _, rest, ok := strings.Cut(props["ExecStart"], "argv[]=") |
| + if !ok { |
| + return false, time.Time{}, nil, fmt.Errorf("ExecStart %q has no argv", props["ExecStart"]) |
| + } |
| + args, _, ok := strings.Cut(rest, " ;") |
| + if !ok { |
| + return false, time.Time{}, nil, fmt.Errorf("ExecStart %q is not in the expected form", props["ExecStart"]) |
| + } |
| + argv := strings.Fields(args) |
| + if len(argv) == 0 { |
| + return false, time.Time{}, nil, fmt.Errorf("ExecStart %q has an empty argv", props["ExecStart"]) |
| } |
| - return true, time.Unix(secs, 0).UTC(), nil |
| + return true, time.Unix(secs, 0).UTC(), argv, nil |
| } |
diff --git a/internal/host/durabilityreplicator_test.go b/internal/host/durabilityreplicator_test.go| index 068ce8731..81fd3c0c6 100644 |
| --- a/internal/host/durabilityreplicator_test.go |
| +++ b/internal/host/durabilityreplicator_test.go |
| @@ -9,8 +9,11 @@ import ( |
| "time" |
| "github.com/carlosframework/platform/internal/ops" |
| + "github.com/carlosframework/platform/internal/registry" |
| ) |
| +const unitExec = "ExecStart={ path=/usr/local/bin/litestream ; argv[]=/usr/local/bin/litestream replicate -config /etc/litestream.yml ; ignore_errors=no ; start_time=[n/a] ; stop_time=[n/a] ; pid=0 ; code=(null) ; status=0/0 }\n" |
| + |
| func TestParseSystemdUnitState(t *testing.T) { |
| cases := []struct { |
| name string |
| @@ -19,29 +22,46 @@ func TestParseSystemdUnitState(t *testing.T) { |
| since int64 |
| fails bool |
| }{ |
| - {"running", "LoadState=loaded\nActiveState=active\nActiveEnterTimestamp=@1790817621\n", true, 1790817621, false}, |
| - {"stopped", "LoadState=loaded\nActiveState=inactive\nActiveEnterTimestamp=\n", false, 0, false}, |
| - {"failed", "LoadState=loaded\nActiveState=failed\nActiveEnterTimestamp=@1790817621\n", false, 0, false}, |
| + {"running", "LoadState=loaded\nActiveState=active\nActiveEnterTimestamp=@1790817621\n" + unitExec, true, 1790817621, false}, |
| + {"stopped", "LoadState=loaded\nActiveState=inactive\nActiveEnterTimestamp=\n" + unitExec, false, 0, false}, |
| + {"failed", "LoadState=loaded\nActiveState=failed\nActiveEnterTimestamp=@1790817621\n" + unitExec, false, 0, false}, |
| {"not installed", "LoadState=not-found\nActiveState=inactive\nActiveEnterTimestamp=\n", false, 0, false}, |
| - {"starting: neither running nor stopped", "LoadState=loaded\nActiveState=activating\nActiveEnterTimestamp=\n", false, 0, true}, |
| - {"stopping", "LoadState=loaded\nActiveState=deactivating\nActiveEnterTimestamp=@1790817621\n", false, 0, true}, |
| - {"a start time with a zone abbreviation", "LoadState=loaded\nActiveState=active\nActiveEnterTimestamp=Thu 2026-10-01 01:40:21 IST\n", false, 0, true}, |
| - {"running with no start time", "LoadState=loaded\nActiveState=active\nActiveEnterTimestamp=\n", false, 0, true}, |
| + {"starting: neither running nor stopped", "LoadState=loaded\nActiveState=activating\nActiveEnterTimestamp=\n" + unitExec, false, 0, true}, |
| + {"stopping", "LoadState=loaded\nActiveState=deactivating\nActiveEnterTimestamp=@1790817621\n" + unitExec, false, 0, true}, |
| + {"a start time with a zone abbreviation", "LoadState=loaded\nActiveState=active\nActiveEnterTimestamp=Thu 2026-10-01 01:40:21 IST\n" + unitExec, false, 0, true}, |
| + {"running with no start time", "LoadState=loaded\nActiveState=active\nActiveEnterTimestamp=\n" + unitExec, false, 0, true}, |
| + {"running with no command", "LoadState=loaded\nActiveState=active\nActiveEnterTimestamp=@1790817621\n", false, 0, true}, |
| + {"running with two commands", "LoadState=loaded\nActiveState=active\nActiveEnterTimestamp=@1790817621\n" + unitExec + unitExec, false, 0, true}, |
| {"masked", "LoadState=masked\nActiveState=inactive\nActiveEnterTimestamp=\n", false, 0, true}, |
| {"nothing at all", "", false, 0, true}, |
| } |
| for _, c := range cases { |
| - active, since, err := parseSystemdUnitState(c.out) |
| + active, since, argv, err := parseSystemdUnitState(c.out) |
| if (err != nil) != c.fails || active != c.active || (c.since != 0 && since.Unix() != c.since) { |
| t.Errorf("%s: active=%v since=%v err=%v", c.name, active, since, err) |
| } |
| + if active && !expectedReplicatorCommand(argv, "/etc/litestream.yml") { |
| + t.Errorf("%s: argv %q is not recognised as the replicator", c.name, argv) |
| + } |
| + } |
| + for name, argv := range map[string][]string{ |
| + "a database on the command line": {"/usr/local/bin/litestream", "replicate", "/data/x.db", "s3://bucket/x"}, |
| + "another config": {"/usr/local/bin/litestream", "replicate", "-config", "/etc/other.yml"}, |
| + "an extra argument": {"/usr/local/bin/litestream", "replicate", "-config", "/etc/litestream.yml", "-exec", "x"}, |
| + "another binary": {"/usr/local/bin/rsync", "replicate", "-config", "/etc/litestream.yml"}, |
| + "nothing": nil, |
| + } { |
| + if expectedReplicatorCommand(argv, "/etc/litestream.yml") { |
| + t.Errorf("%s: accepted as the replicator", name) |
| + } |
| } |
| } |
| func TestReplicatorProbe(t *testing.T) { |
| ctx := context.Background() |
| written := time.Date(2026, 9, 17, 8, 0, 0, 0, time.UTC) |
| - generated := ops.LitestreamConfig([]string{"/data/registry.db", "/data/api.example.com.db"}, "bucket", "prefix", "eu-west-1") |
| + argv := []string{"/usr/local/bin/litestream", "replicate", "-config", "/etc/litestream.yml"} |
| + generated := ops.LitestreamConfig([]string{"/data/registry.db", "/data/api.example.com.db", "/var/lib/carlos-app/kass.example.com/instances/y.db"}, "bucket", "prefix", "eu-west-1") |
| probe := func(cfg *string, active bool, since time.Time) replicatorProbe { |
| return replicatorProbe{configPath: "/etc/litestream.yml", unit: externalReplicatorUnit, |
| readFile: func(string) ([]byte, error) { |
| @@ -50,40 +70,69 @@ func TestReplicatorProbe(t *testing.T) { |
| } |
| return []byte(*cfg), nil |
| }, |
| - modTime: func(string) (time.Time, error) { return written, nil }, |
| - unitState: func(context.Context, string) (bool, time.Time, error) { return active, since, nil }, |
| + modTime: func(string) (time.Time, error) { return written, nil }, |
| + unitState: func(context.Context, string) (bool, time.Time, []string, error) { |
| + if !active { |
| + return false, time.Time{}, nil, nil |
| + } |
| + return active, since, argv, nil |
| + }, |
| + // One link: a file in the unit's tree that points at another |
| + // tenant's database. |
| + resolve: func(path string) (string, error) { |
| + if path == "/var/lib/carlos-app/kass.example.com/instances/y.db" { |
| + return "/data/y.example.com.db", nil |
| + } |
| + return path, nil |
| + }, |
| } |
| } |
| // The ordinary box: a generated config, the unit restarted after it. |
| - got := probe(&generated, true, written.Add(time.Minute)).read(ctx) |
| + // The link is reported as written and as resolved. |
| + got := probe(&generated, true, written.Add(time.Minute)).read(ctx, nil) |
| if got.Err != nil || !got.ConfigPresent || !got.UnitActive || !got.ConfigModified.Equal(written) || |
| - !reflect.DeepEqual(got.Paths, []string{"/data/api.example.com.db", "/data/registry.db"}) { |
| + !reflect.DeepEqual(got.Paths, []string{"/data/api.example.com.db", "/data/registry.db", "/var/lib/carlos-app/kass.example.com/instances/y.db", "/data/y.example.com.db"}) { |
| t.Fatalf("generated config: %+v", got) |
| } |
| // No config and no unit: nothing external replicates. |
| - if got := probe(nil, false, time.Time{}).read(ctx); got.Err != nil || got.ConfigPresent { |
| + if got := probe(nil, false, time.Time{}).read(ctx, nil); got.Err != nil || got.ConfigPresent { |
| t.Fatalf("no config, no unit: %+v", got) |
| } |
| // A running unit with no config is replicating from somewhere unseen. |
| - if got := probe(nil, true, written).read(ctx); got.Err == nil { |
| + if got := probe(nil, true, written).read(ctx, nil); got.Err == nil { |
| t.Fatal("a running unit with no config was accepted") |
| } |
| // A hand-kept config is refused whole. |
| hand := "dbs:\n - path: /data/registry.db\n replicas:\n - url: s3://bucket/x\n" |
| - if got := probe(&hand, true, written.Add(time.Minute)).read(ctx); !errors.Is(got.Err, ops.ErrConfigNotGenerated) || len(got.Paths) != 0 { |
| + if got := probe(&hand, true, written.Add(time.Minute)).read(ctx, nil); !errors.Is(got.Err, ops.ErrConfigNotGenerated) || len(got.Paths) != 0 { |
| t.Fatalf("a hand-kept config: %+v", got) |
| } |
| + // A unit given a database on its command line. |
| + args := probe(&generated, true, written.Add(time.Minute)) |
| + args.unitState = func(context.Context, string) (bool, time.Time, []string, error) { |
| + return true, written.Add(time.Minute), []string{"/usr/local/bin/litestream", "replicate", "/data/x.db", "s3://bucket/x"}, nil |
| + } |
| + if got := args.read(ctx, nil); got.Err == nil { |
| + t.Fatal("a unit replicating from its command line was accepted") |
| + } |
| + // A path that cannot be resolved. |
| + blind := probe(&generated, true, written.Add(time.Minute)) |
| + blind.resolve = func(string) (string, error) { return "", errors.New("permission denied") } |
| + if got := blind.read(ctx, nil); got.Err == nil { |
| + t.Fatal("a config path that could not be resolved was accepted") |
| + } |
| // The unit restarts while the config is being read. |
| calls := 0 |
| racing := probe(&generated, true, written.Add(time.Minute)) |
| - racing.unitState = func(context.Context, string) (bool, time.Time, error) { |
| + racing.unitState = func(context.Context, string) (bool, time.Time, []string, error) { |
| calls++ |
| - return true, written.Add(time.Duration(calls) * time.Minute), nil |
| + return true, written.Add(time.Duration(calls) * time.Minute), argv, nil |
| } |
| - if got := racing.read(ctx); got.Err == nil { |
| + if got := racing.read(ctx, nil); got.Err == nil { |
| t.Fatal("a unit that restarted mid-read was accepted") |
| } |
| - // The config is rewritten while it is being read. |
| + // The config is rewritten while it is being read: its bytes, or only |
| + // its modification time. |
| reads := 0 |
| rewriting := probe(&generated, true, written.Add(time.Minute)) |
| rewriting.readFile = func(string) ([]byte, error) { |
| @@ -93,23 +142,73 @@ func TestReplicatorProbe(t *testing.T) { |
| } |
| return []byte(generated), nil |
| } |
| - if got := rewriting.read(ctx); got.Err == nil { |
| + if got := rewriting.read(ctx, nil); got.Err == nil { |
| t.Fatal("a config rewritten mid-read was accepted") |
| } |
| + stats := 0 |
| + touched := probe(&generated, true, written.Add(time.Minute)) |
| + touched.modTime = func(string) (time.Time, error) { |
| + stats++ |
| + return written.Add(time.Duration(stats) * time.Second), nil |
| + } |
| + if got := touched.read(ctx, nil); got.Err == nil { |
| + t.Fatal("a config whose modification time moved mid-read was accepted") |
| + } |
| // systemd cannot be asked. |
| deaf := probe(&generated, true, written) |
| - deaf.unitState = func(context.Context, string) (bool, time.Time, error) { |
| - return false, time.Time{}, errors.New("systemctl: not found") |
| + deaf.unitState = func(context.Context, string) (bool, time.Time, []string, error) { |
| + return false, time.Time{}, nil, errors.New("systemctl: not found") |
| } |
| - if got := deaf.read(ctx); got.Err == nil { |
| + if got := deaf.read(ctx, nil); got.Err == nil { |
| t.Fatal("an unreadable unit state was accepted") |
| } |
| - // What the probe reads feeds the scope: an unreadable state attests |
| - // nothing, a readable one attests with its exclusions. |
| - if scope, _ := durabilityScope(scopeRoutes(), "/data", deaf.read(ctx)); scope.Attested { |
| - t.Fatal("an unreadable replicator attested") |
| +} |
| + |
| +// The routes and the replicator's state are one observation. A route flips |
| +// from hibernating to always-on, the config is regenerated and the unit |
| +// restarts, all while the scope is being read: the registry read falls |
| +// inside the bracket, the unit is seen to have moved, and nothing is |
| +// attested. And the link in the unit's tree excludes the owner of the |
| +// database it points at. |
| +func TestReadDurabilityScopeIsOneObservation(t *testing.T) { |
| + ctx := context.Background() |
| + written := time.Date(2026, 9, 17, 8, 0, 0, 0, time.UTC) |
| + argv := []string{"/usr/local/bin/litestream", "replicate", "-config", "/etc/litestream.yml"} |
| + routes := []registry.Route{ |
| + {Host: "kass.example.com", Account: "bae", Backing: registry.BackingUnit, DBPath: "/var/lib/carlos-app/kass.example.com"}, |
| + {Host: "y.example.com", Account: "bac", Hibernate: true, DBPath: "/data/y.example.com.db"}, |
| + {Host: "shop.example.com", Account: "bad", Hibernate: true, DBPath: "/data/shop.example.com.db"}, |
| + } |
| + cfg := ops.LitestreamConfig([]string{"/data/registry.db", "/var/lib/carlos-app/kass.example.com/instances/y.db"}, "bucket", "prefix", "eu-west-1") |
| + restarts := 0 |
| + p := replicatorProbe{configPath: "/etc/litestream.yml", unit: externalReplicatorUnit, |
| + readFile: func(string) ([]byte, error) { return []byte(cfg), nil }, |
| + modTime: func(string) (time.Time, error) { return written, nil }, |
| + unitState: func(context.Context, string) (bool, time.Time, []string, error) { |
| + return true, written.Add(time.Duration(1+restarts) * time.Minute), argv, nil |
| + }, |
| + resolve: func(path string) (string, error) { |
| + if path == "/var/lib/carlos-app/kass.example.com/instances/y.db" { |
| + return "/data/y.example.com.db", nil |
| + } |
| + return path, nil |
| + }, |
| + } |
| + scope, _, reason := readDurabilityScope(ctx, p, func() ([]registry.Route, error) { return routes, nil }, "/data") |
| + if !scope.Attested || !reflect.DeepEqual(scope.Unmetered, []string{"bac", "bae"}) { |
| + t.Fatalf("scope %+v (%s): want the unit's owner and the owner of the database its link points at excluded, and bad vouched for", scope, reason) |
| + } |
| + // The unit restarts while the registry is being read. |
| + scope, _, reason = readDurabilityScope(ctx, p, func() ([]registry.Route, error) { |
| + restarts++ |
| + return routes, nil |
| + }, "/data") |
| + if scope.Attested || reason == "" { |
| + t.Fatalf("a restart around the registry read attested: %+v", scope) |
| } |
| - if scope, reason := durabilityScope(scopeRoutes(), "/data", probe(&generated, true, written.Add(time.Minute)).read(ctx)); !scope.Attested { |
| - t.Fatalf("a readable replicator did not attest: %s", reason) |
| + // The registry cannot be read. |
| + scope, _, _ = readDurabilityScope(ctx, p, func() ([]registry.Route, error) { return nil, errors.New("registry locked") }, "/data") |
| + if scope.Attested { |
| + t.Fatal("an unreadable registry attested") |
| } |
| } |
diff --git a/internal/host/durabilityscope.go b/internal/host/durabilityscope.go| index b35e32c7a..b2902eb2a 100644 |
| --- a/internal/host/durabilityscope.go |
| +++ b/internal/host/durabilityscope.go |
| @@ -1,12 +1,15 @@ |
| package host |
| import ( |
| + "context" |
| "fmt" |
| + "log" |
| "path/filepath" |
| "sort" |
| "strings" |
| "time" |
| + "github.com/carlosframework/platform/internal/ops" |
| "github.com/carlosframework/platform/internal/registry" |
| "github.com/carlosframework/platform/internal/release" |
| "github.com/carlosframework/platform/internal/usage" |
| @@ -15,9 +18,9 @@ import ( |
| // The scoped attestation of the replication ledger: what this box may |
| // honestly claim about who replicates through the ledger and who does not |
| // (docs/superpowers/plans/2026-10-01-durability-attestation-path.md, path |
| -// A). NOTHING IN THE AGENT CALLS THIS YET. No production box attests, and |
| -// none will until the call is wired behind an explicit switch that has had |
| -// its own review and a scoped approval to set. |
| +// A). The agent calls this only when CARLOS_DURABILITY_ATTEST=scoped is |
| +// set in its environment (scopedAttestationEnabled). It is unset |
| +// everywhere, and setting it on a production box needs a scoped approval. |
| // |
| // A box has two replicators. The agent's own (the activator's embedded |
| // Litestream, the move worker) records every request in the ledger. The |
| @@ -34,7 +37,8 @@ type replicatorState struct { |
| // ConfigPresent is false when there is no config file at all: the |
| // external replicator has nothing to replicate. |
| ConfigPresent bool |
| - // Paths are the database paths the config names (ops.ParseLitestreamDBs). |
| + // Paths are the database paths the config names, each as written and, |
| + // where that differs, as resolved through symlinks. |
| Paths []string |
| // ConfigModified is the config file's modification time. |
| ConfigModified time.Time |
| @@ -60,12 +64,27 @@ type replicatorState struct { |
| // generator hands to the box-wide replicator). A route with no |
| // database and no unit, a static site for one, excludes nobody; |
| // - the config: it owns a route whose database path the live config |
| -// names, whatever kind of route the registry says it is. |
| +// names, whatever kind of route the registry says it is. EVERY route |
| +// whose path equals or encloses a configured path is an owner, not the |
| +// first one found: a hibernating route's database can sit inside a |
| +// unit's tree, and two routes can record one file. |
| +// |
| +// Paths are compared as written and as resolved. resolve follows symlinks |
| +// (filepath.EvalSymlinks on a real box). A unit tenant controls its own |
| +// state directory and can link a file in it to another tenant's database; |
| +// the generator, which follows links, then names the link, and only the |
| +// resolved path shows whose database the replicator is streaming. The |
| +// caller resolves the config's paths (replicatorState.Paths); this |
| +// function resolves the routes'. A route path that does not resolve (the |
| +// database does not exist yet) is matched as written. |
| // |
| // Nothing is attested, for anyone, when the box cannot tell what the |
| // external replicator is doing: |
| // |
| // - its state could not be read; |
| +// - the unit is running and there is no config file: it is replicating |
| +// from a config that has been removed since it started, or from |
| +// arguments, and neither can be seen; |
| // - the config names a database no route owns and that is not the box's |
| // own (the registry or the console database): somebody's data is being |
| // replicated and nobody can be named as its owner; |
| @@ -76,17 +95,18 @@ type replicatorState struct { |
| // The box's own databases are platform data on the platform's prefix, not |
| // tenant replication, and exclude nobody. |
| // |
| -// What the caller owes before this may be wired: Paths must be every |
| -// database the config names. ops.ParseLitestreamDBs reads only the shape |
| -// the generator writes, so a reader built on it must refuse (Err) a config |
| -// it did not fully account for, or a hand-kept config would hide databases |
| -// from this function. And the console database's path follows the box's |
| -// configuration (ops.consoleDBPath), which the reader must resolve the |
| -// same way the generator does. |
| -func durabilityScope(routes []registry.Route, dataDir string, rs replicatorState) (usage.DurabilityAttestation, string) { |
| +// What the caller owes: Paths must be every database the config names |
| +// (replicatorProbe refuses a config it cannot fully account for), the |
| +// routes and the replicator's state must be one coherent observation |
| +// (readDurabilityScope brackets them). The console database's path is |
| +// resolved the way the generator resolves it (ops.ConsoleDBPath). |
| +func durabilityScope(routes []registry.Route, dataDir string, rs replicatorState, resolve func(string) (string, error)) (usage.DurabilityAttestation, string) { |
| if rs.Err != nil { |
| return usage.DurabilityAttestation{}, fmt.Sprintf("the external replicator's state could not be read: %v", rs.Err) |
| } |
| + if rs.UnitActive && !rs.ConfigPresent { |
| + return usage.DurabilityAttestation{}, "the external replicator is running and has no config file; what it replicates cannot be seen" |
| + } |
| if rs.ConfigPresent && rs.UnitActive && rs.UnitStarted.Before(rs.ConfigModified) { |
| return usage.DurabilityAttestation{}, "the external replicator has been running since before its config was last written; what it replicates is not what the config says" |
| } |
| @@ -102,33 +122,43 @@ func durabilityScope(routes []registry.Route, dataDir string, rs replicatorState |
| if rt.Backing == registry.BackingUnit || (rt.DBPath != "" && !rt.Hibernate) { |
| excluded[acct] = true |
| } |
| - if rt.DBPath != "" { |
| - owners = append(owners, owned{path: filepath.Clean(rt.DBPath), tree: rt.Backing == registry.BackingUnit, account: acct}) |
| + if rt.DBPath == "" { |
| + continue |
| + } |
| + tree := rt.Backing == registry.BackingUnit |
| + written := filepath.Clean(rt.DBPath) |
| + owners = append(owners, owned{path: written, tree: tree, account: acct}) |
| + if resolve != nil { |
| + if real, err := resolve(written); err == nil && filepath.Clean(real) != written { |
| + owners = append(owners, owned{path: filepath.Clean(real), tree: tree, account: acct}) |
| + } |
| } |
| } |
| own := map[string]bool{ |
| - filepath.Clean(filepath.Join(dataDir, "registry.db")): true, |
| - filepath.Clean(filepath.Join(dataDir, "console", "console.db")): true, |
| + filepath.Clean(filepath.Join(dataDir, "registry.db")): true, |
| + filepath.Clean(ops.ConsoleDBPath(dataDir)): true, |
| } |
| if rs.ConfigPresent { |
| for _, raw := range rs.Paths { |
| path := filepath.Clean(raw) |
| + if !filepath.IsAbs(path) { |
| + return usage.DurabilityAttestation{}, fmt.Sprintf("the external replicator's config names %s, which is not an absolute path", raw) |
| + } |
| if own[path] { |
| continue |
| } |
| - account := "" |
| + matched := false |
| for _, o := range owners { |
| // A unit route's path may be a directory the generator |
| // walks for databases (ops.treeDBs). |
| if path == o.path || (o.tree && strings.HasPrefix(path, o.path+string(filepath.Separator))) { |
| - account = o.account |
| - break |
| + excluded[o.account] = true |
| + matched = true |
| } |
| } |
| - if account == "" { |
| + if !matched { |
| return usage.DurabilityAttestation{}, fmt.Sprintf("the external replicator replicates %s, which no route owns", path) |
| } |
| - excluded[account] = true |
| } |
| } |
| out := usage.DurabilityAttestation{Attested: true} |
| @@ -138,3 +168,102 @@ func durabilityScope(routes []registry.Route, dataDir string, rs replicatorState |
| sort.Strings(out.Unmetered) |
| return out, "" |
| } |
| + |
| +// durabilityAttestEnv is the agent's switch for the scoped attestation. |
| +// Unset, or set to anything but "scoped", the ledger attests nothing, as |
| +// it always has. |
| +const durabilityAttestEnv = "CARLOS_DURABILITY_ATTEST" |
| + |
| +func scopedAttestationEnabled(getenv func(string) string) bool { |
| + return strings.TrimSpace(getenv(durabilityAttestEnv)) == "scoped" |
| +} |
| + |
| +// enableScopedAttestation makes every publish tick observe the box and |
| +// apply the scope BEFORE it seals anything, so a day is never closed on a |
| +// claim that was not checked in the same tick. |
| +func (p *durabilityPublisher) enableScopedAttestation(probe replicatorProbe, listRoutes func() ([]registry.Route, error), dataDir string) { |
| + if p == nil { |
| + return |
| + } |
| + lastReason := "\x00" |
| + p.attest = func(ctx context.Context) { |
| + scope, rs, reason := readDurabilityScope(ctx, probe, listRoutes, dataDir) |
| + if err := p.applyScope(scope, rs, p.now()); err != nil { |
| + log.Printf("agent: durability: attestation: %v", err) |
| + } |
| + if reason != lastReason { |
| + if reason != "" { |
| + log.Printf("agent: durability: attesting nothing: %s", reason) |
| + } else { |
| + log.Printf("agent: durability: attesting, with %d account(s) excluded", len(scope.Unmetered)) |
| + } |
| + lastReason = reason |
| + } |
| + } |
| +} |
| + |
| +// applyScope moves the ledger's claim to scope, dating every change at the |
| +// earliest instant the previous claim could have stopped being true. |
| +// |
| +// - Nothing was attested and scope attests: the claim starts now. It |
| +// says nothing about the time before this observation. |
| +// - Scope excludes an account the ledger was vouching for: the exclusion |
| +// starts no later than the last tick that verified the old claim, and |
| +// no later than the instant the replicator's config was last written |
| +// (the replicator cannot touch a database before a config naming it |
| +// existed). That keeps a day from closing complete for an account the |
| +// external replicator began touching just before midnight. |
| +// - Scope attests nothing and the ledger was attesting: the claim ends |
| +// at the last tick that verified it. |
| +// |
| +// An exclusion never lapses within a run: the ledger keeps the union. |
| +// |
| +// When the ledger refuses the instant (it precedes the current coverage |
| +// session), the change is filed as an escape instead, which the ledger |
| +// clamps itself and which also makes the day of the escape incomplete for |
| +// everyone on the box: blunter, and never more generous. |
| +func (p *durabilityPublisher) applyScope(scope usage.DurabilityAttestation, rs replicatorState, now time.Time) error { |
| + cur := p.ledger.Attestation() |
| + verified := p.scopeVerifiedAt |
| + if verified.IsZero() || verified.After(now) { |
| + verified = now |
| + } |
| + defer func() { p.scopeVerifiedAt = now }() |
| + if !scope.Attested { |
| + if !cur.Attested { |
| + return nil |
| + } |
| + if err := p.ledger.SetAttestation(usage.DurabilityAttestation{}, verified); err != nil { |
| + // An owner that cannot be named withdraws the claim for |
| + // everyone, from the instant given. |
| + p.ledger.Escaped("", verified) |
| + } |
| + return nil |
| + } |
| + if !cur.Attested { |
| + return p.ledger.SetAttestation(scope, now) |
| + } |
| + had := map[string]bool{} |
| + for _, acct := range cur.Unmetered { |
| + had[acct] = true |
| + } |
| + var added []string |
| + for _, acct := range scope.Unmetered { |
| + if !had[acct] { |
| + added = append(added, acct) |
| + } |
| + } |
| + if len(added) == 0 { |
| + return nil |
| + } |
| + from := verified |
| + if rs.ConfigPresent && !rs.ConfigModified.IsZero() && rs.ConfigModified.Before(from) { |
| + from = rs.ConfigModified |
| + } |
| + if err := p.ledger.SetAttestation(scope, from); err != nil { |
| + for _, acct := range added { |
| + p.ledger.Escaped(acct, from) |
| + } |
| + } |
| + return nil |
| +} |
diff --git a/internal/host/durabilityscope_test.go b/internal/host/durabilityscope_test.go| index edcf0c1d3..d7070eb66 100644 |
| --- a/internal/host/durabilityscope_test.go |
| +++ b/internal/host/durabilityscope_test.go |
| @@ -7,6 +7,7 @@ import ( |
| "testing" |
| "time" |
| + "github.com/carlosframework/platform/internal/ops" |
| "github.com/carlosframework/platform/internal/registry" |
| "github.com/carlosframework/platform/internal/release" |
| "github.com/carlosframework/platform/internal/usage" |
| @@ -30,6 +31,15 @@ func scopeState(paths ...string) replicatorState { |
| return replicatorState{ConfigPresent: true, Paths: paths, ConfigModified: written, UnitActive: true, UnitStarted: written.Add(time.Minute)} |
| } |
| +// testResolve stands in for the filesystem: one link, everything else as |
| +// written. |
| +func testResolve(path string) (string, error) { |
| + if path == "/data/link.db" { |
| + return "/data/real.db", nil |
| + } |
| + return path, nil |
| +} |
| + |
| func TestDurabilityScope(t *testing.T) { |
| generated := []string{"/data/registry.db", "/data/console/console.db", "/data/api.example.com.db", "/var/lib/carlos-app/kass.example.com/registry.db", "/var/lib/carlos-app/kass.example.com/instances/a.db"} |
| cases := []struct { |
| @@ -41,8 +51,35 @@ func TestDurabilityScope(t *testing.T) { |
| }{ |
| {"the generated config: always-on and unit owners are excluded, hibernating and static owners are vouched for", |
| scopeRoutes(), scopeState(generated...), true, []string{"bad", "bae"}}, |
| - {"no config at all: the registry alone decides", |
| + {"no config and no running unit: the registry alone decides", |
| scopeRoutes(), replicatorState{}, true, []string{"bad", "bae"}}, |
| + {"a running unit with no config file: nothing is attested", |
| + scopeRoutes()[:1], replicatorState{UnitActive: true, UnitStarted: time.Date(2026, 9, 17, 8, 0, 0, 0, time.UTC)}, false, nil}, |
| + {"a hibernating route's database inside a unit's tree: both owners are excluded, in either route order", |
| + []registry.Route{ |
| + {Host: "kass.example.com", Account: "bae", Backing: registry.BackingUnit, DBPath: "/srv/unit"}, |
| + {Host: "a.example.com", Account: "bac", Hibernate: true, DBPath: "/srv/unit/a.db"}, |
| + }, scopeState("/srv/unit/a.db"), true, []string{"bac", "bae"}}, |
| + {"the same, routes reversed", |
| + []registry.Route{ |
| + {Host: "a.example.com", Account: "bac", Hibernate: true, DBPath: "/srv/unit/a.db"}, |
| + {Host: "kass.example.com", Account: "bae", Backing: registry.BackingUnit, DBPath: "/srv/unit"}, |
| + }, scopeState("/srv/unit/a.db"), true, []string{"bac", "bae"}}, |
| + {"two routes record one database: both owners are excluded", |
| + []registry.Route{ |
| + {Host: "x.example.com", Account: "bac", Hibernate: true, DBPath: "/data/x.db"}, |
| + {Host: "y.example.com", Account: "bad", Hibernate: true, DBPath: "/data/x.db"}, |
| + }, scopeState("/data/x.db"), true, []string{"bac", "bad"}}, |
| + {"a link in a unit's tree to a hibernating route's database, reported as written and as resolved: both owners are excluded", |
| + []registry.Route{ |
| + {Host: "kass.example.com", Account: "bae", Backing: registry.BackingUnit, DBPath: "/srv/unit"}, |
| + {Host: "y.example.com", Account: "bac", Hibernate: true, DBPath: "/data/y.example.com.db"}, |
| + }, scopeState("/srv/unit/instances/y.db", "/data/y.example.com.db"), true, []string{"bac", "bae"}}, |
| + {"a route whose own recorded path is a link: the config's resolved path still finds it", |
| + []registry.Route{{Host: "z.example.com", Account: "bac", Hibernate: true, DBPath: "/data/link.db"}}, |
| + scopeState("/data/real.db"), true, []string{"bac"}}, |
| + {"a relative path in the config: nothing is attested", |
| + scopeRoutes(), scopeState("data/x.db"), false, nil}, |
| {"only hibernating exec routes: nobody is excluded", |
| scopeRoutes()[:1], scopeState("/data/registry.db"), true, nil}, |
| {"the config names a hibernating route's database: its owner is excluded on that evidence", |
| @@ -66,7 +103,7 @@ func TestDurabilityScope(t *testing.T) { |
| } |
| for _, c := range cases { |
| t.Run(c.name, func(t *testing.T) { |
| - got, reason := durabilityScope(c.routes, "/data", c.state) |
| + got, reason := durabilityScope(c.routes, "/data", c.state, testResolve) |
| if got.Attested != c.attested || !reflect.DeepEqual(got.Unmetered, c.unmetered) { |
| t.Fatalf("attested=%v unmetered=%v (reason %q), want attested=%v unmetered=%v", got.Attested, got.Unmetered, reason, c.attested, c.unmetered) |
| } |
| @@ -102,7 +139,7 @@ func TestScopedAttestationPublishesADayIncompleteForTheUnitOwner(t *testing.T) { |
| if p.ledger.Attestation().Attested { |
| t.Fatal("a freshly opened production ledger attests") |
| } |
| - scope, reason := durabilityScope(scopeRoutes(), "/data", scopeState("/data/registry.db", "/data/api.example.com.db", "/var/lib/carlos-app/kass.example.com/registry.db")) |
| + scope, reason := durabilityScope(scopeRoutes(), "/data", scopeState("/data/registry.db", "/data/api.example.com.db", "/var/lib/carlos-app/kass.example.com/registry.db"), testResolve) |
| if !scope.Attested { |
| t.Fatalf("fixture scope: %s", reason) |
| } |
| @@ -148,3 +185,181 @@ func TestScopedAttestationPublishesADayIncompleteForTheUnitOwner(t *testing.T) { |
| } |
| } |
| } |
| + |
| +// The switch. Only the exact value turns the attestation on. |
| +func TestScopedAttestationIsOffUnlessAskedForByName(t *testing.T) { |
| + for value, want := range map[string]bool{"": false, "scoped": true, " scoped ": true, "true": false, "1": false, "all": false, "SCOPED": false} { |
| + if got := scopedAttestationEnabled(func(string) string { return value }); got != want { |
| + t.Errorf("%s=%q: enabled=%v, want %v", durabilityAttestEnv, value, got, want) |
| + } |
| + } |
| + // A publisher nobody enabled attests nothing, tick after tick. |
| + p, _, _ := newTestDurabilityPublisher(t) |
| + store := release.NewDirStore(t.TempDir()) |
| + for i := 0; i < 2; i++ { |
| + if err := p.publish(context.Background(), store); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if p.attest != nil || p.ledger.Attestation().Attested { |
| + t.Fatal("a publisher with the switch off attests") |
| + } |
| +} |
| + |
| +// attestFixture is a publisher with the scoped attestation enabled over a |
| +// box whose routes, replicator config and unit the test controls. |
| +type attestFixture struct { |
| + p *durabilityPublisher |
| + clock time.Time |
| + routes []registry.Route |
| + paths []string |
| + written time.Time |
| + started time.Time |
| + stateErr error |
| + store release.Store |
| +} |
| + |
| +func newAttestFixture(t *testing.T) *attestFixture { |
| + t.Helper() |
| + f := &attestFixture{clock: time.Date(2026, 9, 16, 23, 0, 0, 0, time.UTC), store: release.NewDirStore(t.TempDir())} |
| + f.written = f.clock.Add(-time.Hour) |
| + f.started = f.written.Add(time.Minute) |
| + f.routes = []registry.Route{ |
| + {Host: "shop.example.com", Account: "bac", App: "shop", Hibernate: true, DBPath: "/data/shop.example.com.db"}, |
| + {Host: "api.example.com", Account: "bad", App: "api", DBPath: "/data/api.example.com.db"}, |
| + } |
| + f.paths = []string{"/data/registry.db", "/data/api.example.com.db"} |
| + p, err := newDurabilityPublisher(t.TempDir(), "box1", func() time.Time { return f.clock }) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + t.Cleanup(func() { |
| + ctx, cancel := context.WithTimeout(context.Background(), time.Second) |
| + defer cancel() |
| + if !p.stopped.Load() { |
| + _ = p.stop(ctx, nil) |
| + } |
| + _ = p.ledger.WaitLate(ctx) |
| + }) |
| + argv := []string{"/usr/local/bin/litestream", "replicate", "-config", "/etc/litestream.yml"} |
| + probe := replicatorProbe{configPath: "/etc/litestream.yml", unit: externalReplicatorUnit, |
| + readFile: func(string) ([]byte, error) { |
| + return []byte(ops.LitestreamConfig(f.paths, "bucket", "prefix", "eu-west-1")), nil |
| + }, |
| + modTime: func(string) (time.Time, error) { return f.written, nil }, |
| + unitState: func(context.Context, string) (bool, time.Time, []string, error) { |
| + if f.stateErr != nil { |
| + return false, time.Time{}, nil, f.stateErr |
| + } |
| + return true, f.started, argv, nil |
| + }, |
| + resolve: func(path string) (string, error) { return path, nil }, |
| + } |
| + p.enableScopedAttestation(probe, func() ([]registry.Route, error) { return f.routes, nil }, "/data") |
| + f.p = p |
| + return f |
| +} |
| + |
| +func (f *attestFixture) tick(t *testing.T, at time.Time) { |
| + t.Helper() |
| + f.clock = at |
| + if err := f.p.publish(context.Background(), f.store); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +// completeFor reads the published 17th and says whether it is complete for |
| +// acct over the whole day. |
| +func (f *attestFixture) completeFor(t *testing.T, acct string) bool { |
| + t.Helper() |
| + raw, _, err := f.store.Get(context.Background(), usage.DurabilityDayKey("box1", "2026-09-17")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + day, err := usage.ParseDurabilityDay(raw, "box1", "2026-09-17") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + from, to := time.Date(2026, 9, 17, 0, 0, 0, 0, time.UTC), time.Date(2026, 9, 18, 0, 0, 0, 0, time.UTC) |
| + for _, snap := range day.Epochs { |
| + c, err := snap.Decode() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + ok, _ := c.CompletenessFor(acct, from, to) |
| + return ok |
| + } |
| + return false |
| +} |
| + |
| +// With the switch on, each tick observes the box before it seals. The |
| +// hibernating route's owner is vouched for and the always-on route's is |
| +// not. |
| +func TestEnabledScopedAttestationVouchesPerAccount(t *testing.T) { |
| + f := newAttestFixture(t) |
| + f.tick(t, f.clock) |
| + if got := f.p.ledger.Attestation(); !got.Attested || !reflect.DeepEqual(got.Unmetered, []string{"bad"}) { |
| + t.Fatalf("after the first tick: %+v", got) |
| + } |
| + f.tick(t, time.Date(2026, 9, 17, 12, 0, 0, 0, time.UTC)) |
| + f.tick(t, time.Date(2026, 9, 18, 0, 5, 0, 0, time.UTC)) |
| + if !f.completeFor(t, "bac") { |
| + t.Fatal("the day is not complete for the hibernating route's owner") |
| + } |
| + if f.completeFor(t, "bad") { |
| + t.Fatal("the day reads complete for the always-on route's owner") |
| + } |
| +} |
| + |
| +// At 23:58 a route stops hibernating, the config is regenerated to name |
| +// its database and the unit restarts. The next tick is at 00:05. The day |
| +// that is about to close must NOT close complete for that account: the |
| +// exclusion is dated at the config's writing, before the seal. |
| +func TestScopedAttestationExcludesFromWhenTheConfigWasWritten(t *testing.T) { |
| + f := newAttestFixture(t) |
| + f.tick(t, f.clock) |
| + f.tick(t, time.Date(2026, 9, 17, 23, 0, 0, 0, time.UTC)) |
| + if !reflect.DeepEqual(f.p.ledger.Attestation().Unmetered, []string{"bad"}) { |
| + t.Fatalf("fixture: %+v", f.p.ledger.Attestation()) |
| + } |
| + f.routes[0].Hibernate = false |
| + f.paths = append(f.paths, "/data/shop.example.com.db") |
| + f.written = time.Date(2026, 9, 17, 23, 58, 0, 0, time.UTC) |
| + f.started = f.written.Add(10 * time.Second) |
| + f.tick(t, time.Date(2026, 9, 18, 0, 5, 0, 0, time.UTC)) |
| + if got := f.p.ledger.Attestation(); !got.Attested || !reflect.DeepEqual(got.Unmetered, []string{"bac", "bad"}) { |
| + t.Fatalf("after the route changed: %+v", got) |
| + } |
| + if f.completeFor(t, "bac") { |
| + t.Fatal("the 17th closed complete for an account the external replicator began touching at 23:58") |
| + } |
| +} |
| + |
| +// When the box can no longer tell what the replicator is doing, the claim |
| +// ends, and it ends at the last tick that verified it, not at the tick |
| +// that noticed. |
| +func TestScopedAttestationEndsWhenThePremiseFails(t *testing.T) { |
| + for name, breakIt := range map[string]func(f *attestFixture){ |
| + "the unit's state cannot be read": func(f *attestFixture) { f.stateErr = errors.New("systemctl: not found") }, |
| + "the config names a database no route owns": func(f *attestFixture) { f.paths = append(f.paths, "/data/leftover.db") }, |
| + "the config is newer than the running unit": func(f *attestFixture) { f.written = f.started.Add(time.Hour) }, |
| + } { |
| + t.Run(name, func(t *testing.T) { |
| + f := newAttestFixture(t) |
| + f.tick(t, f.clock) |
| + f.tick(t, time.Date(2026, 9, 17, 6, 0, 0, 0, time.UTC)) |
| + if !f.p.ledger.Attestation().Attested { |
| + t.Fatal("fixture: should be attesting") |
| + } |
| + breakIt(f) |
| + f.tick(t, time.Date(2026, 9, 17, 18, 0, 0, 0, time.UTC)) |
| + if f.p.ledger.Attestation().Attested { |
| + t.Fatal("still attesting after the premise failed") |
| + } |
| + f.tick(t, time.Date(2026, 9, 18, 0, 5, 0, 0, time.UTC)) |
| + if f.completeFor(t, "bac") { |
| + t.Fatal("the day reads complete for an account although the claim ended during it") |
| + } |
| + }) |
| + } |
| +} |
diff --git a/internal/ops/litestream.go b/internal/ops/litestream.go| index 8fa88badf..c75eeb0c5 100644 |
| --- a/internal/ops/litestream.go |
| +++ b/internal/ops/litestream.go |
| @@ -369,6 +369,10 @@ func treeDBs(root string) []string { |
| // layout <data>/console/console.db — expressed against dataDir rather than |
| // hardcoded to /data so a test box's console is found under its own root and |
| // never the real one. |
| +// ConsoleDBPath is where this box's console keeps its database, as the |
| +// config generator resolves it. |
| +func ConsoleDBPath(dataDir string) string { return consoleDBPath(dataDir) } |
| + |
| func consoleDBPath(dataDir string) string { |
| if p := strings.TrimSpace(os.Getenv("CONSOLE_DB")); p != "" { |
| return filepath.Clean(p) |