amadan

carlos / platform Public

hostclaim: Plan 15 fix round 1 (chain starts empty, live origins proven via the #250 backfill; head can't be forged or erased through Update; cache bound to one store; early exit on full proof; one switch parser; console can't write origin candidates; reads bounded through every wrapper)

Answers Opus's P2-1 and P3-1..5 and Astra's two P2s and P3 on f5af489f.
Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
codex pushed by claude 0b2cbf7b0e7130e4969fc612bd485033b1dfaf3b parent f5af489
14 files changed, +552 −129
  • internal/console/plan15_stream_test.go +84 −0
  • internal/host/claimoriginproof.go +1 −6
  • internal/hostclaim/origins.go +66 −67
  • internal/hostclaim/plan15_batch_test.go +36 −29
  • internal/hostclaim/plan15_fix1_test.go +171 −0
  • internal/hostclaim/plan15_origins_test.go +19 −6
  • internal/hostclaim/store.go +9 −0
  • internal/hostops/mutation_boundary.go +6 −0
  • internal/hostops/mutation_classifier.go +0 −6
  • internal/hostops/plan15_origins_test.go +9 −12
  • internal/hostops/plan15_stream_test.go +99 −0
  • internal/provision/plan9_b8_census_test.go +3 −3
  • internal/release/store.go +11 −0
  • internal/release/stream_required_test.go +38 −0

CI passed — run details

CI log 
=== gate ===
cd launchctx && go vet ./... && go test ./... -count=1
ok  	amadan.net/carlos/platform.git/launchctx	0.012s
?   	amadan.net/carlos/platform.git/launchctx/controlname	[no test files]
?   	amadan.net/carlos/platform.git/launchctx/wireid	[no test files]
go vet ./...
gofmt -l .: clean
go test ./... -count=1 -timeout 30m
ok  	github.com/carlosframework/platform	338.928s
?   	github.com/carlosframework/platform/cmd/changelog	[no test files]
ok  	github.com/carlosframework/platform/cmd/console	3.232s
ok  	github.com/carlosframework/platform/cmd/moveprobe	92.229s
ok  	github.com/carlosframework/platform/internal/activator	231.804s
ok  	github.com/carlosframework/platform/internal/agentsock	0.016s
ok  	github.com/carlosframework/platform/internal/alerting	0.005s
ok  	github.com/carlosframework/platform/internal/analytics	0.019s
ok  	github.com/carlosframework/platform/internal/appconfig	2.323s
ok  	github.com/carlosframework/platform/internal/apperr	0.071s
ok  	github.com/carlosframework/platform/internal/applog	0.145s
ok  	github.com/carlosframework/platform/internal/awsinv	0.004s
ok  	github.com/carlosframework/platform/internal/boxchoice	0.004s
ok  	github.com/carlosframework/platform/internal/boxid	0.680s
?   	github.com/carlosframework/platform/internal/boxid/boxidtest	[no test files]
?   	github.com/carlosframework/platform/internal/boxid/internal/signtest	[no test files]
ok  	github.com/carlosframework/platform/internal/boxvolumes	0.009s
ok  	github.com/carlosframework/platform/internal/caps	0.015s
ok  	github.com/carlosframework/platform/internal/certwarm	0.002s
ok  	github.com/carlosframework/platform/internal/cgroup	0.192s
ok  	github.com/carlosframework/platform/internal/changelog	0.256s
ok  	github.com/carlosframework/platform/internal/clamd	0.055s
?   	github.com/carlosframework/platform/internal/cliflag	[no test files]
ok  	github.com/carlosframework/platform/internal/console	361.730s
?   	github.com/carlosframework/platform/internal/controlname	[no test files]
?   	github.com/carlosframework/platform/internal/controlsock	[no test files]
ok  	github.com/carlosframework/platform/internal/dbdoor	6.425s
ok  	github.com/carlosframework/platform/internal/dbdoor/dbdoortest	0.009s
ok  	github.com/carlosframework/platform/internal/dbverify	0.004s
ok  	github.com/carlosframework/platform/internal/dnsrec	0.003s
ok  	github.com/carlosframework/platform/internal/docsite	0.005s
ok  	github.com/carlosframework/platform/internal/domains	2.005s
ok  	github.com/carlosframework/platform/internal/doorkey	0.006s
ok  	github.com/carlosframework/platform/internal/economics	0.004s
ok  	github.com/carlosframework/platform/internal/edge	25.977s
?   	github.com/carlosframework/platform/internal/edgeproof	[no test files]
ok  	github.com/carlosframework/platform/internal/edgesfile	0.006s
ok  	github.com/carlosframework/platform/internal/fleetapi	0.125s
ok  	github.com/carlosframework/platform/internal/gate	1.655s
ok  	github.com/carlosframework/platform/internal/genselect	0.015s
ok  	github.com/carlosframework/platform/internal/geo	0.014s
ok  	github.com/carlosframework/platform/internal/geoip	0.030s
ok  	github.com/carlosframework/platform/internal/host	61.826s
ok  	github.com/carlosframework/platform/internal/hostactors	0.409s
ok  	github.com/carlosframework/platform/internal/hostbox	1.250s
ok  	github.com/carlosframework/platform/internal/hostclaim	0.299s
ok  	github.com/carlosframework/platform/internal/hostevidence	0.010s
ok  	github.com/carlosframework/platform/internal/hostexec	0.297s
ok  	github.com/carlosframework/platform/internal/hostfence	0.035s
ok  	github.com/carlosframework/platform/internal/hostfiles	4.172s
ok  	github.com/carlosframework/platform/internal/hostfloor	0.013s
ok  	github.com/carlosframework/platform/internal/hostguard	0.134s
ok  	github.com/carlosframework/platform/internal/hostops	1.371s
?   	github.com/carlosframework/platform/internal/hostops/fencerelease	[no test files]
ok  	github.com/carlosframework/platform/internal/hostops/internal/fencerelease	0.011s
ok  	github.com/carlosframework/platform/internal/hostquarantine	0.012s
ok  	github.com/carlosframework/platform/internal/hostresolve	0.133s
ok  	github.com/carlosframework/platform/internal/hostwrite	0.011s
ok  	github.com/carlosframework/platform/internal/infraview	0.010s
ok  	github.com/carlosframework/platform/internal/instancectl	1.709s
ok  	github.com/carlosframework/platform/internal/instanceexec	2.697s
ok  	github.com/carlosframework/platform/internal/instancefs	0.013s
ok  	github.com/carlosframework/platform/internal/instancekey	0.011s
ok  	github.com/carlosframework/platform/internal/instanceupdates	0.017s
ok  	github.com/carlosframework/platform/internal/layout	1.409s
ok  	github.com/carlosframework/platform/internal/lease	9.443s
ok  	github.com/carlosframework/platform/internal/lease/leasetest	0.015s
ok  	github.com/carlosframework/platform/internal/ledger	0.026s
ok  	github.com/carlosframework/platform/internal/mail	0.175s
ok  	github.com/carlosframework/platform/internal/maintenance	0.077s
ok  	github.com/carlosframework/platform/internal/maintenanceexec	5.496s
ok  	github.com/carlosframework/platform/internal/maintenanceguard	0.067s
ok  	github.com/carlosframework/platform/internal/metrics	0.006s
ok  	github.com/carlosframework/platform/internal/moves	21.164s
ok  	github.com/carlosframework/platform/internal/moves/archive	0.740s
ok  	github.com/carlosframework/platform/internal/objstore	0.152s
ok  	github.com/carlosframework/platform/internal/ops	0.157s
ok  	github.com/carlosframework/platform/internal/park	0.197s
ok  	github.com/carlosframework/platform/internal/privatefile	0.003s
ok  	github.com/carlosframework/platform/internal/provision	3.242s
ok  	github.com/carlosframework/platform/internal/publicrelease	0.006s
?   	github.com/carlosframework/platform/internal/regionname	[no test files]
ok  	github.com/carlosframework/platform/internal/registry	3.862s
ok  	github.com/carlosframework/platform/internal/relay	7.216s
ok  	github.com/carlosframework/platform/internal/release	19.427s
ok  	github.com/carlosframework/platform/internal/rollout	3.276s
ok  	github.com/carlosframework/platform/internal/rolloutwords	0.003s
ok  	github.com/carlosframework/platform/internal/rootaws	0.007s
ok  	github.com/carlosframework/platform/internal/s3compat	0.005s
ok  	github.com/carlosframework/platform/internal/safefile	0.004s
ok  	github.com/carlosframework/platform/internal/schedule	2.273s
ok  	github.com/carlosframework/platform/internal/sim/hostlife	2.486s
ok  	github.com/carlosframework/platform/internal/sim/litestreamfacts	4.385s
ok  	github.com/carlosframework/platform/internal/status	0.040s
ok  	github.com/carlosframework/platform/internal/steering	0.006s
ok  	github.com/carlosframework/platform/internal/stopevidence	0.189s
ok  	github.com/carlosframework/platform/internal/svcrole	0.002s
ok  	github.com/carlosframework/platform/internal/telemetry	0.004s
ok  	github.com/carlosframework/platform/internal/termfence	0.004s
ok  	github.com/carlosframework/platform/internal/tunnelclient	0.004s
ok  	github.com/carlosframework/platform/internal/unixdial	0.003s
ok  	github.com/carlosframework/platform/internal/updatewords	0.003s
ok  	github.com/carlosframework/platform/internal/usage	0.004s
ok  	github.com/carlosframework/platform/internal/version	0.002s
ok  	github.com/carlosframework/platform/internal/vet	1.547s
=== cross ===
--- shard 0 of 5 ---
cross-compiling darwin/arm64
--- shard 1 of 5 ---
cross-compiling darwin/amd64
--- shard 2 of 5 ---
cross-compiling linux/arm64
--- shard 3 of 5 ---
cross-compiling windows/amd64
--- shard 4 of 5 ---
cross-compiling windows/arm64
=== minio-contract ===
minio contract: arm64, minio RELEASE.2025-09-07T16-13-09Z, mc RELEASE.2025-08-13T08-35-41Z
  minio: cached, digest verified
  mc: cached, digest verified
  minio healthy on 127.0.0.1:20999
Added `local` successfully.
Bucket created successfully `local/contract-bucket`.
Bucket created successfully `local/denied-bucket`.
Created policy `contract-only` successfully.
Added user `scopeduser` successfully.
Attached Policies: [contract-only]
To User: scopeduser
  buckets and scoped user ready
=== RUN   TestDeliveredContractMinio
=== RUN   TestDeliveredContractMinio/put
=== RUN   TestDeliveredContractMinio/get_round-trips_the_body
=== RUN   TestDeliveredContractMinio/list_shows_the_key
=== RUN   TestDeliveredContractMinio/head_and_list_agree_on_the_version
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_its_own_object
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_its_own_object/not_by_PutObjectTagging
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_its_own_object/and_not_by_x-amz-tagging_on_upload
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CreateMultipartUpload
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CopyObject_with_a_REPLACE_tagging_directive
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/but_multipart_itself_is_allowed
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_copy_itself_is_allowed
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/a_POST-policy_forged_tag_with_a_matching_version_is_still_not_a_verdict
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_an_ordinary_tagged-with-something-else_upload_is_allowed
=== RUN   TestDeliveredContractMinio/delete_removes_it
=== RUN   TestDeliveredContractMinio/cross-bucket_denial
--- PASS: TestDeliveredContractMinio (0.02s)
    --- PASS: TestDeliveredContractMinio/put (0.00s)
    --- PASS: TestDeliveredContractMinio/get_round-trips_the_body (0.00s)
    --- PASS: TestDeliveredContractMinio/list_shows_the_key (0.00s)
    --- PASS: TestDeliveredContractMinio/head_and_list_agree_on_the_version (0.00s)
    --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_its_own_object (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_its_own_object/not_by_PutObjectTagging (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_its_own_object/and_not_by_x-amz-tagging_on_upload (0.00s)
    --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route (0.01s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CreateMultipartUpload (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CopyObject_with_a_REPLACE_tagging_directive (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/but_multipart_itself_is_allowed (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_copy_itself_is_allowed (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/a_POST-policy_forged_tag_with_a_matching_version_is_still_not_a_verdict (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_an_ordinary_tagged-with-something-else_upload_is_allowed (0.00s)
    --- PASS: TestDeliveredContractMinio/delete_removes_it (0.00s)
    --- PASS: TestDeliveredContractMinio/cross-bucket_denial (0.00s)
PASS
ok  	github.com/carlosframework/platform/internal/objstore	0.024s
=== RUN   TestT4StorageProbeMinio
2026/10/12 22:56:27 moves: T4: STORE: list: operation error S3: ListObjectsV2, https response error StatusCode: 403, RequestID: 18DDEA40096FB6EC, HostID: 7c420053e0fbcd7f41653fdd92ffa2802c903d05303b1aa15adc2e8bb65d183f, api error SignatureDoesNotMatch: The request signature we calculated does not match the signature you provided. Check your key and signing method.
--- PASS: TestT4StorageProbeMinio (0.01s)
PASS
ok  	github.com/carlosframework/platform/internal/host	0.022s
minio contract: PASS
diff --git a/internal/console/plan15_stream_test.go b/internal/console/plan15_stream_test.go
new file mode 100644
index 000000000..a5eb84ab1
--- /dev/null
+++ b/internal/console/plan15_stream_test.go
@@ -0,0 +1,84 @@
+package console
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "io"
+ "strings"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostops"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+type plan15ConsoleStream struct {
+ release.Store
+ read, closed int
+}
+
+func (s *plan15ConsoleStream) Get(ctx context.Context, key string) ([]byte, string, error) {
+ if strings.Contains(key, "control/host-origins/") {
+ panic("console wrapper buffered origin")
+ }
+ return s.Store.Get(ctx, key)
+}
+func (s *plan15ConsoleStream) Open(context.Context, string) (io.ReadCloser, string, error) {
+ return plan15ConsoleReader{s, strings.NewReader(strings.Repeat("x", 8192))}, "", nil
+}
+
+type plan15ConsoleReader struct {
+ s *plan15ConsoleStream
+ r io.Reader
+}
+
+func (r plan15ConsoleReader) Read(b []byte) (int, error) {
+ n, e := r.r.Read(b)
+ r.s.read += n
+ return n, e
+}
+func (r plan15ConsoleReader) Close() error { r.s.closed++; return nil }
+
+type plan15ConsoleGetOnly struct{ release.Store }
+
+func TestPlan15ConsoleStreamingCompositions(t *testing.T) {
+ for _, stream := range []bool{false, true} {
+ t.Run(map[bool]string{false: "get-only", true: "streaming"}[stream], func(t *testing.T) {
+ raw := release.NewDirStore(t.TempDir())
+ logical := release.NewPrefixStore(raw, "deployment")
+ claim := &hostclaim.Claim{V: 2, Host: "x.example", Gen: 1, Origins: &hostclaim.OriginHead{Digest: strings.Repeat("a", 64), Count: 1}}
+ body, _ := json.Marshal(claim)
+ key, _ := hostclaim.Key(claim.Host)
+ tag, err := logical.Put(t.Context(), key, body, "")
+ if err != nil {
+ t.Fatal(err)
+ }
+ obs := hostclaim.Observed{Host: claim.Host, Claim: claim, ETag: tag}
+ counter := &plan15ConsoleStream{Store: raw}
+ var s release.Store = counter
+ if !stream {
+ s = plan15ConsoleGetOnly{s}
+ }
+ s = release.NewPrefixStore(s, "deployment")
+ s, err = hostops.NewConditionalRecordStore(s, make([]byte, 32))
+ if err != nil {
+ t.Fatal(err)
+ }
+ s, err = hostops.NewMutationBoundary(true, s, func(context.Context, hostops.MutationTarget) error { return nil })
+ if err != nil {
+ t.Fatal(err)
+ }
+ s = claimAliasStore{data: claimEffectStore{data: s}}
+ if got, err := hostclaim.AdmittedOrigin(t.Context(), s, obs, strings.Repeat("b", 64), "instances/x.example", "ie", "real"); got || !errors.Is(err, hostclaim.ErrMalformed) {
+ t.Fatalf("proof %v %v", got, err)
+ }
+ if stream && (counter.read != 4097 || counter.closed != 1) {
+ t.Fatalf("read=%d closed=%d", counter.read, counter.closed)
+ }
+ if !stream && (counter.read != 0 || counter.closed != 0) {
+ t.Fatal("opened get-only store")
+ }
+ })
+ }
+}
diff --git a/internal/host/claimoriginproof.go b/internal/host/claimoriginproof.go
index 33e153eb3..ff5276087 100644
--- a/internal/host/claimoriginproof.go
+++ b/internal/host/claimoriginproof.go
@@ -13,7 +13,6 @@ import (
// Admission.NewGeneration is deliberately not evidence of admission.
type ClaimOriginProof struct {
Claims release.Store
- Cache *hostclaim.OriginProofCache
HomeRegion, Region string
}
@@ -36,11 +35,7 @@ func (p ClaimOriginProof) Proven(ctx context.Context, o hostclaim.Observed, _ ho
set = append(set, hostclaim.OriginTuple{Incarnation: h.Incarnation, Location: h.Location, Region: region, Generation: generation})
}
}
- cache := p.Cache
- if cache == nil {
- cache = new(hostclaim.OriginProofCache)
- }
- proven, err := cache.Proven(ctx, p.Claims, o, set)
+ proven, err := hostclaim.AdmittedOrigins(ctx, p.Claims, o, set)
if err != nil {
return nil, err
}
diff --git a/internal/hostclaim/origins.go b/internal/hostclaim/origins.go
index 500fe527c..ca27c81df 100644
--- a/internal/hostclaim/origins.go
+++ b/internal/hostclaim/origins.go
@@ -9,11 +9,11 @@ import (
"errors"
"io"
"os"
- "reflect"
"sort"
"strings"
"sync"
+ "github.com/carlosframework/platform/internal/hostwrite"
"github.com/carlosframework/platform/internal/release"
)
@@ -44,7 +44,9 @@ type originNode struct {
}
func originEnabled() bool {
- return os.Getenv("CARLOS_HOST_CLAIM") == "1" && os.Getenv("CARLOS_HOST_CLAIM_ORIGINS") == "1"
+ return hostwrite.Enabled(os.Getenv) && hostwrite.Enabled(func(string) string {
+ return os.Getenv("CARLOS_HOST_CLAIM_ORIGINS")
+ })
}
func originDigest(b []byte) string {
@@ -170,7 +172,7 @@ func commitOrigins(ctx context.Context, s release.Store, obs Observed, next *Cla
if a.Session != h.Session || a.Generation != want.Generation || a.Region != want.Region {
return ErrWrongState
}
- // Bind migration to the stored observation, never a caller-edited admission list.
+ // Bind admission to the stored observation, never a caller-edited admission list.
fresh, err := Read(ctx, s, obs.Host)
if err != nil {
return err
@@ -180,58 +182,46 @@ func commitOrigins(ctx context.Context, s release.Store, obs Observed, next *Cla
if fresh.Absent || fresh.ETag != obs.ETag || !bytes.Equal(before, stored) {
return ErrConflict
}
- var origins []originTuple
- if old == nil {
- for _, a := range obs.Claim.Holder.Admissions {
- if a.Generation == "" {
- continue
- }
- origins = append(origins, originTuple{obs.Claim.Holder.Incarnation, obs.Claim.Holder.Location, a.Region, a.Generation})
- }
- }
- origins = append(origins, want)
+ // The chain starts empty at the independent writer boundary. Retained
+ // admissions may have been forged before that boundary; never seed them.
+ // Activation must re-copy every live origin through the #250 files backfill
+ // into a prefix admitted after the boundary.
head := OriginHead{}
if old != nil {
head = *old
- }
- for _, o := range origins {
- if !o.valid(obs.Host) {
- return ErrWrongState
- }
- if head.Count != 0 {
- last, err := readOrigin(ctx, s, obs.Host, head)
- if err != nil {
- return err
- }
- if last.Origin == o {
- continue
- }
- }
- if head.Count == originCountLimit {
- return ErrWrongState
- }
- node := originNode{V: 1, Host: obs.Host, Count: head.Count + 1, Previous: head.Digest, Origin: o}
- body, err := json.Marshal(node)
+ last, err := readOrigin(ctx, s, obs.Host, head)
if err != nil {
return err
}
- key := originKey(obs.Host, originDigest(body))
- if err = ValidateOriginObject(key, body); err != nil {
- return err
+ if last.Origin == want {
+ next.originAdmission = nil
+ return nil
}
- if _, err = s.Put(ctx, key, body, ""); errors.Is(err, release.ErrCASConflict) {
- existing, readErr := readOriginBytes(ctx, s, key)
- if readErr != nil {
- return readErr
- }
- if !bytes.Equal(existing, body) {
- return ErrMalformed
- }
- } else if err != nil {
- return err
+ }
+ if !want.valid(obs.Host) || head.Count == originCountLimit {
+ return ErrWrongState
+ }
+ node := originNode{V: 1, Host: obs.Host, Count: head.Count + 1, Previous: head.Digest, Origin: want}
+ body, err := json.Marshal(node)
+ if err != nil {
+ return err
+ }
+ key := originKey(obs.Host, originDigest(body))
+ if err = ValidateOriginObject(key, body); err != nil {
+ return err
+ }
+ if _, err = s.Put(ctx, key, body, ""); errors.Is(err, release.ErrCASConflict) {
+ existing, readErr := readOriginBytes(ctx, s, key)
+ if readErr != nil {
+ return readErr
}
- head = OriginHead{Digest: originDigest(body), Count: node.Count}
+ if !bytes.Equal(existing, body) {
+ return ErrMalformed
+ }
+ } else if err != nil {
+ return err
}
+ head = OriginHead{Digest: originDigest(body), Count: node.Count}
next.Origins = &head
next.originAdmission = nil
return nil
@@ -245,7 +235,7 @@ func sameOriginHead(a, b *OriginHead) bool {
}
// OriginProofCache retains one exact batch, after authenticating its observation.
-// Stores with noncomparable identities are never cached. Zero value is usable.
+// Construct it with NewOriginProofCache; its store cannot be replaced.
type OriginProofCache struct {
mu sync.Mutex
store release.Store
@@ -253,16 +243,21 @@ type OriginProofCache struct {
result map[OriginTuple]bool
}
+// NewOriginProofCache binds proof and reuse to one authoritative store. Queries
+// accept no store argument, so a cache cannot carry answers to another store.
+func NewOriginProofCache(s release.Store) *OriginProofCache {
+ return &OriginProofCache{store: s}
+}
+
func readOriginBytes(ctx context.Context, s release.Store, key string) ([]byte, error) {
if err := ctx.Err(); err != nil {
return nil, err
}
- // OpenObject's Get fallback would allocate an unbounded object.
- opener, ok := s.(release.Opener)
- if !ok {
+ // Require streaming through every OpenObject wrapper in the chain.
+ reader, _, err := release.OpenObjectStreaming(ctx, s, key)
+ if errors.Is(err, errors.ErrUnsupported) {
return nil, ErrMalformed
}
- reader, _, err := opener.Open(ctx, key)
if err != nil {
return nil, err
}
@@ -284,15 +279,20 @@ func readOriginBytes(ctx context.Context, s release.Store, key string) ([]byte,
}
// AdmittedOrigins authenticates once and proves the complete requested set in
-// one walk. It publishes no partial results on a malformed or unreadable chain.
+// one walk, stopping once every tuple is proven. A malformed or unreadable node
+// needed by the walk returns an error without partial results.
func AdmittedOrigins(ctx context.Context, s release.Store, obs Observed, set []OriginTuple) (map[OriginTuple]bool, error) {
- return new(OriginProofCache).Proven(ctx, s, obs, set)
+ return NewOriginProofCache(s).Proven(ctx, obs, set)
}
-func (cache *OriginProofCache) Proven(ctx context.Context, s release.Store, obs Observed, set []OriginTuple) (map[OriginTuple]bool, error) {
+func (cache *OriginProofCache) Proven(ctx context.Context, obs Observed, set []OriginTuple) (map[OriginTuple]bool, error) {
+ if cache == nil {
+ return nil, ErrMalformed
+ }
+ s := cache.store
cache.mu.Lock()
defer cache.mu.Unlock()
- invalidate := func() { cache.store, cache.key, cache.result = nil, "", nil }
+ invalidate := func() { cache.key, cache.result = "", nil }
if err := ctx.Err(); err != nil {
invalidate()
return nil, err
@@ -324,8 +324,8 @@ func (cache *OriginProofCache) Proven(ctx context.Context, s release.Store, obs
invalidate()
return nil, err
}
- // The store identity, host, head digest/count and exact canonical tuple set
- // are all in the cache key. ETag and the full body also bind legacy admissions.
+ // The store is fixed at construction. The key binds host, head, exact tuple
+ // set, ETag and full claim body, including legacy admissions.
tuples := make([]string, 0, len(wanted))
for tuple := range wanted {
b, _ := json.Marshal(tuple)
@@ -337,8 +337,7 @@ func (cache *OriginProofCache) Proven(ctx context.Context, s release.Store, obs
Set []string
}{obs.Host, obs.ETag, string(stored), tuples})
key := string(keyBytes)
- comparable := reflect.TypeOf(s).Comparable()
- if comparable && cache.store == s && cache.key == key {
+ if cache.key == key {
for tuple := range wanted {
wanted[tuple] = cache.result[tuple]
}
@@ -356,12 +355,14 @@ func (cache *OriginProofCache) Proven(ctx context.Context, s release.Store, obs
}
}
} else {
- for head := *c.Origins; head.Count > 0; {
+ remaining := len(wanted)
+ for head := *c.Origins; head.Count > 0 && remaining > 0; {
node, err := readOrigin(ctx, s, obs.Host, head)
if err != nil {
return nil, err
}
- if _, ok := wanted[node.Origin]; ok {
+ if proven, ok := wanted[node.Origin]; ok && !proven {
+ remaining--
wanted[node.Origin] = true
}
head = OriginHead{Digest: node.Previous, Count: head.Count - 1}
@@ -370,12 +371,10 @@ func (cache *OriginProofCache) Proven(ctx context.Context, s release.Store, obs
if err := ctx.Err(); err != nil {
return nil, err
}
- if comparable {
- cache.store, cache.key = s, key
- cache.result = make(map[OriginTuple]bool, len(wanted))
- for tuple, proven := range wanted {
- cache.result[tuple] = proven
- }
+ cache.key = key
+ cache.result = make(map[OriginTuple]bool, len(wanted))
+ for tuple, proven := range wanted {
+ cache.result[tuple] = proven
}
return wanted, nil
}
diff --git a/internal/hostclaim/plan15_batch_test.go b/internal/hostclaim/plan15_batch_test.go
index 4d8e5656a..58595f38f 100644
--- a/internal/hostclaim/plan15_batch_test.go
+++ b/internal/hostclaim/plan15_batch_test.go
@@ -3,6 +3,7 @@ package hostclaim
import (
"context"
"errors"
+ "fmt"
"io"
"strings"
"testing"
@@ -17,14 +18,14 @@ type plan15BoundedStore struct {
}
func (s *plan15BoundedStore) Get(ctx context.Context, key string) ([]byte, string, error) {
- if strings.HasPrefix(key, "control/host-origins/") {
+ if strings.Contains(key, "control/host-origins/") {
panic("unbounded node Get")
}
s.gets++
return s.DirStore.Get(ctx, key)
}
func (s *plan15BoundedStore) Put(ctx context.Context, key string, b []byte, tag string) (string, error) {
- if s.conflict && strings.HasPrefix(key, "control/host-origins/") {
+ if s.conflict && strings.Contains(key, "control/host-origins/") {
return "", release.ErrCASConflict
}
return s.DirStore.Put(ctx, key, b, tag)
@@ -49,27 +50,33 @@ func (s *plan15BoundedStore) Open(ctx context.Context, key string) (io.ReadClose
return s.DirStore.Open(ctx, key)
}
func TestPlan15BoundedReachableAndConflicting(t *testing.T) {
- for _, conflict := range []bool{false, true} {
- t.Run(map[bool]string{false: "reachable", true: "candidate-conflict"}[conflict], func(t *testing.T) {
- plan15Enable(t)
- s := &plan15BoundedStore{DirStore: release.NewDirStore(t.TempDir())}
- o := servingClaim(t, t.Context(), s)
- if !conflict {
- o = updateGenerationTest(t, t.Context(), s, o, plan15Admit(o, "real", "ie"))
- }
- s.oversized, s.conflict = true, conflict
- s.opens, s.closed, s.read = 0, 0, 0
- var err error
- if conflict {
- _, err = Update(t.Context(), s, o, plan15Admit(o, "candidate", "ie"))
- } else {
- h := o.Claim.Holder
- _, err = AdmittedOrigin(t.Context(), s, o, h.Incarnation, h.Location, "ie", "real")
- }
- if !errors.Is(err, ErrMalformed) || s.read != 4097 || s.closed != 1 || s.opens != 1 {
- t.Fatalf("err=%v bytes=%d closes=%d opens=%d", err, s.read, s.closed, s.opens)
- }
- })
+ for _, prefixed := range []bool{false, true} {
+ for _, conflict := range []bool{false, true} {
+ t.Run(fmt.Sprintf("prefix=%v/conflict=%v", prefixed, conflict), func(t *testing.T) {
+ plan15Enable(t)
+ counter := &plan15BoundedStore{DirStore: release.NewDirStore(t.TempDir())}
+ var s release.Store = counter
+ if prefixed {
+ s = release.NewPrefixStore(release.NewPrefixStore(s, "outer"), "inner")
+ }
+ o := servingClaim(t, t.Context(), s)
+ if !conflict {
+ o = updateGenerationTest(t, t.Context(), s, o, plan15Admit(o, "real", "ie"))
+ }
+ counter.oversized, counter.conflict = true, conflict
+ counter.opens, counter.closed, counter.read = 0, 0, 0
+ var err error
+ if conflict {
+ _, err = Update(t.Context(), s, o, plan15Admit(o, "candidate", "ie"))
+ } else {
+ h := o.Claim.Holder
+ _, err = AdmittedOrigin(t.Context(), s, o, h.Incarnation, h.Location, "ie", "real")
+ }
+ if !errors.Is(err, ErrMalformed) || counter.read != 4097 || counter.closed != 1 || counter.opens != 1 {
+ t.Fatalf("err=%v bytes=%d closes=%d opens=%d", err, counter.read, counter.closed, counter.opens)
+ }
+ })
+ }
}
}
func TestPlan15CacheBindings(t *testing.T) {
@@ -79,10 +86,10 @@ func TestPlan15CacheBindings(t *testing.T) {
o = updateGenerationTest(t, t.Context(), s, o, plan15Admit(o, "first", "ie"))
h := o.Claim.Holder
first := OriginTuple{h.Incarnation, h.Location, "ie", "first"}
- cache := new(OriginProofCache)
+ cache := NewOriginProofCache(s)
prove := func(obs Observed, set []OriginTuple, want bool) {
t.Helper()
- result, err := cache.Proven(t.Context(), s, obs, set)
+ result, err := cache.Proven(t.Context(), obs, set)
if err != nil || result[set[0]] != want {
t.Fatalf("%v %v", result, err)
}
@@ -108,7 +115,7 @@ func TestPlan15CacheBindings(t *testing.T) {
if _, err := other.Put(t.Context(), key, body, ""); err != nil {
t.Fatal(err)
}
- if got, err := cache.Proven(t.Context(), other, o, []OriginTuple{different}); got != nil || err == nil {
+ if got, err := NewOriginProofCache(other).Proven(t.Context(), o, []OriginTuple{different}); got != nil || err == nil {
t.Fatal("foreign store reused")
}
prove(o, []OriginTuple{first}, true)
@@ -116,12 +123,12 @@ func TestPlan15CacheBindings(t *testing.T) {
c := *o.Claim
edited.Claim = &c
c.Origins = &OriginHead{Digest: strings.Repeat("f", 64), Count: o.Claim.Origins.Count}
- if got, err := cache.Proven(t.Context(), s, edited, []OriginTuple{first}); got != nil || !errors.Is(err, ErrConflict) {
+ if got, err := cache.Proven(t.Context(), edited, []OriginTuple{first}); got != nil || !errors.Is(err, ErrConflict) {
t.Fatal("edited head reused")
}
prove(o, []OriginTuple{first}, true)
newer := updateGenerationTest(t, t.Context(), s, o, plan15Admit(o, "second", "ie"))
- if got, err := cache.Proven(t.Context(), s, o, []OriginTuple{first}); got != nil || !errors.Is(err, ErrConflict) {
+ if got, err := cache.Proven(t.Context(), o, []OriginTuple{first}); got != nil || !errors.Is(err, ErrConflict) {
t.Fatal("stale observation reused")
}
s.opens = 0
@@ -135,7 +142,7 @@ func TestPlan15CacheBindings(t *testing.T) {
}
second := first
second.Generation = "second"
- if got, err := cache.Proven(t.Context(), s, newer, []OriginTuple{first, second}); got != nil || err == nil {
+ if got, err := cache.Proven(t.Context(), newer, []OriginTuple{first, second}); got != nil || err == nil {
t.Fatal("partial proof escaped")
}
}
diff --git a/internal/hostclaim/plan15_fix1_test.go b/internal/hostclaim/plan15_fix1_test.go
new file mode 100644
index 000000000..7e4eea695
--- /dev/null
+++ b/internal/hostclaim/plan15_fix1_test.go
@@ -0,0 +1,171 @@
+package hostclaim
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "io"
+ "reflect"
+ "strings"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func TestPlan15ActivationDoesNotTrustRetainedAdmissions(t *testing.T) {
+ t.Setenv("CARLOS_HOST_CLAIM", "")
+ t.Setenv("CARLOS_HOST_CLAIM_ORIGINS", "")
+ s := release.NewDirStore(t.TempDir())
+ o := servingClaim(t, t.Context(), s)
+ o.Claim.Holder.Admissions = append(o.Claim.Holder.Admissions, AdmissionRecord{Region: "ie", Generation: "injected", Mode: "cold"})
+ key, _ := Key(o.Host)
+ body, _ := json.Marshal(o.Claim)
+ var err error
+ o.ETag, err = s.Put(t.Context(), key, body, o.ETag)
+ if err != nil {
+ t.Fatal(err)
+ }
+ plan15Enable(t)
+ o = updateGenerationTest(t, t.Context(), s, o, plan15Admit(o, "after-boundary", "ie"))
+ if o.Claim.Origins.Count != 1 {
+ t.Fatalf("migration seeded %d origins", o.Claim.Origins.Count)
+ }
+ plan15Proof(t, s, o, "injected", "ie", false)
+ plan15Proof(t, s, o, "after-boundary", "ie", true)
+}
+
+func TestPlan15UpdateAuthenticatesObservedHead(t *testing.T) {
+ for _, erase := range []bool{false, true} {
+ t.Run(map[bool]string{false: "forged", true: "erased"}[erase], func(t *testing.T) {
+ plan15Enable(t)
+ s := release.NewDirStore(t.TempDir())
+ o := servingClaim(t, t.Context(), s)
+ o = updateGenerationTest(t, t.Context(), s, o, plan15Admit(o, "real", "ie"))
+ stored := *o.Claim.Origins
+ if erase {
+ o.Claim.Origins = nil
+ } else {
+ o.Claim.Origins = &OriginHead{strings.Repeat("f", 64), 1}
+ t.Setenv("CARLOS_HOST_CLAIM_ORIGINS", "")
+ }
+ if _, err := Update(t.Context(), s, o, func(*Claim) error { return nil }); !errors.Is(err, ErrConflict) {
+ t.Fatalf("edited head: %v", err)
+ }
+ fresh, err := Read(t.Context(), s, o.Host)
+ if err != nil || !reflect.DeepEqual(fresh.Claim.Origins, &stored) {
+ t.Fatalf("stored head changed: %+v %v", fresh, err)
+ }
+ })
+ }
+}
+
+func TestPlan15SwitchParsingMatchesClaim(t *testing.T) {
+ for _, claim := range []string{"", " ", "1", "true", " 1", "false", "0"} {
+ for _, origins := range []string{"", " ", "1", "true", " 1", "false", "0"} {
+ t.Setenv("CARLOS_HOST_CLAIM", claim)
+ t.Setenv("CARLOS_HOST_CLAIM_ORIGINS", origins)
+ enabled := func(v string) bool { return hostwrite.Enabled(func(string) string { return v }) }
+ if got, want := originEnabled(), enabled(claim) && enabled(origins); got != want {
+ t.Errorf("%q/%q = %v want %v", claim, origins, got, want)
+ }
+ }
+ }
+}
+
+func TestPlan15ProofEarlyExit(t *testing.T) {
+ plan15Enable(t)
+ s := &plan15BoundedStore{DirStore: release.NewDirStore(t.TempDir())}
+ o := servingClaim(t, t.Context(), s)
+ for _, g := range []string{"first", "middle", "last"} {
+ o = updateGenerationTest(t, t.Context(), s, o, plan15Admit(o, g, "ie"))
+ }
+ h := o.Claim.Holder
+ last := OriginTuple{h.Incarnation, h.Location, "ie", "last"}
+ s.opens = 0
+ if got, err := AdmittedOrigins(t.Context(), s, o, []OriginTuple{last, last}); err != nil || !got[last] || s.opens != 1 {
+ t.Fatalf("head proof=%v err=%v opens=%d", got, err, s.opens)
+ }
+ s.opens = 0
+ if _, err := AdmittedOrigins(t.Context(), s, o, nil); err != nil || s.opens != 0 {
+ t.Fatalf("empty proof err=%v opens=%d", err, s.opens)
+ }
+ missing := last
+ missing.Generation = "never"
+ s.opens = 0
+ if got, err := AdmittedOrigins(t.Context(), s, o, []OriginTuple{last, missing}); err != nil || !got[last] || got[missing] || s.opens != 3 {
+ t.Fatalf("negative proof=%v err=%v opens=%d", got, err, s.opens)
+ }
+}
+
+type plan15GetOnly struct{ release.Store }
+
+func (s plan15GetOnly) Get(ctx context.Context, key string) ([]byte, string, error) {
+ if strings.Contains(key, "control/host-origins/") {
+ panic("unbounded node Get through wrapper")
+ }
+ return s.Store.Get(ctx, key)
+}
+func TestPlan15PrefixCannotBufferOrigin(t *testing.T) {
+ plan15Enable(t)
+ raw := release.NewDirStore(t.TempDir())
+ prefixed := release.NewPrefixStore(raw, "deployment")
+ o := servingClaim(t, t.Context(), prefixed)
+ o = updateGenerationTest(t, t.Context(), prefixed, o, plan15Admit(o, "real", "ie"))
+ s := release.NewPrefixStore(plan15GetOnly{raw}, "deployment")
+ h := o.Claim.Holder
+ if ok, err := AdmittedOrigin(t.Context(), s, o, h.Incarnation, h.Location, "ie", "real"); ok || err == nil {
+ t.Fatalf("buffered origin %v %v", ok, err)
+ }
+}
+
+// A comparable outer type can still hold a noncomparable dynamic value.
+type plan15Noncomparable struct {
+ *plan15BoundedStore
+ slice []byte
+}
+type plan15InterfaceWrapper struct{ release.Store }
+
+func (s plan15InterfaceWrapper) Open(ctx context.Context, key string) (io.ReadCloser, string, error) {
+ return release.OpenObject(ctx, s.Store, key)
+}
+
+func TestPlan15CacheBoundAtConstruction(t *testing.T) {
+ plan15Enable(t)
+ raw := &plan15BoundedStore{DirStore: release.NewDirStore(t.TempDir())}
+ s := plan15InterfaceWrapper{plan15Noncomparable{raw, []byte{1}}}
+ o := servingClaim(t, t.Context(), s)
+ o = updateGenerationTest(t, t.Context(), s, o, plan15Admit(o, "real", "ie"))
+ h := o.Claim.Holder
+ tuple := OriginTuple{h.Incarnation, h.Location, "ie", "real"}
+ cache := NewOriginProofCache(s)
+ raw.opens, raw.gets = 0, 0
+ for i := 0; i < 2; i++ {
+ got, err := cache.Proven(t.Context(), o, []OriginTuple{tuple})
+ if err != nil || !got[tuple] {
+ t.Fatalf("proof %v %v", got, err)
+ }
+ got[tuple] = false
+ }
+ if raw.opens != 1 || raw.gets != 2 {
+ t.Fatalf("opens=%d gets=%d", raw.opens, raw.gets)
+ }
+ // The method has no store argument: changing the constructor's variable
+ // cannot redirect this cache, and a second cache cannot inherit its result.
+ other := release.NewDirStore(t.TempDir())
+ key, _ := Key(o.Host)
+ body, _ := json.Marshal(o.Claim)
+ if _, err := other.Put(t.Context(), key, body, ""); err != nil {
+ t.Fatal(err)
+ }
+ s = plan15InterfaceWrapper{other}
+ if got, err := NewOriginProofCache(s).Proven(t.Context(), o, []OriginTuple{tuple}); got != nil || err == nil {
+ t.Fatalf("foreign store reused proof %v %v", got, err)
+ }
+ if got, err := cache.Proven(t.Context(), o, []OriginTuple{tuple}); err != nil || !got[tuple] {
+ t.Fatalf("bound store changed %v %v", got, err)
+ }
+ if got, err := new(OriginProofCache).Proven(t.Context(), o, []OriginTuple{tuple}); got != nil || !errors.Is(err, ErrMalformed) {
+ t.Fatalf("unbound cache %v %v", got, err)
+ }
+}
diff --git a/internal/hostclaim/plan15_origins_test.go b/internal/hostclaim/plan15_origins_test.go
index 5f2595a66..f66474021 100644
--- a/internal/hostclaim/plan15_origins_test.go
+++ b/internal/hostclaim/plan15_origins_test.go
@@ -63,10 +63,12 @@ func TestPlan15PruningMigrationAndScope(t *testing.T) {
})
o = updateGenerationTest(t, t.Context(), s, o, plan15Admit(o, fmt.Sprintf("new%d", i), "ie"))
}
- for _, g := range []string{"first", "old4", "old11", "new0", "new23"} {
+ for _, g := range []string{"first", "new0", "new23"} {
plan15Proof(t, s, o, g, "ie", true)
}
plan15Proof(t, s, o, "old0", "ie", false)
+ plan15Proof(t, s, o, "old4", "ie", false)
+ plan15Proof(t, s, o, "old11", "ie", false)
plan15Proof(t, s, o, "never", "ie", false)
plan15Proof(t, s, o, "first", "uk", false)
h := o.Claim.Holder
@@ -94,9 +96,13 @@ func TestPlan15PruningMigrationAndScope(t *testing.T) {
type plan15Store struct {
release.Store
failNode, failClaim, conflict, uncertain bool
- puts int
+ puts, gets int
}
+func (s *plan15Store) Get(ctx context.Context, key string) ([]byte, string, error) {
+ s.gets++
+ return s.Store.Get(ctx, key)
+}
func (s *plan15Store) Put(ctx context.Context, key string, b []byte, tag string) (string, error) {
s.puts++
if strings.HasPrefix(key, "control/host-origins/") && s.failNode {
@@ -175,8 +181,11 @@ func TestPlan15NoSelfAuthentication(t *testing.T) {
t.Fatalf("head bypass: %v", err)
}
}
- // A caller-edited migration seed is refused against the stored observation.
- fresh := servingClaim(t, t.Context(), release.NewDirStore(t.TempDir()))
+ // Only the observed admissions change; the store and ETag are current.
+ fresh, err := Read(t.Context(), raw, o.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
fresh.Claim.Holder.Admissions = append(fresh.Claim.Holder.Admissions, AdmissionRecord{Region: "ie", Generation: "invented", Mode: "cold"})
if _, err := Update(t.Context(), raw, fresh, plan15Admit(fresh, "new", "ie")); !errors.Is(err, ErrConflict) {
t.Fatalf("forged migration: %v", err)
@@ -208,8 +217,12 @@ func TestPlan15FlagOffAndWarm(t *testing.T) {
raw := release.NewDirStore(t.TempDir())
o := servingClaim(t, t.Context(), raw)
s := &plan15Store{Store: raw}
- o = updateGenerationTest(t, t.Context(), s, o, plan15Admit(o, "legacy", "ie"))
- if o.Claim.Origins != nil || s.puts != 1 {
+ var err error
+ o, err = Update(t.Context(), s, o, plan15Admit(o, "legacy", "ie"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if o.Claim.Origins != nil || s.puts != 1 || s.gets != 0 {
t.Fatal("flag-off wrote provenance")
}
b, _ := json.Marshal(o.Claim)
diff --git a/internal/hostclaim/store.go b/internal/hostclaim/store.go
index 7ede91e4d..06cd84255 100644
--- a/internal/hostclaim/store.go
+++ b/internal/hostclaim/store.go
@@ -84,6 +84,15 @@ func Update(ctx context.Context, s release.Store, obs Observed, tr func(*Claim)
return Observed{}, err
}
oldGen = obs.Claim.Gen
+ if obs.Claim.Origins != nil || originEnabled() {
+ fresh, err := Read(ctx, s, obs.Host)
+ if err != nil {
+ return Observed{}, err
+ }
+ if fresh.Absent || fresh.ETag != obs.ETag || !sameOriginHead(fresh.Claim.Origins, obs.Claim.Origins) {
+ return Observed{}, ErrConflict
+ }
+ }
}
if err := tr(next); err != nil {
return Observed{}, err
diff --git a/internal/hostops/mutation_boundary.go b/internal/hostops/mutation_boundary.go
index 28ced71d2..0d4e3661f 100644
--- a/internal/hostops/mutation_boundary.go
+++ b/internal/hostops/mutation_boundary.go
@@ -72,6 +72,12 @@ func NewObservedMutationBoundary(enabled bool, raw release.Store, check Mutation
// structuralRefusal runs before the checker and cannot be bypassed by any
// context value: canonical control objects and malformed app shapes.
func structuralRefusal(key string) error {
+ // The console never admits writers and has no origin-candidate write grant.
+ // Refuse before Put's read so a planted candidate cannot trigger buffering.
+ if strings.HasPrefix(key, "control/host-origins/") {
+ return ErrUnclassified
+ }
+
if IsControlOnlyKey(key) {
return ErrControlKey
}
diff --git a/internal/hostops/mutation_classifier.go b/internal/hostops/mutation_classifier.go
index a31f18b5d..69ae1ce2a 100644
--- a/internal/hostops/mutation_classifier.go
+++ b/internal/hostops/mutation_classifier.go
@@ -13,7 +13,6 @@ import (
"strings"
"github.com/carlosframework/platform/internal/controlname"
- "github.com/carlosframework/platform/internal/hostclaim"
"github.com/carlosframework/platform/internal/release"
)
@@ -71,11 +70,6 @@ var nonAppFamilies = []nonAppFamily{
// are never deleted (C3).
func lifecycleFamily(ctx context.Context, raw release.Store, w *ControlWriter, t MutationTarget) (string, bool, error) {
switch {
- case strings.HasPrefix(t.Key, "control/host-origins/"):
- if t.Method != "Put" || t.ETag != "" || len(t.Before) != 0 || hostclaim.ValidateOriginObject(t.Key, t.After) != nil {
- return "", true, ErrIdentity
- }
- return "lifecycle-origin-candidate", true, nil
case strings.HasPrefix(t.Key, "control/hosts/") && strings.HasSuffix(t.Key, ".json") && strings.Count(t.Key, "/") == 2:
host := strings.TrimSuffix(strings.TrimPrefix(t.Key, "control/hosts/"), ".json")
if t.Method != "Put" || !controlname.ValidHostShape(host) {
diff --git a/internal/hostops/plan15_origins_test.go b/internal/hostops/plan15_origins_test.go
index 8a58456c1..702579cf3 100644
--- a/internal/hostops/plan15_origins_test.go
+++ b/internal/hostops/plan15_origins_test.go
@@ -3,6 +3,7 @@ package hostops
import (
"crypto/sha256"
"encoding/hex"
+ "errors"
"strings"
"testing"
@@ -17,24 +18,20 @@ func TestPlan15OriginCandidateBoundary(t *testing.T) {
for _, tc := range []struct {
name string
target MutationTarget
- ok bool
}{
- {"candidate", MutationTarget{Method: "Put", Key: key, After: body}, true},
- {"overwrite", MutationTarget{Method: "Put", Key: key, After: body, ETag: "v"}, false},
- {"delete", MutationTarget{Method: "Delete", Key: key}, false},
- {"wrong host", MutationTarget{Method: "Put", Key: strings.Replace(key, "x.example", "other.example", 1), After: body}, false},
- {"wrong digest", MutationTarget{Method: "Put", Key: key, After: append(append([]byte(nil), body...), ' ')}, false},
- {"unknown field", MutationTarget{Method: "Put", Key: key, After: []byte(`{"bogus":true}`)}, false},
- {"oversize", MutationTarget{Method: "Put", Key: key, After: make([]byte, 4097)}, false},
+ {"candidate", MutationTarget{Method: "Put", Key: key, After: body}},
+ {"overwrite", MutationTarget{Method: "Put", Key: key, After: body, ETag: "v"}},
+ {"delete", MutationTarget{Method: "Delete", Key: key}},
+ {"wrong host", MutationTarget{Method: "Put", Key: strings.Replace(key, "x.example", "other.example", 1), After: body}},
+ {"wrong digest", MutationTarget{Method: "Put", Key: key, After: append(append([]byte(nil), body...), ' ')}},
+ {"unknown field", MutationTarget{Method: "Put", Key: key, After: []byte(`{"bogus":true}`)}},
+ {"oversize", MutationTarget{Method: "Put", Key: key, After: make([]byte, 4097)}},
} {
t.Run(tc.name, func(t *testing.T) {
c, err := ClassifyMutation(t.Context(), raw, w, tc.target)
- if (err == nil) != tc.ok {
+ if !errors.Is(err, ErrUnclassified) {
t.Fatalf("classification %+v %v", c, err)
}
- if tc.ok && (c.Family != "lifecycle-origin-candidate" || len(c.Scopes) != 0 || c.NonApp == "") {
- t.Fatalf("bad class %+v", c)
- }
})
}
}
diff --git a/internal/hostops/plan15_stream_test.go b/internal/hostops/plan15_stream_test.go
new file mode 100644
index 000000000..ba1465505
--- /dev/null
+++ b/internal/hostops/plan15_stream_test.go
@@ -0,0 +1,99 @@
+package hostops
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "strings"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+type plan15StreamStore struct {
+ release.Store
+ bytes, closes, opens, puts int
+}
+
+func (s *plan15StreamStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ if strings.Contains(key, "control/host-origins/") {
+ panic("unbounded origin Get")
+ }
+ return s.Store.Get(ctx, strings.TrimPrefix(key, "deployment/"))
+}
+func (s *plan15StreamStore) Put(ctx context.Context, key string, b []byte, tag string) (string, error) {
+ s.puts++
+ return "", release.ErrCASConflict
+}
+func (s *plan15StreamStore) Open(context.Context, string) (io.ReadCloser, string, error) {
+ s.opens++
+ return plan15StreamReader{s, strings.NewReader(strings.Repeat("x", 8192))}, "", nil
+}
+
+type plan15StreamReader struct {
+ s *plan15StreamStore
+ r io.Reader
+}
+
+func (r plan15StreamReader) Read(b []byte) (int, error) {
+ n, e := r.r.Read(b)
+ r.s.bytes += n
+ return n, e
+}
+func (r plan15StreamReader) Close() error { r.s.closes++; return nil }
+
+type plan15NoOpen struct{ release.Store }
+
+func TestPlan15StreamingCompositions(t *testing.T) {
+ for _, stream := range []bool{false, true} {
+ for _, boundary := range []bool{false, true} {
+ t.Run(fmt.Sprintf("stream=%v/boundary=%v", stream, boundary), func(t *testing.T) {
+ w, o, _ := wakeResetFixture(t)
+ o.Claim.Origins = &hostclaim.OriginHead{Digest: strings.Repeat("a", 64), Count: 1}
+ key, _ := hostclaim.Key(o.Host)
+ body, _ := json.Marshal(o.Claim)
+ var err error
+ o.ETag, err = w.raw.Put(t.Context(), key, body, o.ETag)
+ if err != nil {
+ t.Fatal(err)
+ }
+ raw := &plan15StreamStore{Store: w.raw}
+ var s release.Store = raw
+ if !stream {
+ s = plan15NoOpen{s}
+ }
+ s = release.NewPrefixStore(s, "deployment")
+ s = conditionalRecordStore{raw: s}
+ if boundary {
+ s = &mutationBoundary{raw: s, check: func(context.Context, MutationTarget) error { return nil }}
+ }
+ h := o.Claim.Holder
+ if ok, err := hostclaim.AdmittedOrigin(t.Context(), s, o, h.Incarnation, h.Location, "ie", "real"); ok || !errors.Is(err, hostclaim.ErrMalformed) {
+ t.Fatalf("proof %v %v", ok, err)
+ }
+ if stream && (raw.bytes != 4097 || raw.opens != 1 || raw.closes != 1) {
+ t.Fatalf("bytes=%d opens=%d closes=%d", raw.bytes, raw.opens, raw.closes)
+ }
+ if !stream && (raw.bytes != 0 || raw.opens != 0) {
+ t.Fatal("nonstreaming store opened")
+ }
+ })
+ }
+ }
+}
+func TestPlan15BoundaryRefusesCandidateBeforeRead(t *testing.T) {
+ for _, flag := range []string{"", "1"} {
+ t.Setenv("CARLOS_HOST_CLAIM_ORIGINS", flag)
+ raw := &plan15StreamStore{}
+ s := &mutationBoundary{raw: conditionalRecordStore{raw: release.NewPrefixStore(raw, "deployment")}, check: func(context.Context, MutationTarget) error { t.Fatal("origin reached grant checker"); return nil }}
+ if _, err := s.Put(t.Context(), "control/host-origins/x.example/node.json", []byte("candidate"), ""); !errors.Is(err, ErrUnclassified) {
+ t.Fatal(err)
+ }
+ if raw.puts != 0 || raw.opens != 0 {
+ t.Fatal("candidate touched raw store")
+ }
+ }
+}
diff --git a/internal/provision/plan9_b8_census_test.go b/internal/provision/plan9_b8_census_test.go
index b762f324a..e97ca2eca 100644
--- a/internal/provision/plan9_b8_census_test.go
+++ b/internal/provision/plan9_b8_census_test.go
@@ -755,8 +755,8 @@ var plan9B8CensusInventory = []censusEntry{
{"internal/hostclaim/lossrecord.go", "CreateLossRecord", "foreign-recordkey", "key:LossRecordKey,sel:Put", "e62263731fa572ed6f195a267abb63ff5a1bf44284db3db76a819ccaec238191"},
{"internal/hostclaim/lossrecord.go", "ReplaceLossRecord", "foreign-recordkey", "key:LossRecordKey,sel:Put", "576589457e4a18b64e6b56892603051982bd01770c355504305e72d53149bc52"},
{"internal/hostclaim/origins.go", "originTuple.valid", "key-site", "lit:instances", "7f6f20f51e51694d665588f0b986def85bda4789f24623ee1993289265b01355"},
- {"internal/hostclaim/origins.go", "commitOrigins", "nonrecord", "sel:Put", "f5f506d8b1e5d4dced896c6be3a6e498354e5f28d81452131e220fec58e8bd78"},
- {"internal/hostclaim/store.go", "Update", "nonrecord", "sel:Put", "ec5de8702797ce97a24147cbf16ad8f8e254e2da82d3a0859c0407501f0b575f"},
+ {"internal/hostclaim/origins.go", "commitOrigins", "nonrecord", "sel:Put", "2d3a27e56510db9ec2d54d85b354f72b72da1080c7b4182535040f2d30e36888"},
+ {"internal/hostclaim/store.go", "Update", "nonrecord", "sel:Put", "083ddd3e076a07db2b5fca70a9b56fa022782deafb9ad56c77f8f32515633906"},
{"internal/hostevidence/signature.go", "immutable", "nonrecord", "sel:Put", "54fb2bdc230c2cb4193f3b808004b63bcfd092266846fc0c03e61617c3ad0ab5"},
{"internal/hostexec/controller.go", "Controller.run", "nonrecord", "sel:Put", "03df11b375761730700ce30c0db05c57f0828629eda0a5de126ea1a2f2f7501c"},
{"internal/hostfence/termination.go", "TerminationWorker.persist", "nonrecord", "sel:Put", "47758114adcc3c493a9d1ead0fa7087ad0d1e49595c891bab4dfb51c9058a8a1"},
@@ -788,7 +788,7 @@ var plan9B8CensusInventory = []censusEntry{
{"internal/hostops/mutation_boundary.go", "mutationBoundary.Put", "record-transport", "def:Put,sel:Put,str:Put", "c4222199cdbfa8cce1f97f91bf41fe94857f4d833e30b17636c40f1aae39ff67"},
{"internal/hostops/mutation_boundary_putfrom.go", "mutationBoundary.PutFrom", "record-transport", "def:PutFrom,sel:Put,sel:PutObjectFrom,str:Put", "8d015a924901f0c41e142181f986cb9d2a3bdfa6a9e078dfda05a2d1e7cf9b24"},
{"internal/hostops/mutation_boundary_putfrom.go", "conditionalRecordStore.PutFrom", "record-transport", "def:PutFrom,sel:Put,sel:PutObjectFrom", "3c3cb33a601eab5c4d74042a7ca8baf411c0549508b5a8273ef88f4ee0e9636c"},
- {"internal/hostops/mutation_classifier.go", "lifecycleFamily", "reference", "str:Put", "f43b261b71020751372753c17d0de3b5ea51866a9c2c08e7b514c38d775e6f90"},
+ {"internal/hostops/mutation_classifier.go", "lifecycleFamily", "reference", "str:Put", "75df2a3966d0bfe4b4b367617de3ebe28055a38ef4738f352859ba94ddbff241"},
// Classifies mutation keys and method names without executing a store write.
{"internal/hostops/mutation_classifier.go", "ClassifyMutation", "reference", "str:Put", "2d22e1d4a99a69ebff6f033a667f59c778855a66828ee5c7646a6b1d18573b65"},
{"internal/hostops/mutation_classifier.go", "ownerUnion", "reference", "str:Put", "0c35233653dfe9ca6cc78bc54bac2c6991f82c9a7c2fd27c228c125f099f489b"},
diff --git a/internal/release/store.go b/internal/release/store.go
index 1c8af98bf..ff9fe870e 100644
--- a/internal/release/store.go
+++ b/internal/release/store.go
@@ -281,11 +281,22 @@ func PutObjectFrom(ctx context.Context, s Store, key string, body io.ReadSeeker,
return s.Put(ctx, key, b, ifMatch)
}
+type streamingReadKey struct{}
+
+// OpenObjectStreaming refuses Get fallbacks, including inside wrappers that
+// forward this context through OpenObject. Callers must still bound their reads.
+func OpenObjectStreaming(ctx context.Context, s Store, key string) (io.ReadCloser, string, error) {
+ return OpenObject(context.WithValue(ctx, streamingReadKey{}, true), s, key)
+}
+
// OpenObject streams key from s, buffering through Get when s cannot stream.
func OpenObject(ctx context.Context, s Store, key string) (io.ReadCloser, string, error) {
if o, ok := s.(Opener); ok {
return o.Open(ctx, key)
}
+ if required, _ := ctx.Value(streamingReadKey{}).(bool); required {
+ return nil, "", errors.ErrUnsupported
+ }
b, etag, err := s.Get(ctx, key)
if err != nil {
return nil, "", err
diff --git a/internal/release/stream_required_test.go b/internal/release/stream_required_test.go
new file mode 100644
index 000000000..2403d7d8a
--- /dev/null
+++ b/internal/release/stream_required_test.go
@@ -0,0 +1,38 @@
+package release
+
+import (
+ "context"
+ "errors"
+ "io"
+ "testing"
+)
+
+type requiredStreamGetOnly struct {
+ Store
+ gets int
+}
+
+func (s *requiredStreamGetOnly) Get(ctx context.Context, key string) ([]byte, string, error) {
+ s.gets++
+ return s.Store.Get(ctx, key)
+}
+func TestOpenObjectStreamingRequiresWholeWrapperChain(t *testing.T) {
+ raw := NewDirStore(t.TempDir())
+ if _, err := raw.Put(t.Context(), "outer/inner/object", []byte("small"), ""); err != nil {
+ t.Fatal(err)
+ }
+ base := &requiredStreamGetOnly{Store: raw}
+ s := NewPrefixStore(NewPrefixStore(base, "outer"), "inner")
+ if r, _, err := OpenObjectStreaming(t.Context(), s, "object"); r != nil || !errors.Is(err, errors.ErrUnsupported) || base.gets != 0 {
+ t.Fatalf("stream=%v err=%v gets=%d", r, err, base.gets)
+ }
+ r, _, err := OpenObject(t.Context(), s, "object")
+ if err != nil {
+ t.Fatal(err)
+ }
+ body, err := io.ReadAll(r)
+ r.Close()
+ if err != nil || string(body) != "small" || base.gets != 1 {
+ t.Fatalf("fallback %q %v gets=%d", body, err, base.gets)
+ }
+}