carlos / platform Public

Ownership is decided on the file, not the spelling of its path; shadow state leaves the tenant's directory

From the first review of this branch (both reviewers, P1): a unit tenant
could put a link in its state directory, named like a database and
pointing at another tenant's database or the console's, and have it
streamed into its own replica prefix and metered to itself.

- A listed path is owned only if nothing beneath the claiming route's
  directory is a link and no other listed path is the same file.
  Otherwise it is unowned and the replicator refuses to start.
- NewMeteredStore compares files by identity, not by path text.
- Each database's litestream metadata directory is given by the caller
  and lives under the platform's data directory, not beside the
  database where a tenant could plant a link under its name.

Open: the check is made when the store is built (a file can be swapped
for a link later), and the owner is still resolved once.

Nothing calls this code. Nothing attests.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

claude 3 weeks ago 1eace90e7b34e5eb2ef8d8c6b956d3edada31b33 parent b85355c

6 files changed, +210 −21

docs/superpowers/specs/DISPOSITIONS-metered-replicator.md +4 −2
internal/lease/meteredstore.go +28 −4
internal/lease/meteredstore_test.go +27 −13
internal/ops/litestream_metered.go +5 −1
internal/ops/litestream_owners.go +66 −1
internal/ops/litestream_owners_test.go +80 −0

diff --git a/docs/superpowers/specs/DISPOSITIONS-metered-replicator.md b/docs/superpowers/specs/DISPOSITIONS-metered-replicator.md

index d745e87cd..1d211ca1a 100644

--- a/docs/superpowers/specs/DISPOSITIONS-metered-replicator.md

+++ b/docs/superpowers/specs/DISPOSITIONS-metered-replicator.md

@@ -4,12 +4,12 @@ Branch `feat/carloku-metered-replicator-20261001`. Nothing calls this code and n

## Review one (reviewed `ebab1a37`; Astra rejected, Opus rejected)

-Reports: `CODEX-FINDINGS-metered-replicator-review1.md`, `OPUS-REVIEW-metered-replicator-review1.md`. Nothing below is fixed yet.

+Reports: `CODEX-FINDINGS-metered-replicator-review1.md`, `OPUS-REVIEW-metered-replicator-review1.md`. One row below is fixed; the rest are open.

| Finding | Disposition |

| --- | --- |

| Both, P1: the owner of a database is resolved once, when its client is built, and never again. A route moved to another account, or an instance deleted and made again, goes on being counted to the first owner. | Accepted. Open. Ownership has to be bound to the registry generation it was read at and re-checked, with the client refused or rebuilt on a change. |

-| Both, P1: ownership and the duplicate-file check compare paths as text. A tenant symlink, or two spellings of one path, bills one tenant for another's database. | Accepted. Open. Resolve to the file's identity (device and inode, opened beneath the data root) before comparing. |

+| Both, P1: ownership and the duplicate-file check compare paths as text. A tenant symlink, or two spellings of one path, bills one tenant for another's database. | Fixed in the commit after `40558c1d`. A listed path is owned only if nothing beneath the claiming route's directory is a link, and no two listed paths resolve to one file; either makes it unowned and the replicator refuses to start. The store compares files by identity (`os.SameFile`). Shadow state moved out of the tenant's directory to `<data dir>/.replicator/<replica key>`. Reproductions: `TestFleetDBOwnersDoesNotFollowATenantsLink`, `TestMeteredStoreRefusesWhatItCannotAttribute`. Still open: the check is made when the store is built, and a tenant can replace a file with a link afterwards; closing that needs the databases opened beneath their root, and the ownership re-check of the first row. |

| Astra P2, Opus P3-5: a discovery error, or an unreadable or vanished database, becomes successful partial coverage | Accepted. Open. Must refuse or hold. |

| Astra P2: unknown databases are dropped before the refusal check | Accepted. Open. |

| Astra P2, Opus P2-1: "refuses anything it cannot meter" is true only while the store is built; the returned store still exposes an ordinary registration path | Accepted. Open. |

