amadan
host: the probe's publisher is built from the bytes it checked and their token (Task 6 fix round 2, finding 2)
openPrepared read and checked P's current manifest, then hostfiles.Open read it again for the publisher: a replacement between the two reads became the baseline and supplied the compare-and-swap token, so a publication could succeed over bytes never compared with the journalled version. hostfiles.OpenAt builds a publisher from bytes already read and the token answered for them; openPrepared reads the key once, decodes those bytes for its checks and opens the publisher over exactly them. A replacement after that read fails the commit (ErrSuperseded). Regressions: a store that replaces the manifest right after the first read of it makes the establishment fail and leaves the replacement as it was; OpenAt refuses a missing token and another manifest version, fails over a replacement, and commits as usual over the same bytes.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
5 files changed,
+159
−6
internal/host/movesfiles.go+7 −3internal/host/movesfiles_test.go+70 −0internal/hostfiles/manifest.go+12 −3internal/hostfiles/prepared_test.go+47 −0internal/hostfiles/publisher.go+23 −0
diff --git a/internal/host/movesfiles.go b/internal/host/movesfiles.go| index f0eb656dd..827a11924 100644 |
| --- a/internal/host/movesfiles.go |
| +++ b/internal/host/movesfiles.go |
| @@ -495,12 +495,16 @@ func (f *movesFiles) openPrepared(ctx context.Context, files release.Store, o ho |
| if prepared.Seq != 0 { |
| return nil, fmt.Errorf("%w: the prepared files manifest %s@%s is at sequence %d, not 0", hostbox.ErrChanged, key, pm.Version, prepared.Seq) |
| } |
| - cur, _, err := files.Get(ctx, key) |
| + // One read of the key: the bytes and the token the publisher is built |
| + // from are the ones checked here, never a second read (fix round 2, |
| + // finding 2). A replacement after this read fails the commit's |
| + // compare-and-swap. |
| + cur, token, err := files.Get(ctx, key) |
| if err != nil { |
| return nil, err |
| } |
| if !bytes.Equal(cur, exact) { |
| - current, found, err := hostfiles.Load(ctx, files, prefix) |
| + current, found, err := hostfiles.Load(ctx, exactBytesStore{key: key, body: cur, version: token}, prefix) |
| if err != nil || !found { |
| return nil, fmt.Errorf("host: the files manifest %s: %v", key, err) |
| } |
| @@ -508,7 +512,7 @@ func (f *movesFiles) openPrepared(ctx context.Context, files release.Store, o ho |
| return nil, fmt.Errorf("%w: the files manifest %s is neither the prepared manifest %s nor this box's establishment of it", hostbox.ErrChanged, key, pm.Version) |
| } |
| } |
| - return hostfiles.Open(ctx, files, prefix) |
| + return hostfiles.OpenAt(files, prefix, cur, token) |
| } |
| // establishedOver reports whether cur is an establishment of prepared by |
diff --git a/internal/host/movesfiles_test.go b/internal/host/movesfiles_test.go| index c6181792a..187eb44ce 100644 |
| --- a/internal/host/movesfiles_test.go |
| +++ b/internal/host/movesfiles_test.go |
| @@ -557,6 +557,76 @@ func TestEstablishKeepsOriginsPastAdmissionPruning(t *testing.T) { |
| } |
| } |
| +// replacingStore replaces one key's object right after the first Get of |
| +// it returns: another writer landing between a reader's check and its |
| +// use. |
| +type replacingStore struct { |
| + release.Store |
| + key string |
| + replace func() |
| + once sync.Once |
| +} |
| + |
| +func (s *replacingStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + body, etag, err := s.Store.Get(ctx, key) |
| + if key == s.key && err == nil { |
| + s.once.Do(s.replace) |
| + } |
| + return body, etag, err |
| +} |
| + |
| +// preparedFixture builds a prepared manifest for mv1 with its binding |
| +// journalled, the env file carrying the minted key, and the probe's |
| +// observation and inputs. |
| +func preparedFixture(t *testing.T) (*filesFixture, preparedManifest, hostclaim.Observed, ClaimLaunchInputs) { |
| + t.Helper() |
| + x := newFilesFixture(t) |
| + ctx := context.Background() |
| + dir := filepath.Join(x.dir, filesHost) |
| + if err := os.MkdirAll(dir, 0o755); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := os.WriteFile(filepath.Join(dir, "a.txt"), []byte("A"), 0o640); err != nil { |
| + t.Fatal(err) |
| + } |
| + x.writeEnv("K=1\n") |
| + pm, err := x.f.BuildPrepared(ctx, filesLoc, "mv1", "mv1", 7, dir, x.envPath(), true, hostfiles.WakeState{V: 1, Host: filesHost, Timers: []hostfiles.WakeTimer{}}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + x.journal("mv1", pm) |
| + x.writeEnv("K=1\nCARLOS_INSTANCE_KEY=abc\n") |
| + o := servingObserved("cur", "cur:ie:@root:ie") |
| + o.Claim.Op = hostclaim.Op{Kind: hostclaim.KindLanding, ID: "mv1", Phase: "probing", Exec: "uk-box"} |
| + in := launchInputs("mv1") |
| + in.Prepared = true |
| + return x, pm, o, in |
| +} |
| + |
| +// The probe's publisher is built from the bytes the establishment checked |
| +// and the token captured with them, never from a second read (fix round |
| +// 2, finding 2). Another writer replacing P's manifest between the check |
| +// and the publication makes the compare-and-swap fail: nothing is |
| +// published over bytes that were never compared with the journalled |
| +// version, and the replacement stands as it was. |
| +func TestEstablishPublishesOnlyOverTheBytesItChecked(t *testing.T) { |
| + x, pm, o, in := preparedFixture(t) |
| + ctx := context.Background() |
| + other := hostfiles.Manifest{V: hostfiles.ManifestVersion, Seq: 5, Entries: map[string]hostfiles.Entry{"b.txt": {Object: movesFilesPrefix(filesLoc, "mv1") + "objects/0000000000000005-" + sha([]byte("B")), SHA256: sha([]byte("B")), Size: 1, Seq: 5}}, Platform: map[string]hostfiles.Entry{}} |
| + replacement := mustJSON(t, other) |
| + x.f.Files = &replacingStore{Store: x.files, key: pm.Key, replace: func() { x.bucket.Put("data", pm.Key, replacement) }} |
| + if err := x.f.Establish(ctx, o, in, x); err == nil { |
| + t.Fatal("published over a manifest replaced after the check") |
| + } |
| + cur, _ := x.bucket.Current("data", pm.Key) |
| + if string(cur) != string(replacement) { |
| + t.Fatalf("the replacement was published over: %s", cur) |
| + } |
| + if x.workRan != 0 && x.workRan > 2 { |
| + t.Fatalf("the local effect ran after a refused publication (%d)", x.workRan) |
| + } |
| +} |
| + |
| // stallingStore holds one key's write until released, signalling that it |
| // is stalled: a slow S3 request. |
| type stallingStore struct { |
diff --git a/internal/hostfiles/manifest.go b/internal/hostfiles/manifest.go| index bf441538a..7a15abfef 100644 |
| --- a/internal/hostfiles/manifest.go |
| +++ b/internal/hostfiles/manifest.go |
| @@ -126,12 +126,21 @@ func load(ctx context.Context, s release.Store, prefix string) (Manifest, string |
| if err != nil { |
| return Manifest{}, "", false, err |
| } |
| + m, err := decode(prefix, raw) |
| + if err != nil { |
| + return Manifest{}, "", false, err |
| + } |
| + return m, etag, true, nil |
| +} |
| + |
| +// decode is a manifest from its exact bytes. |
| +func decode(prefix string, raw []byte) (Manifest, error) { |
| var m Manifest |
| if err := json.Unmarshal(raw, &m); err != nil { |
| - return Manifest{}, "", false, fmt.Errorf("hostfiles: manifest %s: %w", prefix, err) |
| + return Manifest{}, fmt.Errorf("hostfiles: manifest %s: %w", prefix, err) |
| } |
| if m.V != ManifestVersion { |
| - return Manifest{}, "", false, fmt.Errorf("hostfiles: manifest %s is version %d; this binary reads %d", prefix, m.V, ManifestVersion) |
| + return Manifest{}, fmt.Errorf("hostfiles: manifest %s is version %d; this binary reads %d", prefix, m.V, ManifestVersion) |
| } |
| if m.Entries == nil { |
| m.Entries = map[string]Entry{} |
| @@ -139,5 +148,5 @@ func load(ctx context.Context, s release.Store, prefix string) (Manifest, string |
| if m.Platform == nil { |
| m.Platform = map[string]Entry{} |
| } |
| - return m, etag, true, nil |
| + return m, nil |
| } |
diff --git a/internal/hostfiles/prepared_test.go b/internal/hostfiles/prepared_test.go| index dbcd5f833..6ef352aae 100644 |
| --- a/internal/hostfiles/prepared_test.go |
| +++ b/internal/hostfiles/prepared_test.go |
| @@ -197,6 +197,53 @@ func TestPreparedStageSkippedNamesNoObject(t *testing.T) { |
| } |
| } |
| +// OpenAt publishes over the bytes its caller checked and the token read |
| +// with them: a replacement at the key after that read fails the commit |
| +// (ErrSuperseded) and the replacement stands; the same bytes still at the |
| +// key commit as usual. |
| +func TestOpenAtPublishesOnlyOverTheBytesRead(t *testing.T) { |
| + ctx, s := context.Background(), release.NewDirStore(t.TempDir()) |
| + src := open(t, s) |
| + stage(t, src, "a.txt", "A") |
| + commit(t, src) |
| + body, etag, err := s.Get(ctx, ManifestKey(pfx)) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + p, err := OpenAt(s, pfx, body, etag) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := OpenAt(s, pfx, body, ""); err == nil { |
| + t.Fatal("opened without a token") |
| + } |
| + if _, err := OpenAt(s, pfx, []byte(`{"v":1}`), etag); err == nil { |
| + t.Fatal("opened a manifest of another version") |
| + } |
| + // Another writer replaces the key after the read. |
| + other := open(t, s) |
| + stage(t, other, "b.txt", "B") |
| + commit(t, other) |
| + replaced, _, _ := s.Get(ctx, ManifestKey(pfx)) |
| + stage(t, p, "c.txt", "C") |
| + if _, err := p.Commit(ctx); !errors.Is(err, ErrSuperseded) { |
| + t.Fatalf("commit over a replacement: %v", err) |
| + } |
| + if again, _, _ := s.Get(ctx, ManifestKey(pfx)); string(again) != string(replaced) { |
| + t.Fatal("the replacement was published over") |
| + } |
| + // The same bytes still at the key: the commit goes through. |
| + body, etag, _ = s.Get(ctx, ManifestKey(pfx)) |
| + q, err := OpenAt(s, pfx, body, etag) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + stage(t, q, "c.txt", "C") |
| + if seq := commit(t, q); seq != 3 { |
| + t.Fatalf("seq %d", seq) |
| + } |
| +} |
| + |
| // NewFromCopy seeds a publisher for a generation that has no manifest yet |
| // with a predecessor's validated manifest: every entry keeps its |
| // origin-qualified object, the sequence continues from the copy, and the |
diff --git a/internal/hostfiles/publisher.go b/internal/hostfiles/publisher.go| index a55621888..0abf0b136 100644 |
| --- a/internal/hostfiles/publisher.go |
| +++ b/internal/hostfiles/publisher.go |
| @@ -49,6 +49,29 @@ func Open(ctx context.Context, s release.Store, prefix string) (*Publisher, erro |
| return &Publisher{s: s, prefix: prefix, man: m, etag: etag, next: m.Seq}, nil |
| } |
| +// OpenAt is a publisher over the manifest bytes a caller already read and |
| +// checked, with the token the store answered for them: the publisher's |
| +// baseline and its compare-and-swap token are exactly those, never a |
| +// second read (Task 6 fix round 2, finding 2). A manifest replaced at the |
| +// key after that read makes the first Commit fail (ErrSuperseded) rather |
| +// than publish over bytes nobody checked. |
| +func OpenAt(s release.Store, prefix string, body []byte, etag string) (*Publisher, error) { |
| + if s == nil { |
| + return nil, errors.New("hostfiles: open: no store") |
| + } |
| + if err := checkPrefix(prefix); err != nil { |
| + return nil, err |
| + } |
| + if etag == "" { |
| + return nil, fmt.Errorf("hostfiles: open %s at its read bytes: no token", prefix) |
| + } |
| + m, err := decode(prefix, body) |
| + if err != nil { |
| + return nil, err |
| + } |
| + return &Publisher{s: s, prefix: prefix, man: m, etag: etag, next: m.Seq}, nil |
| +} |
| + |
| func sum(body []byte) string { |
| h := sha256.Sum256(body) |
| return hex.EncodeToString(h[:]) |