@@ -20,3 +20,5 @@ Reports: `CODEX-FINDINGS-metered-replicator-review1.md`, `OPUS-REVIEW-metered-re

| Both, tests: none shows the right recorder on the right client or a store-driven request being metered; several only exercise a fake | Accepted. Open. |

| Opus P3-1 to P3-4: two non-atomic registry reads; an arbitrary instance for one account and app; prefix uniqueness by exact string; a typed-nil recorder | Accepted. Open. |

| Opus: what is still missing before this could attest, beyond the launch branch's item 1 | Recorded in the report; to be carried into the plan. |

+Mutation checks in a scratch copy, each failing its test: the link check off; a file listed twice accepted; the store comparing paths and not files.

diff --git a/internal/lease/meteredstore.go b/internal/lease/meteredstore.go

index 032357bcd..367dbbf6a 100644

--- a/internal/lease/meteredstore.go

+++ b/internal/lease/meteredstore.go

@@ -3,6 +3,7 @@ package lease

import (

"errors"

"fmt"

+ "os"

"path/filepath"

"time"

@@ -48,6 +49,13 @@ type MeteredDB struct {

// Meter records every request the replica client makes. It is bound

// by the caller to the database's owner. Required.

Meter RequestRecorder

+ // MetaDir is where litestream keeps this database's shadow state.

+ // Required, absolute, and not shared with any other database. The

+ // caller puts it in a tree only the platform can write: beside the

+ // database would be inside a directory a unit tenant owns, where the

+ // tenant could plant a link under the predictable name and have the

+ // replicator, running as root, write and delete through it.

+ MetaDir string

// SyncInterval is litestream's replica sync tick. Zero keeps its

// default.

SyncInterval time.Duration

@@ -65,12 +73,27 @@ func NewMeteredStore(bucket, region string, dbs []MeteredDB) (*litestream.Store,

if len(dbs) == 0 {

return nil, errors.New("lease: metered store: no databases")

}

- paths, prefixes := map[string]bool{}, map[string]bool{}

+ paths, prefixes, metas := map[string]bool{}, map[string]bool{}, map[string]bool{}

+ var files []os.FileInfo

built := make([]*litestream.DB, 0, len(dbs))

for _, d := range dbs {

+ // Two entries that are one file, however their paths are spelled

+ // (a link, a second mount point): compared as files, not as text.

+ if fi, serr := os.Stat(d.Path); serr == nil {

+ for _, other := range files {

+ if os.SameFile(fi, other) {

+ return nil, fmt.Errorf("lease: metered store: %s is a file already listed under another path", d.Path)

+ }

+ files = append(files, fi)

+ }

switch {

case d.Path == "" || !filepath.IsAbs(d.Path):

return nil, fmt.Errorf("lease: metered store: database path %q is not absolute", d.Path)

+ case d.MetaDir == "" || !filepath.IsAbs(d.MetaDir):

+ return nil, fmt.Errorf("lease: metered store: %s has no absolute metadata directory", d.Path)

+ case metas[filepath.Clean(d.MetaDir)]:

+ return nil, fmt.Errorf("lease: metered store: metadata directory %q is used by two databases", d.MetaDir)

case d.Prefix == "":

return nil, fmt.Errorf("lease: metered store: %s has no replica prefix", d.Path)

case d.Meter == nil:

@@ -80,7 +103,7 @@ func NewMeteredStore(bucket, region string, dbs []MeteredDB) (*litestream.Store,

case prefixes[d.Prefix]:

return nil, fmt.Errorf("lease: metered store: replica prefix %q is used by two databases", d.Prefix)

}

- paths[filepath.Clean(d.Path)], prefixes[d.Prefix] = true, true

+ paths[filepath.Clean(d.Path)], prefixes[d.Prefix], metas[filepath.Clean(d.MetaDir)] = true, true, true

option, err := requestMeterOption(d.Meter)

if err != nil {

return nil, fmt.Errorf("lease: metered store: %s: %w", d.Path, err)

@@ -88,8 +111,9 @@ func NewMeteredStore(bucket, region string, dbs []MeteredDB) (*litestream.Store,

db := litestream.NewDB(d.Path)

// A metadata directory of its own: sharing one with another

// replicator (the activator's, or the external binary's default)

- // corrupts shadow state, even across sequential runs.

- db.SetMetaPath(filepath.Join(filepath.Dir(d.Path), ".replicator-"+filepath.Base(d.Path)+"-litestream"))

+ // corrupts shadow state, even across sequential runs. It is the

+ // caller's, in a tree tenants cannot write (MeteredDB.MetaDir).

+ db.SetMetaPath(filepath.Clean(d.MetaDir))

db.MonitorInterval = dbMonitorInterval

db.Replica = litestream.NewReplicaWithClient(db, newReplicaClient(bucket, d.Prefix, region, []func(*awss3.Options){option}...))

if d.SyncInterval > 0 {

diff --git a/internal/lease/meteredstore_test.go b/internal/lease/meteredstore_test.go

index 8ebc5ac62..bfd400823 100644

--- a/internal/lease/meteredstore_test.go

+++ b/internal/lease/meteredstore_test.go

@@ -4,6 +4,7 @@ import (

"bytes"

"context"

"errors"

+ "os"

"path/filepath"

"testing"

)

@@ -19,8 +20,8 @@ func TestMeteredStoreCountsEveryRequestOnItsOwnersMeter(t *testing.T) {

dir := t.TempDir()

one, two := &memoryRequestRecorder{}, &memoryRequestRecorder{}

store, err := NewMeteredStore("test-bucket", "eu-west-1", []MeteredDB{

- {Path: filepath.Join(dir, "one.db"), Prefix: "replicas/bab/one", Meter: one},

- {Path: filepath.Join(dir, "two.db"), Prefix: "replicas/bac/two", Meter: two},

+ {Path: filepath.Join(dir, "one.db"), Prefix: "replicas/bab/one", Meter: one, MetaDir: filepath.Join(dir, ".replicator", "one")},

+ {Path: filepath.Join(dir, "two.db"), Prefix: "replicas/bac/two", Meter: two, MetaDir: filepath.Join(dir, ".replicator", "two")},

})

if err != nil {

t.Fatal(err)

@@ -29,8 +30,8 @@ func TestMeteredStoreCountsEveryRequestOnItsOwnersMeter(t *testing.T) {

if len(dbs) != 2 {

t.Fatalf("%d databases in the store, want 2", len(dbs))

}

- if dbs[0].MetaPath() == dbs[1].MetaPath() || filepath.Base(dbs[0].MetaPath()) != ".replicator-one.db-litestream" {

- t.Fatalf("metadata paths %q and %q: want one each, in the replicator's own namespace", dbs[0].MetaPath(), dbs[1].MetaPath())

+ if a, b := dbs[0].MetaPath(), dbs[1].MetaPath(); a == b || a != filepath.Join(dir, ".replicator", "one") || b != filepath.Join(dir, ".replicator", "two") {

+ t.Fatalf("metadata paths %q and %q: want the directory each database was given", a, b)

}

ctx := context.Background()

if _, err := dbs[0].Replica.Client.WriteLTXFile(ctx, 0, 2, 2, bytes.NewReader(mustLTXBytes(t))); err != nil {

@@ -64,21 +65,34 @@ func TestMeteredStoreCountsEveryRequestOnItsOwnersMeter(t *testing.T) {

func TestMeteredStoreRefusesWhatItCannotAttribute(t *testing.T) {

dir := t.TempDir()

m := &memoryRequestRecorder{}

- ok := MeteredDB{Path: filepath.Join(dir, "a.db"), Prefix: "replicas/a", Meter: m}

+ metaA, metaB := filepath.Join(dir, ".replicator", "a"), filepath.Join(dir, ".replicator", "b")

+ ok := MeteredDB{Path: filepath.Join(dir, "a.db"), Prefix: "replicas/a", Meter: m, MetaDir: metaA}

+ // The same file under another name: a link a tenant could plant.

+ if err := os.WriteFile(ok.Path, []byte("x"), 0o644); err != nil {

+ t.Fatal(err)

+ }

+ link := filepath.Join(dir, "link.db")

+ if err := os.Symlink(ok.Path, link); err != nil {

+ t.Fatal(err)

+ }

for name, dbs := range map[string][]MeteredDB{

- "no databases": nil,

- "no meter": {{Path: ok.Path, Prefix: ok.Prefix}},

- "no prefix": {{Path: ok.Path, Meter: m}},

- "a relative path": {{Path: "a.db", Prefix: ok.Prefix, Meter: m}},

- "one file listed twice": {ok, {Path: filepath.Join(dir, ".", "a.db"), Prefix: "replicas/b", Meter: m}},

- "one prefix for two": {ok, {Path: filepath.Join(dir, "b.db"), Prefix: ok.Prefix, Meter: m}},

- "a meter missing in many": {ok, {Path: filepath.Join(dir, "b.db"), Prefix: "replicas/b"}},

+ "no databases": nil,

+ "no meter": {{Path: ok.Path, Prefix: ok.Prefix, MetaDir: metaA}},

+ "no prefix": {{Path: ok.Path, Meter: m, MetaDir: metaA}},

+ "a relative path": {{Path: "a.db", Prefix: ok.Prefix, Meter: m, MetaDir: metaA}},

+ "no metadata directory": {{Path: ok.Path, Prefix: ok.Prefix, Meter: m}},

+ "a relative metadata directory": {{Path: ok.Path, Prefix: ok.Prefix, Meter: m, MetaDir: ".replicator/a"}},

+ "one metadata directory for two": {ok, {Path: filepath.Join(dir, "b.db"), Prefix: "replicas/b", Meter: m, MetaDir: metaA}},

+ "one file listed twice": {ok, {Path: filepath.Join(dir, ".", "a.db"), Prefix: "replicas/b", Meter: m, MetaDir: metaB}},

+ "one file under a link": {ok, {Path: link, Prefix: "replicas/b", Meter: m, MetaDir: metaB}},

+ "one prefix for two": {ok, {Path: filepath.Join(dir, "b.db"), Prefix: ok.Prefix, Meter: m, MetaDir: metaB}},

+ "a meter missing in many": {ok, {Path: filepath.Join(dir, "b.db"), Prefix: "replicas/b", MetaDir: metaB}},

} {

if store, err := NewMeteredStore("test-bucket", "eu-west-1", dbs); err == nil || store != nil {

t.Errorf("%s: accepted", name)

}

- if _, err := NewMeteredStore("test-bucket", "eu-west-1", []MeteredDB{{Path: ok.Path, Prefix: ok.Prefix}}); !errors.Is(err, errRequestMeterRequired) {

+ if _, err := NewMeteredStore("test-bucket", "eu-west-1", []MeteredDB{{Path: ok.Path, Prefix: ok.Prefix, MetaDir: metaA}}); !errors.Is(err, errRequestMeterRequired) {

t.Errorf("no meter: %v, want the meter-required error", err)

}

if _, err := NewMeteredStore("", "eu-west-1", []MeteredDB{ok}); err == nil {

diff --git a/internal/ops/litestream_metered.go b/internal/ops/litestream_metered.go

index a752607e7..2fc7f4c0f 100644

--- a/internal/ops/litestream_metered.go

+++ b/internal/ops/litestream_metered.go

@@ -3,6 +3,7 @@ package ops

import (

"errors"

"fmt"

+ "path/filepath"

"strings"

"github.com/benbjohnson/litestream"

@@ -61,7 +62,10 @@ func MeteredFleetStore(reg *registry.Registry, dataDir, bucket, prefix, region s

if prefix != "" {

key = prefix + "/" + key

}

- dbs = append(dbs, lease.MeteredDB{Path: o.Path, Prefix: key, Meter: rec})

+ // Shadow state lives under the platform's data directory, keyed by

+ // the replica key, never beside a database in a tenant's tree.

+ meta := filepath.Join(dataDir, ".replicator", replicaKey(o.Path))

+ dbs = append(dbs, lease.MeteredDB{Path: o.Path, Prefix: key, Meter: rec, MetaDir: meta})

}

store, err := lease.NewMeteredStore(bucket, region, dbs)

if err != nil {

diff --git a/internal/ops/litestream_owners.go b/internal/ops/litestream_owners.go

index 630403deb..564a5e64d 100644

--- a/internal/ops/litestream_owners.go

+++ b/internal/ops/litestream_owners.go

@@ -1,6 +1,7 @@

package ops

import (

+ "os"

"path/filepath"

"sort"

"strings"

@@ -30,9 +31,19 @@ type OwnedDB struct {

// DBPath is a directory the file lies under (a unit tenant's tree). The

// box's registry and console databases are the platform's.

+// Ownership is decided on the file, not on the spelling of its path. A unit

+// tenant owns its state directory and can put a symbolic link in it. A

+// link named like a database and pointing at another tenant's database, or

+// at the console's, would otherwise be listed under the tenant's directory,

+// claimed by the tenant's route as text, streamed into the tenant's replica

+// prefix and metered to the tenant. So a listed path is owned only if no

+// part of it beneath the claiming route's directory is a link, and no two

+// listed paths are the same file.

+//

// unowned lists the databases FleetDBs would replicate that no single

// account can be shown to own: a file two routes of different accounts

-// both claim, or a route with no account. They are named, never dropped

+// both claim, a route with no account, a path that reaches its file

+// through a link, or a file listed twice. They are named, never dropped

// and never guessed. A replicator must not stream them unmetered, and a box

// that replicates one cannot say that every write was counted.

func FleetDBOwners(reg *registry.Registry, dataDir string) (owned []OwnedDB, unowned []string, skipped []string, err error) {

@@ -48,9 +59,30 @@ func FleetDBOwners(reg *registry.Registry, dataDir string) (owned []OwnedDB, uno

filepath.Clean(filepath.Join(dataDir, "registry.db")): true,

filepath.Clean(consoleDBPath(dataDir)): true,

}

+ // The file each listed path is, by its resolved name. Two listed paths

+ // that are one file are both unowned: one stream must never be written

+ // twice, nor two owners billed for it.

+ real := map[string]string{}

+ times := map[string]int{}

+ for _, p := range dbs {

+ p = filepath.Clean(p)

+ if r, rerr := filepath.EvalSymlinks(p); rerr == nil {

+ real[p] = filepath.Clean(r)

+ times[real[p]]++

+ }

for _, p := range dbs {

p = filepath.Clean(p)

+ if real[p] == "" || times[real[p]] > 1 {

+ // It cannot be resolved, or another listed path is this file.

+ unowned = append(unowned, p)

+ continue

+ }

if platform[p] {

+ if !noLinkBeneath(filepath.Clean(dataDir), p) {

+ unowned = append(unowned, p)

+ continue

+ }

owned = append(owned, OwnedDB{Path: p, Account: release.OpsAccount, Platform: true})

continue

}

@@ -73,6 +105,18 @@ func FleetDBOwners(reg *registry.Registry, dataDir string) (owned []OwnedDB, uno

unowned = append(unowned, p)

continue

}

+ // The claim is on a directory the tenant can write in, or on the

+ // file itself. Either way the path must lead to its file without a

+ // link: from the claimed directory down for a unit tenant, and the

+ // file itself for an exact claim.

+ root := filepath.Clean(owner.DBPath)

+ if root == p {

+ root = filepath.Dir(p)

+ }

+ if !noLinkBeneath(root, p) {

+ unowned = append(unowned, p)

+ continue

+ }

owned = append(owned, OwnedDB{Path: p, Account: owner.Account, App: owner.App, Host: owner.Host, Platform: owner.Account == release.OpsAccount})

}

sort.Slice(owned, func(i, j int) bool { return owned[i].Path < owned[j].Path })

@@ -100,3 +144,24 @@ func singleOwner(claims []registry.Route) (registry.Route, bool) {

}

return first, true

}

+// noLinkBeneath reports whether path leads to its file from root without

+// passing through a symbolic link: every component after root, the file

+// included, is what it is named and not a link to something else. root

+// itself may be reached however the operator arranged it (a data directory

+// behind a link is the operator's, not a tenant's).

+func noLinkBeneath(root, path string) bool {

+ rel, err := filepath.Rel(root, path)

+ if err != nil || rel == "." || rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) {

+ return false

+ }

+ at := root

+ for _, part := range strings.Split(rel, string(filepath.Separator)) {

+ at = filepath.Join(at, part)

+ fi, err := os.Lstat(at)

+ if err != nil || fi.Mode()&os.ModeSymlink != 0 {

+ return false

+ }

+ return true

diff --git a/internal/ops/litestream_owners_test.go b/internal/ops/litestream_owners_test.go

index 4bf886efc..af3b57bf0 100644

--- a/internal/ops/litestream_owners_test.go

+++ b/internal/ops/litestream_owners_test.go

@@ -1,6 +1,8 @@

package ops

import (

+ "errors"

+ "github.com/carlosframework/platform/internal/lease"

"os"

"path/filepath"

"testing"

@@ -96,3 +98,81 @@ func TestFleetDBOwnersNamesAnOwnerOrSaysThereIsNone(t *testing.T) {

t.Fatalf("FleetDBs lists %d databases; %d owned and %d unowned", len(dbs), len(owned), len(unowned))

}

+// A unit tenant owns its state directory and can put a link in it. A link

+// named like a database and pointing at another tenant's database, or at

+// the platform's own, must not be streamed and metered as the tenant's: it

+// is another owner's data and another owner's stream. Such a path is

+// unowned, and so is the real file when it is listed too (one file, two

+// paths). The replicator then refuses to start.

+func TestFleetDBOwnersDoesNotFollowATenantsLink(t *testing.T) {

+ dir := t.TempDir()

+ reg, err := registry.Open(filepath.Join(dir, "registry.db"))

+ if err != nil {

+ t.Fatal(err)

+ }

+ write := func(p string) string {

+ t.Helper()

+ if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil {

+ t.Fatal(err)

+ }

+ if err := os.WriteFile(p, []byte("x"), 0o644); err != nil {

+ t.Fatal(err)

+ }

+ return p

+ }

+ stateA := filepath.Join(t.TempDir(), "unit-a")

+ stateB := filepath.Join(t.TempDir(), "unit-b")

+ ownA := write(filepath.Join(stateA, "main.db"))

+ theirs := write(filepath.Join(stateB, "main.db"))

+ // Not a database by name, so it is not listed by itself; reached only

+ // through the link.

+ outside := write(filepath.Join(t.TempDir(), "elsewhere", "secrets.sqlite"))

+ toB := filepath.Join(stateA, "link-to-b.db")

+ toOutside := filepath.Join(stateA, "link-outside.db")

+ linkedDir := filepath.Join(stateA, "linked")

+ for target, link := range map[string]string{theirs: toB, outside: toOutside, stateB: linkedDir} {

+ if err := os.Symlink(target, link); err != nil {

+ t.Fatal(err)

+ }

+ for _, r := range []registry.Route{

+ {Host: "a.example.com", Account: "bac", App: "one", DBPath: stateA, Backing: registry.BackingUnit},

+ {Host: "b.example.com", Account: "bae", App: "two", DBPath: stateB, Backing: registry.BackingUnit},

+ } {

+ if err := reg.Upsert(r); err != nil {

+ t.Fatal(err)

+ }

+ owned, unowned, _, err := FleetDBOwners(reg, dir)

+ if err != nil {

+ t.Fatal(err)

+ }

+ isUnowned := map[string]bool{}

+ for _, p := range unowned {

+ isUnowned[p] = true

+ }

+ for _, o := range owned {

+ switch {

+ case o.Path == ownA:

+ if o.Account != "bac" {

+ t.Errorf("the tenant's own database: %+v", o)

+ }

+ case o.Platform:

+ default:

+ t.Errorf("owned: %+v; only the tenant's own database and the platform's may be owned here", o)

+ }

+ for _, p := range []string{toB, toOutside} {

+ if !isUnowned[p] {

+ t.Errorf("%s is a link out of the tenant's directory and is not named as unowned (unowned: %v)", p, unowned)

+ }

+ // B's database is one file under two listed paths.

+ if !isUnowned[theirs] {

+ t.Errorf("a file listed twice (once through a link) is not named as unowned: %v", unowned)

+ }

+ if _, _, err := MeteredFleetStore(reg, dir, "test-bucket", "replicas", "eu-west-1", func(OwnedDB) lease.RequestRecorder { return nopRecorder{} }); !errors.Is(err, ErrUnmeterable) {

+ t.Fatalf("the replicator over a tenant's link: %v, want ErrUnmeterable", err)

+ }