amadan

carlos / platform Public

feat/claim-admitted-provenance

Squashed from feat/claim-admitted-provenance into main.
Co-Authored-By: Paul Campbell <paul@tito.io>
codex pushed by claude 01edf9015b55218ef970eafa54cff92c35c2dfdb parent d612253
17 files changed, +1450 −3
  • internal/console/plan15_stream_test.go +84 −0
  • internal/host/claimoriginproof.go +47 −0
  • internal/host/plan15_originproof_test.go +67 −0
  • internal/hostclaim/claim.go +6 −0
  • internal/hostclaim/generation.go +3 −0
  • internal/hostclaim/origins.go +383 −0
  • internal/hostclaim/plan15_batch_test.go +148 −0
  • internal/hostclaim/plan15_fix1_test.go +171 −0
  • internal/hostclaim/plan15_origins_test.go +308 −0
  • internal/hostclaim/store.go +12 −0
  • internal/hostops/lifecycle_grant.go +4 −1
  • internal/hostops/mutation_boundary.go +6 −0
  • internal/hostops/plan15_origins_test.go +59 −0
  • internal/hostops/plan15_stream_test.go +99 −0
  • internal/provision/plan9_b8_census_test.go +4 −2
  • 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	334.723s
?   	github.com/carlosframework/platform/cmd/changelog	[no test files]
ok  	github.com/carlosframework/platform/cmd/console	3.019s
ok  	github.com/carlosframework/platform/cmd/moveprobe	92.219s
ok  	github.com/carlosframework/platform/internal/activator	239.202s
ok  	github.com/carlosframework/platform/internal/agentsock	0.017s
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.349s
ok  	github.com/carlosframework/platform/internal/apperr	0.073s
ok  	github.com/carlosframework/platform/internal/applog	0.142s
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.576s
?   	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.010s
ok  	github.com/carlosframework/platform/internal/caps	0.014s
ok  	github.com/carlosframework/platform/internal/certwarm	0.004s
ok  	github.com/carlosframework/platform/internal/cgroup	0.206s
ok  	github.com/carlosframework/platform/internal/changelog	0.269s
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	360.484s
?   	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.522s
ok  	github.com/carlosframework/platform/internal/dbdoor/dbdoortest	0.009s
ok  	github.com/carlosframework/platform/internal/dbverify	0.008s
ok  	github.com/carlosframework/platform/internal/dnsrec	0.008s
ok  	github.com/carlosframework/platform/internal/docsite	0.008s
ok  	github.com/carlosframework/platform/internal/domains	2.045s
ok  	github.com/carlosframework/platform/internal/doorkey	0.008s
ok  	github.com/carlosframework/platform/internal/economics	0.008s
ok  	github.com/carlosframework/platform/internal/edge	25.962s
?   	github.com/carlosframework/platform/internal/edgeproof	[no test files]
ok  	github.com/carlosframework/platform/internal/edgesfile	0.008s
ok  	github.com/carlosframework/platform/internal/fleetapi	0.140s
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.011s
ok  	github.com/carlosframework/platform/internal/geoip	0.035s
ok  	github.com/carlosframework/platform/internal/host	61.194s
ok  	github.com/carlosframework/platform/internal/hostactors	0.261s
ok  	github.com/carlosframework/platform/internal/hostbox	1.273s
ok  	github.com/carlosframework/platform/internal/hostclaim	0.309s
ok  	github.com/carlosframework/platform/internal/hostevidence	0.014s
ok  	github.com/carlosframework/platform/internal/hostexec	0.319s
ok  	github.com/carlosframework/platform/internal/hostfence	0.031s
ok  	github.com/carlosframework/platform/internal/hostfiles	4.212s
ok  	github.com/carlosframework/platform/internal/hostfloor	0.018s
ok  	github.com/carlosframework/platform/internal/hostguard	0.143s
ok  	github.com/carlosframework/platform/internal/hostops	1.373s
?   	github.com/carlosframework/platform/internal/hostops/fencerelease	[no test files]
ok  	github.com/carlosframework/platform/internal/hostops/internal/fencerelease	0.012s
ok  	github.com/carlosframework/platform/internal/hostquarantine	0.002s
ok  	github.com/carlosframework/platform/internal/hostresolve	0.131s
ok  	github.com/carlosframework/platform/internal/hostwrite	0.003s
ok  	github.com/carlosframework/platform/internal/infraview	0.003s
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.004s
ok  	github.com/carlosframework/platform/internal/instancekey	0.003s
ok  	github.com/carlosframework/platform/internal/instanceupdates	0.009s
ok  	github.com/carlosframework/platform/internal/layout	1.381s
ok  	github.com/carlosframework/platform/internal/lease	9.434s
ok  	github.com/carlosframework/platform/internal/lease/leasetest	0.006s
ok  	github.com/carlosframework/platform/internal/ledger	0.013s
ok  	github.com/carlosframework/platform/internal/mail	0.154s
ok  	github.com/carlosframework/platform/internal/maintenance	0.079s
ok  	github.com/carlosframework/platform/internal/maintenanceexec	5.529s
ok  	github.com/carlosframework/platform/internal/maintenanceguard	0.099s
ok  	github.com/carlosframework/platform/internal/metrics	0.007s
ok  	github.com/carlosframework/platform/internal/moves	21.057s
ok  	github.com/carlosframework/platform/internal/moves/archive	0.733s
ok  	github.com/carlosframework/platform/internal/objstore	0.165s
ok  	github.com/carlosframework/platform/internal/ops	0.160s
ok  	github.com/carlosframework/platform/internal/park	0.194s
ok  	github.com/carlosframework/platform/internal/privatefile	0.004s
ok  	github.com/carlosframework/platform/internal/provision	3.226s
ok  	github.com/carlosframework/platform/internal/publicrelease	0.010s
?   	github.com/carlosframework/platform/internal/regionname	[no test files]
ok  	github.com/carlosframework/platform/internal/registry	3.830s
ok  	github.com/carlosframework/platform/internal/relay	7.224s
ok  	github.com/carlosframework/platform/internal/release	19.623s
ok  	github.com/carlosframework/platform/internal/rollout	3.254s
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.249s
ok  	github.com/carlosframework/platform/internal/sim/hostlife	2.386s
ok  	github.com/carlosframework/platform/internal/sim/litestreamfacts	4.031s
ok  	github.com/carlosframework/platform/internal/status	0.038s
ok  	github.com/carlosframework/platform/internal/steering	0.006s
ok  	github.com/carlosframework/platform/internal/stopevidence	0.100s
ok  	github.com/carlosframework/platform/internal/svcrole	0.002s
ok  	github.com/carlosframework/platform/internal/telemetry	0.006s
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.002s
ok  	github.com/carlosframework/platform/internal/updatewords	0.002s
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.554s
=== 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.023s
=== RUN   TestT4StorageProbeMinio
2026/10/12 23:15:48 moves: T4: STORE: list: operation error S3: ListObjectsV2, https response error StatusCode: 403, RequestID: 18DDEB4E4C95EC44, 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.021s
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
new file mode 100644
index 000000000..ff5276087
--- /dev/null
+++ b/internal/host/claimoriginproof.go
@@ -0,0 +1,47 @@
+package host
+
+import (
+ "context"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/regionname"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// ClaimOriginProof implements moves' Proven contract. HomeRegion must come from
+// the verified roster/placement binding, and Claims is the deployment store.
+// Admission.NewGeneration is deliberately not evidence of admission.
+type ClaimOriginProof struct {
+ Claims release.Store
+ HomeRegion, Region string
+}
+
+func (p ClaimOriginProof) Proven(ctx context.Context, o hostclaim.Observed, _ hostclaim.Admission, generations []string) (map[string]bool, error) {
+ if o.Absent || o.Claim == nil || o.Claim.Holder == nil || p.Region == "" {
+ return nil, hostclaim.ErrWrongState
+ }
+ h := o.Claim.Holder
+ regions := []string{p.Region}
+ if p.HomeRegion != "" && regionname.PlacementRegion(p.HomeRegion) == p.HomeRegion && regionname.SameRegion(p.HomeRegion, p.Region, regionname.HomeRegion) {
+ if p.Region == regionname.HomeRegion {
+ regions = append(regions, p.HomeRegion)
+ } else {
+ regions = append(regions, regionname.HomeRegion)
+ }
+ }
+ set := make([]hostclaim.OriginTuple, 0, len(regions)*len(generations))
+ for _, generation := range generations {
+ for _, region := range regions {
+ set = append(set, hostclaim.OriginTuple{Incarnation: h.Incarnation, Location: h.Location, Region: region, Generation: generation})
+ }
+ }
+ proven, err := hostclaim.AdmittedOrigins(ctx, p.Claims, o, set)
+ if err != nil {
+ return nil, err
+ }
+ result := make(map[string]bool, len(generations))
+ for _, tuple := range set {
+ result[tuple.Generation] = result[tuple.Generation] || proven[tuple]
+ }
+ return result, nil
+}
diff --git a/internal/host/plan15_originproof_test.go b/internal/host/plan15_originproof_test.go
new file mode 100644
index 000000000..62858930f
--- /dev/null
+++ b/internal/host/plan15_originproof_test.go
@@ -0,0 +1,67 @@
+package host
+
+import (
+ "context"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+var _ interface {
+ Proven(context.Context, hostclaim.Observed, hostclaim.Admission, []string) (map[string]bool, error)
+} = ClaimOriginProof{}
+
+// The moves session will adopt the batch signature after Plan 15 lands.
+// This test-only bridge checks compatibility with its current singleton API.
+type plan15SingleOriginAdapter struct{ ClaimOriginProof }
+
+var _ movesOriginProof = plan15SingleOriginAdapter{}
+
+func (p plan15SingleOriginAdapter) Proven(ctx context.Context, o hostclaim.Observed, in hostclaim.Admission, generation string) (bool, error) {
+ proven, err := p.ClaimOriginProof.Proven(ctx, o, in, []string{generation})
+ return proven[generation], err
+}
+
+func TestPlan15MovesProof(t *testing.T) {
+ t.Setenv("CARLOS_HOST_CLAIM", "1")
+ t.Setenv("CARLOS_HOST_CLAIM_ORIGINS", "1")
+ s := release.NewDirStore(t.TempDir())
+ host := "test.example"
+ o, err := hostclaim.Read(t.Context(), s, host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ n := hostclaim.NewHolder{Family: "instance", Key: "apps/bab/app/instances/test.example.json", Account: "bab", App: "app", Host: host, Region: "ie", CreatedAt: time.Date(2026, 10, 12, 0, 0, 0, 0, time.UTC)}
+ o, err = hostclaim.Update(t.Context(), s, o, hostclaim.BeginCreate(n, []byte(`{"host":"test.example","account":"bab","app":"app","created_at":"2026-10-12T00:00:00Z"}`), "create", "console", time.Now()))
+ if err != nil {
+ t.Fatal(err)
+ }
+ o, err = hostclaim.Update(t.Context(), s, o, hostclaim.CompleteCreate("create", "box"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ o, err = hostclaim.Update(t.Context(), s, o, hostclaim.AdmitWriter(hostclaim.Admission{Box: "box", Mode: "cold", Region: "ie", NewGeneration: "admitted"}))
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, tc := range []struct {
+ home, region, generation string
+ want bool
+ }{
+ {"", "ie", "admitted", true}, {"ie", "@home", "admitted", true}, {"", "@home", "admitted", false},
+ {"ie", "uk", "admitted", false}, {"", "ie", "prepared", false},
+ } {
+ p := ClaimOriginProof{Claims: s, HomeRegion: tc.home, Region: tc.region}
+ got, err := p.Proven(t.Context(), o, hostclaim.Admission{NewGeneration: tc.generation}, []string{tc.generation})
+ if got[tc.generation] != tc.want || err != nil {
+ t.Fatalf("%+v: %v %v", tc, got, err)
+ }
+ var adapter movesOriginProof = plan15SingleOriginAdapter{p}
+ one, err := adapter.Proven(t.Context(), o, hostclaim.Admission{NewGeneration: tc.generation}, tc.generation)
+ if one != tc.want || err != nil {
+ t.Fatalf("singleton adapter %+v: %v %v", tc, one, err)
+ }
+ }
+}
diff --git a/internal/hostclaim/claim.go b/internal/hostclaim/claim.go
index 86fc8b95e..56ff9abdf 100644
--- a/internal/hostclaim/claim.go
+++ b/internal/hostclaim/claim.go
@@ -155,6 +155,9 @@ type CancelledCreate struct {
}
type Claim struct {
+ Origins *OriginHead `json:"admitted_origins,omitempty"`
+ originAdmission *originTuple
+
RetiredLocations map[string]string `json:"retired_locations,omitempty"`
EvidenceRefs []AutomaticLoss `json:"evidence_refs,omitempty"`
V int `json:"v"`
@@ -225,6 +228,9 @@ func isHex64(s string) bool {
}
func (c *Claim) Validate() error {
+ if c != nil && !originHeadValid(c.Origins) {
+ return ErrMalformed
+ }
if c == nil || c.V != schemaVersion || !controlname.ValidHostShape(c.Host) || c.Gen == 0 {
return fmt.Errorf("%w: header", ErrMalformed)
}
diff --git a/internal/hostclaim/generation.go b/internal/hostclaim/generation.go
index 351462b6b..c58c4469a 100644
--- a/internal/hostclaim/generation.go
+++ b/internal/hostclaim/generation.go
@@ -163,6 +163,9 @@ func AdmitWriter(in Admission) func(*Claim) error {
entry.WritesLost = &WritesLost{Box: predecessor.Box, Generation: predecessor.Generation, Session: predecessor.Session, RemoteMaxTXID: in.RemoteMaxTXID}
}
}
+ if err := pendingOrigin(c, entry); err != nil {
+ return err
+ }
h.Admissions = append(h.Admissions, entry)
for len(h.Admissions) > 8 && h.Admissions[0].Covered && admissionLossArchived(h.Admissions[0]) {
h.Admissions = h.Admissions[1:]
diff --git a/internal/hostclaim/origins.go b/internal/hostclaim/origins.go
new file mode 100644
index 000000000..c45b14f19
--- /dev/null
+++ b/internal/hostclaim/origins.go
@@ -0,0 +1,383 @@
+package hostclaim
+
+import (
+ "bytes"
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "io"
+ "os"
+ "sort"
+ "strings"
+ "sync"
+
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+const originNodeLimit = 4096
+const originCountLimit = 1 << 20
+
+// OriginHead authenticates only nodes reachable from this claim's committed head.
+type OriginHead struct {
+ Digest string `json:"digest"`
+ Count uint64 `json:"count"`
+}
+
+type OriginTuple struct {
+ Incarnation string `json:"incarnation"`
+ Location string `json:"location"`
+ Region string `json:"region"`
+ Generation string `json:"generation"`
+}
+
+type originTuple = OriginTuple
+
+type originNode struct {
+ V int `json:"v"`
+ Host string `json:"host"`
+ Count uint64 `json:"count"`
+ Previous string `json:"previous,omitempty"`
+ Origin originTuple `json:"origin"`
+}
+
+func originEnabled() bool {
+ return hostwrite.Enabled(os.Getenv) && hostwrite.Enabled(func(string) string {
+ return os.Getenv("CARLOS_HOST_CLAIM_ORIGINS")
+ })
+}
+
+func originDigest(b []byte) string {
+ h := sha256.New()
+ h.Write([]byte("carlos-admitted-origin-v1\x00"))
+ h.Write(b)
+ return hex.EncodeToString(h.Sum(nil))
+}
+
+func originKey(host, digest string) string {
+ return "control/host-origins/" + host + "/" + digest + ".json"
+}
+
+func originHeadValid(h *OriginHead) bool {
+ return h == nil || h.Count > 0 && h.Count <= originCountLimit && isHex64(h.Digest) && h.Digest == strings.ToLower(h.Digest)
+}
+
+func (o originTuple) valid(host string) bool {
+ if !isHex64(o.Incarnation) || o.Incarnation != strings.ToLower(o.Incarnation) ||
+ o.Region == "" || len(o.Region) > 64 || strings.ContainsAny(o.Region, "/\\\x00") ||
+ o.Generation == "" || len(o.Generation) > 256 || o.Generation == RootGeneration ||
+ o.Generation == "." || o.Generation == ".." || strings.ContainsAny(o.Generation, "/\\\x00") {
+ return false
+ }
+ return o.Location == "instances/"+host || o.Location == "instances/"+host+"/"+o.Incarnation
+}
+
+// ValidateOriginObject classifies inert, content-addressed candidates. Creating a
+// candidate grants nothing; only the claim CAS can make one reachable.
+func ValidateOriginObject(key string, body []byte) error {
+ if len(body) == 0 || len(body) > originNodeLimit {
+ return ErrMalformed
+ }
+ d := json.NewDecoder(bytes.NewReader(body))
+ d.DisallowUnknownFields()
+ var n originNode
+ if err := d.Decode(&n); err != nil {
+ return ErrMalformed
+ }
+ if _, err := d.Token(); err != io.EOF {
+ return ErrMalformed
+ }
+ if _, err := Key(n.Host); err != nil {
+ return ErrMalformed
+ }
+ if n.V != 1 || n.Count == 0 || n.Count > originCountLimit || !n.Origin.valid(n.Host) ||
+ (n.Count == 1) != (n.Previous == "") ||
+ n.Previous != "" && (!isHex64(n.Previous) || n.Previous != strings.ToLower(n.Previous)) ||
+ key != originKey(n.Host, originDigest(body)) {
+ return ErrMalformed
+ }
+ return nil
+}
+
+func readOrigin(ctx context.Context, s release.Store, host string, h OriginHead) (originNode, error) {
+ var n originNode
+ if err := ctx.Err(); err != nil {
+ return n, err
+ }
+ key := originKey(host, h.Digest)
+ b, err := readOriginBytes(ctx, s, key)
+ if err != nil {
+ return n, err
+ }
+ if err = ValidateOriginObject(key, b); err != nil {
+ return n, err
+ }
+ if err = json.Unmarshal(b, &n); err != nil {
+ return n, err
+ }
+ if n.Host != host || n.Count != h.Count {
+ return n, ErrMalformed
+ }
+ return n, nil
+}
+
+func pendingOrigin(c *Claim, entry AdmissionRecord) error {
+ if !originEnabled() {
+ // Once written, provenance cannot silently stop collecting admissions.
+ if c.Origins != nil {
+ return ErrWrongState
+ }
+ return nil
+ }
+ if c.originAdmission != nil || c.Holder == nil {
+ return ErrWrongState
+ }
+ o := originTuple{c.Holder.Incarnation, c.Holder.Location, entry.Region, entry.Generation}
+ // The empty legacy root has no files/g-<generation>/ origin.
+ if entry.Generation == "" {
+ return nil
+ }
+ if !o.valid(c.Host) {
+ return ErrWrongState
+ }
+ c.originAdmission = &o
+ return nil
+}
+
+// commitOrigins stages immutable candidates before Update's existing claim CAS.
+// A failed or uncertain CAS leaves candidates without granting them authority.
+func commitOrigins(ctx context.Context, s release.Store, obs Observed, next *Claim) error {
+ var old *OriginHead
+ if obs.Claim != nil {
+ old = obs.Claim.Origins
+ }
+ if !sameOriginHead(old, next.Origins) {
+ return ErrWrongState
+ }
+ if next.originAdmission == nil {
+ return nil
+ }
+ if obs.Absent || obs.Claim == nil || obs.Claim.Holder == nil || next.Holder == nil {
+ return ErrWrongState
+ }
+ h := next.Holder
+ want := *next.originAdmission
+ if want.Incarnation != h.Incarnation || want.Location != h.Location || want.Generation != h.Generation ||
+ h.Session != next.Gen || len(h.Admissions) == 0 {
+ return ErrWrongState
+ }
+ a := h.Admissions[len(h.Admissions)-1]
+ if a.Session != h.Session || a.Generation != want.Generation || a.Region != want.Region {
+ return ErrWrongState
+ }
+ // Bind admission to the stored observation, never a caller-edited admission list.
+ fresh, err := Read(ctx, s, obs.Host)
+ if err != nil {
+ return err
+ }
+ before, _ := json.Marshal(obs.Claim)
+ stored, _ := json.Marshal(fresh.Claim)
+ if fresh.Absent || fresh.ETag != obs.ETag || !bytes.Equal(before, stored) {
+ return ErrConflict
+ }
+ // 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
+ last, err := readOrigin(ctx, s, obs.Host, head)
+ if err != nil {
+ return err
+ }
+ if last.Origin == want {
+ next.originAdmission = nil
+ return nil
+ }
+ }
+ 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
+ }
+ 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
+}
+
+func sameOriginHead(a, b *OriginHead) bool {
+ if a == nil || b == nil {
+ return a == nil && b == nil
+ }
+ return *a == *b
+}
+
+// OriginProofCache retains one exact batch, after authenticating its observation.
+// Construct it with NewOriginProofCache; its store cannot be replaced.
+type OriginProofCache struct {
+ mu sync.Mutex
+ store release.Store
+ key string
+ 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
+ }
+ // 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
+ }
+ if err != nil {
+ return nil, err
+ }
+ b, readErr := io.ReadAll(io.LimitReader(reader, originNodeLimit+1))
+ closeErr := reader.Close()
+ if readErr != nil {
+ return nil, readErr
+ }
+ if len(b) > originNodeLimit {
+ return nil, ErrMalformed
+ }
+ if closeErr != nil {
+ return nil, closeErr
+ }
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+ return b, nil
+}
+
+// AdmittedOrigins authenticates once and proves the complete requested set in
+// 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 NewOriginProofCache(s).Proven(ctx, obs, set)
+}
+
+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.key, cache.result = "", nil }
+ if err := ctx.Err(); err != nil {
+ invalidate()
+ return nil, err
+ }
+ if s == nil || obs.Absent || obs.Claim == nil || obs.ETag == "" || obs.Host != obs.Claim.Host {
+ invalidate()
+ return nil, ErrMalformed
+ }
+ wanted := make(map[OriginTuple]bool, len(set))
+ for _, tuple := range set {
+ if !tuple.valid(obs.Host) {
+ invalidate()
+ return nil, ErrMalformed
+ }
+ wanted[tuple] = false
+ }
+ fresh, err := Read(ctx, s, obs.Host)
+ if err != nil {
+ invalidate()
+ return nil, err
+ }
+ before, marshalErr := json.Marshal(obs.Claim)
+ stored, storedErr := json.Marshal(fresh.Claim)
+ if fresh.Absent || fresh.ETag != obs.ETag || marshalErr != nil || storedErr != nil || !bytes.Equal(before, stored) {
+ invalidate()
+ return nil, ErrConflict
+ }
+ if err := ctx.Err(); err != nil {
+ invalidate()
+ return nil, err
+ }
+ // 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)
+ tuples = append(tuples, string(b))
+ }
+ sort.Strings(tuples)
+ keyBytes, _ := json.Marshal(struct {
+ Host, ETag, Claim string
+ Set []string
+ }{obs.Host, obs.ETag, string(stored), tuples})
+ key := string(keyBytes)
+ if cache.key == key {
+ for tuple := range wanted {
+ wanted[tuple] = cache.result[tuple]
+ }
+ return wanted, nil
+ }
+ invalidate()
+ c := fresh.Claim
+ // The chain starts empty: a claim with no recorded head proves nothing.
+ // Retained admissions are never evidence (a box can write them), so live
+ // origins are proven only through admissions recorded after the boundary
+ // or the #250 files backfill into an admitted prefix.
+ if c.Origins != nil {
+ 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 proven, ok := wanted[node.Origin]; ok && !proven {
+ remaining--
+ wanted[node.Origin] = true
+ }
+ head = OriginHead{Digest: node.Previous, Count: head.Count - 1}
+ }
+ }
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+ cache.key = key
+ cache.result = make(map[OriginTuple]bool, len(wanted))
+ for tuple, proven := range wanted {
+ cache.result[tuple] = proven
+ }
+ return wanted, nil
+}
+
+// AdmittedOrigin is the singleton convenience form; multi-origin callers use
+// AdmittedOrigins or one OriginProofCache.Proven call for their entire set.
+func AdmittedOrigin(ctx context.Context, s release.Store, obs Observed, incarnation, location, region, generation string) (bool, error) {
+ tuple := OriginTuple{incarnation, location, region, generation}
+ proven, err := AdmittedOrigins(ctx, s, obs, []OriginTuple{tuple})
+ return proven[tuple], err
+}
diff --git a/internal/hostclaim/plan15_batch_test.go b/internal/hostclaim/plan15_batch_test.go
new file mode 100644
index 000000000..58595f38f
--- /dev/null
+++ b/internal/hostclaim/plan15_batch_test.go
@@ -0,0 +1,148 @@
+package hostclaim
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "strings"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+type plan15BoundedStore struct {
+ *release.DirStore
+ oversized, conflict bool
+ read, closed, opens, gets int
+}
+
+func (s *plan15BoundedStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ 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.Contains(key, "control/host-origins/") {
+ return "", release.ErrCASConflict
+ }
+ return s.DirStore.Put(ctx, key, b, tag)
+}
+
+type plan15Reader struct {
+ io.Reader
+ store *plan15BoundedStore
+}
+
+func (r plan15Reader) Read(b []byte) (int, error) {
+ n, e := r.Reader.Read(b)
+ r.store.read += n
+ return n, e
+}
+func (r plan15Reader) Close() error { r.store.closed++; return nil }
+func (s *plan15BoundedStore) Open(ctx context.Context, key string) (io.ReadCloser, string, error) {
+ s.opens++
+ if s.oversized {
+ return plan15Reader{strings.NewReader(strings.Repeat("x", 1<<20)), s}, "oversized", nil
+ }
+ return s.DirStore.Open(ctx, key)
+}
+func TestPlan15BoundedReachableAndConflicting(t *testing.T) {
+ 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) {
+ plan15Enable(t)
+ s := &plan15BoundedStore{DirStore: release.NewDirStore(t.TempDir())}
+ o := servingClaim(t, t.Context(), s)
+ o = updateGenerationTest(t, t.Context(), s, o, plan15Admit(o, "first", "ie"))
+ h := o.Claim.Holder
+ first := OriginTuple{h.Incarnation, h.Location, "ie", "first"}
+ cache := NewOriginProofCache(s)
+ prove := func(obs Observed, set []OriginTuple, want bool) {
+ t.Helper()
+ result, err := cache.Proven(t.Context(), obs, set)
+ if err != nil || result[set[0]] != want {
+ t.Fatalf("%v %v", result, err)
+ }
+ }
+ s.opens, s.gets = 0, 0
+ prove(o, []OriginTuple{first, first}, true)
+ prove(o, []OriginTuple{first}, true)
+ if s.opens != int(o.Claim.Origins.Count) || s.gets != 2 {
+ t.Fatalf("cache opens=%d reads=%d", s.opens, s.gets)
+ }
+ different := first
+ different.Region = "uk"
+ prove(o, []OriginTuple{different}, false)
+ if s.opens != 2*int(o.Claim.Origins.Count) {
+ t.Fatal("tuple set reused")
+ }
+ other := &plan15BoundedStore{DirStore: release.NewDirStore(t.TempDir())}
+ key, _ := Key(o.Host)
+ body, _, err := s.Get(t.Context(), key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := other.Put(t.Context(), key, body, ""); err != nil {
+ t.Fatal(err)
+ }
+ 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)
+ edited := o
+ 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(), 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(), o, []OriginTuple{first}); got != nil || !errors.Is(err, ErrConflict) {
+ t.Fatal("stale observation reused")
+ }
+ s.opens = 0
+ prove(newer, []OriginTuple{first}, true)
+ if s.opens != int(newer.Claim.Origins.Count) {
+ t.Fatal("changed head did not traverse")
+ }
+ // No partial positives escape a failed batch, including a corrupt tail.
+ if err := s.Delete(t.Context(), originKey(o.Host, o.Claim.Origins.Digest)); err != nil {
+ t.Fatal(err)
+ }
+ second := first
+ second.Generation = "second"
+ 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
new file mode 100644
index 000000000..941d9cfe6
--- /dev/null
+++ b/internal/hostclaim/plan15_origins_test.go
@@ -0,0 +1,308 @@
+package hostclaim
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "os"
+ "reflect"
+ "strings"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func plan15Enable(t *testing.T) {
+ t.Helper()
+ t.Setenv("CARLOS_HOST_CLAIM", "1")
+ t.Setenv("CARLOS_HOST_CLAIM_ORIGINS", "1")
+}
+func plan15Admit(o Observed, generation, region string) func(*Claim) error {
+ h := o.Claim.Holder
+ return AdmitWriter(Admission{ObservedOpID: o.Claim.Op.ID, ObservedPhase: o.Claim.Op.Phase,
+ ObservedSession: h.Session, ObservedGeneration: h.Generation, Box: o.Claim.Op.Exec,
+ Mode: "cold", NewGeneration: generation, Region: region, PredecessorRegion: region, PredecessorCertified: true})
+}
+func plan15Proof(t *testing.T, s release.Store, o Observed, generation, region string, want bool) {
+ t.Helper()
+ h := o.Claim.Holder
+ got, err := AdmittedOrigin(t.Context(), s, o, h.Incarnation, h.Location, region, generation)
+ if err != nil || got != want {
+ t.Fatalf("proof %s/%s=%v,%v want %v", region, generation, got, err, want)
+ }
+}
+func TestPlan15PruningMigrationAndScope(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)
+ // A legacy claim can have more admissions than the covered retention window.
+ for i := 0; i < 12; i++ {
+ o = updateGenerationTest(t, t.Context(), s, o, plan15Admit(o, fmt.Sprintf("old%d", i), "ie"))
+ o = updateGenerationTest(t, t.Context(), s, o, func(c *Claim) error {
+ for i := range c.Holder.Admissions {
+ c.Holder.Admissions[i].Covered = true
+ }
+ return nil
+ })
+ }
+ if len(o.Claim.Holder.Admissions) != 8 {
+ t.Fatal("fixture did not prune")
+ }
+ plan15Enable(t)
+ // The chain starts empty: before the first recorded admission, even a
+ // retained (and possibly injected) admission proves nothing.
+ plan15Proof(t, s, o, "old0", "ie", false)
+ plan15Proof(t, s, o, "old4", "ie", false)
+ plan15Proof(t, s, o, "old11", "ie", false)
+ o = updateGenerationTest(t, t.Context(), s, o, plan15Admit(o, "first", "ie"))
+ for i := 0; i < 24; i++ {
+ o = updateGenerationTest(t, t.Context(), s, o, func(c *Claim) error {
+ for i := range c.Holder.Admissions {
+ c.Holder.Admissions[i].Covered = true
+ }
+ return nil
+ })
+ o = updateGenerationTest(t, t.Context(), s, o, plan15Admit(o, fmt.Sprintf("new%d", i), "ie"))
+ }
+ 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
+ for _, q := range []struct{ inc, loc string }{{strings.Repeat("b", 64), "instances/" + o.Host}, {h.Incarnation, "instances/" + o.Host}} {
+ got, err := AdmittedOrigin(t.Context(), s, o, q.inc, q.loc, "ie", "first")
+ if got || err != nil {
+ t.Fatalf("wrong scope %v %v", got, err)
+ }
+ }
+ // The old incarnation's history survives holder removal and the retirement cap.
+ old := *h
+ o = updateGenerationTest(t, t.Context(), s, o, func(c *Claim) error {
+ c.Holder = nil
+ c.Op = Op{}
+ c.Obligations = nil
+ c.RetiredIncarnations = nil
+ return nil
+ })
+ got, err := AdmittedOrigin(t.Context(), s, o, old.Incarnation, old.Location, "ie", "first")
+ if !got || err != nil {
+ t.Fatalf("retired origin %v %v", got, err)
+ }
+}
+
+type plan15Store struct {
+ release.Store
+ failNode, failClaim, conflict, uncertain bool
+ 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 {
+ return "", os.ErrPermission
+ }
+ if strings.HasPrefix(key, "control/hosts/") {
+ if s.failClaim {
+ return "", os.ErrPermission
+ }
+ if s.conflict {
+ return "", release.ErrCASConflict
+ }
+ if s.uncertain {
+ _, err := s.Store.Put(ctx, key, b, tag)
+ if err != nil {
+ return "", err
+ }
+ return "", os.ErrDeadlineExceeded
+ }
+ }
+ return s.Store.Put(ctx, key, b, tag)
+}
+func TestPlan15AtomicityAndRetry(t *testing.T) {
+ for _, fault := range []string{"node", "claim", "conflict", "uncertain"} {
+ t.Run(fault, func(t *testing.T) {
+ plan15Enable(t)
+ raw := release.NewDirStore(t.TempDir())
+ o := servingClaim(t, t.Context(), raw)
+ s := &plan15Store{Store: raw, failNode: fault == "node", failClaim: fault == "claim", conflict: fault == "conflict", uncertain: fault == "uncertain"}
+ if _, err := Update(t.Context(), s, o, plan15Admit(o, "candidate", "ie")); err == nil {
+ t.Fatal("fault ignored")
+ }
+ fresh, err := Read(t.Context(), raw, o.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ committed := fault == "uncertain"
+ if (fresh.Claim.Holder.Generation == "candidate") != committed || (fresh.Claim.Origins != nil) != committed {
+ t.Fatal("split commit")
+ }
+ plan15Proof(t, raw, fresh, "candidate", "ie", committed)
+ if !committed {
+ fresh = updateGenerationTest(t, t.Context(), raw, fresh, plan15Admit(fresh, "candidate", "ie"))
+ plan15Proof(t, raw, fresh, "candidate", "ie", true)
+ }
+ if _, err := Update(t.Context(), raw, o, plan15Admit(o, "stale", "ie")); !errors.Is(err, ErrConflict) {
+ t.Fatalf("stale: %v", err)
+ }
+ plan15Proof(t, raw, fresh, "stale", "ie", false)
+ })
+ }
+}
+func TestPlan15NoSelfAuthentication(t *testing.T) {
+ plan15Enable(t)
+ raw := release.NewDirStore(t.TempDir())
+ o := servingClaim(t, t.Context(), raw)
+ o = updateGenerationTest(t, t.Context(), raw, o, plan15Admit(o, "real", "ie"))
+ h := o.Claim.Holder
+ fake := originNode{V: 1, Host: o.Host, Count: o.Claim.Origins.Count + 1, Previous: o.Claim.Origins.Digest, Origin: originTuple{h.Incarnation, h.Location, "ie", "forged"}}
+ body, _ := json.Marshal(fake)
+ digest := originDigest(body)
+ if _, err := raw.Put(t.Context(), originKey(o.Host, digest), body, ""); err != nil {
+ t.Fatal(err)
+ }
+ plan15Proof(t, raw, o, "forged", "ie", false)
+ // Even a caller who knows the ETag cannot substitute a new head in the reader.
+ tampered := o
+ clone := *o.Claim
+ tampered.Claim = &clone
+ clone.Origins = &OriginHead{digest, fake.Count}
+ if ok, err := AdmittedOrigin(t.Context(), raw, tampered, h.Incarnation, h.Location, "ie", "forged"); ok || !errors.Is(err, ErrConflict) {
+ t.Fatalf("edited observation: %v %v", ok, err)
+ }
+ for _, change := range []func(*Claim){func(c *Claim) { c.Origins = &OriginHead{digest, fake.Count} }, func(c *Claim) { c.Origins = nil }} {
+ if _, err := Update(t.Context(), raw, o, func(c *Claim) error { change(c); return nil }); !errors.Is(err, ErrWrongState) {
+ t.Fatalf("head bypass: %v", err)
+ }
+ }
+ // 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)
+ }
+ // Missing and corrupt reachable nodes are errors, never an affirmative answer.
+ key := originKey(o.Host, o.Claim.Origins.Digest)
+ b, tag, err := raw.Get(t.Context(), key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err = raw.Put(t.Context(), key, append(b, ' '), tag); err != nil {
+ t.Fatal(err)
+ }
+ if ok, err := AdmittedOrigin(t.Context(), raw, o, h.Incarnation, h.Location, "ie", "real"); ok || err == nil {
+ t.Fatal("corruption accepted")
+ }
+ if err = raw.Delete(t.Context(), key); err != nil {
+ t.Fatal(err)
+ }
+ if ok, err := AdmittedOrigin(t.Context(), raw, o, h.Incarnation, h.Location, "ie", "real"); ok || err == nil {
+ t.Fatal("missing node accepted")
+ }
+}
+func TestPlan15FlagOffAndWarm(t *testing.T) {
+ for _, flags := range [][2]string{{"", ""}, {"", "1"}, {"1", ""}} {
+ t.Run(fmt.Sprint(flags), func(t *testing.T) {
+ t.Setenv("CARLOS_HOST_CLAIM", flags[0])
+ t.Setenv("CARLOS_HOST_CLAIM_ORIGINS", flags[1])
+ raw := release.NewDirStore(t.TempDir())
+ o := servingClaim(t, t.Context(), raw)
+ s := &plan15Store{Store: raw}
+ 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)
+ if strings.Contains(string(b), "admitted_origins") {
+ t.Fatal("flag-off wire changed")
+ }
+ })
+ }
+ plan15Enable(t)
+ raw := release.NewDirStore(t.TempDir())
+ o := servingClaim(t, t.Context(), raw)
+ o = updateGenerationTest(t, t.Context(), raw, o, plan15Admit(o, "same", "ie"))
+ before := *o.Claim.Origins
+ o = updateGenerationTest(t, t.Context(), raw, o, AdmitWriter(Admission{ObservedSession: o.Claim.Holder.Session, ObservedGeneration: "same", Box: o.Claim.Op.Exec, Mode: "warm", Region: "ie", PredecessorCertified: true}))
+ if !reflect.DeepEqual(before, *o.Claim.Origins) {
+ t.Fatal("warm duplicated origin")
+ }
+ plan15Proof(t, raw, o, "same", "ie", true)
+ t.Setenv("CARLOS_HOST_CLAIM_ORIGINS", "")
+ if _, err := Update(t.Context(), raw, o, plan15Admit(o, "gap", "ie")); !errors.Is(err, ErrWrongState) {
+ t.Fatalf("disabled writer made gap: %v", err)
+ }
+}
+
+func TestPlan15LimitsAndMalformedHistory(t *testing.T) {
+ plan15Enable(t)
+ s := release.NewDirStore(t.TempDir())
+ o := servingClaim(t, t.Context(), s)
+ for _, gen := range []string{"../escape", RootGeneration, strings.Repeat("x", 257)} {
+ if _, err := Update(t.Context(), s, o, plan15Admit(o, gen, "ie")); err == nil {
+ t.Fatalf("bad generation %q", gen)
+ }
+ }
+ o = updateGenerationTest(t, t.Context(), s, o, plan15Admit(o, "first", "ie"))
+ h := o.Claim.Holder
+ // A synthetic near-limit history tests refusal without writing a million nodes.
+ node := originNode{V: 1, Host: o.Host, Count: originCountLimit, Previous: strings.Repeat("a", 64), Origin: originTuple{h.Incarnation, h.Location, "ie", "first"}}
+ b, _ := json.Marshal(node)
+ head := OriginHead{originDigest(b), originCountLimit}
+ if _, err := s.Put(t.Context(), originKey(o.Host, head.Digest), b, ""); err != nil {
+ t.Fatal(err)
+ }
+ o.Claim.Origins = &head
+ key, _ := Key(o.Host)
+ b, _ = json.Marshal(o.Claim)
+ tag, err := s.Put(t.Context(), key, b, o.ETag)
+ if err != nil {
+ t.Fatal(err)
+ }
+ o.ETag = tag
+ if _, err := Update(t.Context(), s, o, plan15Admit(o, "over-limit", "ie")); !errors.Is(err, ErrWrongState) {
+ t.Fatalf("limit: %v", err)
+ }
+ ctx, cancel := context.WithCancel(t.Context())
+ cancel()
+ if ok, err := AdmittedOrigin(ctx, s, o, h.Incarnation, h.Location, "ie", "absent"); ok || err == nil {
+ t.Fatal("ignored cancellation")
+ }
+ node.Count--
+ b, _ = json.Marshal(node)
+ // Same digest-shaped head with a mismatched count cannot prove membership.
+ if _, err := s.Put(t.Context(), originKey(o.Host, originDigest(b)), b, ""); err != nil {
+ t.Fatal(err)
+ }
+ o.Claim.Origins = &OriginHead{originDigest(b), originCountLimit}
+ b, _ = json.Marshal(o.Claim)
+ o.ETag, err = s.Put(t.Context(), key, b, o.ETag)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if ok, err := AdmittedOrigin(t.Context(), s, o, h.Incarnation, h.Location, "ie", "first"); ok || err == nil {
+ t.Fatal("count mismatch accepted")
+ }
+ for _, head := range []*OriginHead{{"bad", 1}, {strings.Repeat("a", 64), 0}, {strings.Repeat("a", 64), originCountLimit + 1}} {
+ o.Claim.Origins = head
+ if err := o.Claim.Validate(); err == nil {
+ t.Fatal("bad head parsed")
+ }
+ }
+}
diff --git a/internal/hostclaim/store.go b/internal/hostclaim/store.go
index b421bab85..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
@@ -98,6 +107,9 @@ func Update(ctx context.Context, s release.Store, obs Observed, tr func(*Claim)
if err := next.Validate(); err != nil {
return Observed{}, err
}
+ if err := commitOrigins(ctx, s, obs, next); err != nil {
+ return Observed{}, err
+ }
body, err := json.Marshal(next)
if err != nil {
return Observed{}, err
diff --git a/internal/hostops/lifecycle_grant.go b/internal/hostops/lifecycle_grant.go
index fb94ded30..ed55d85e5 100644
--- a/internal/hostops/lifecycle_grant.go
+++ b/internal/hostops/lifecycle_grant.go
@@ -9,6 +9,7 @@ import (
"io"
"math"
"os"
+ "reflect"
"strings"
"time"
@@ -169,7 +170,9 @@ func authorizeClaimWrite(ctx context.Context, raw release.Store, host string, t
if err != nil || next == nil {
return ErrIdentity
}
- if !claimTransitionShape(host, old, next) {
+ if !claimTransitionShape(host, old, next) ||
+ (old == nil && next.Origins != nil) ||
+ (old != nil && !reflect.DeepEqual(old.Origins, next.Origins)) {
return ErrIdentity
}
for _, g := range grants {
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/plan15_origins_test.go b/internal/hostops/plan15_origins_test.go
new file mode 100644
index 000000000..702579cf3
--- /dev/null
+++ b/internal/hostops/plan15_origins_test.go
@@ -0,0 +1,59 @@
+package hostops
+
+import (
+ "crypto/sha256"
+ "encoding/hex"
+ "errors"
+ "strings"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+)
+
+func TestPlan15OriginCandidateBoundary(t *testing.T) {
+ w, raw, _ := testWriter(t)
+ body := []byte(`{"v":1,"host":"x.example","count":1,"origin":{"incarnation":"` + strings.Repeat("a", 64) + `","location":"instances/x.example","region":"ie","generation":"g1"}}`)
+ sum := sha256.Sum256(append([]byte("carlos-admitted-origin-v1\x00"), body...))
+ key := "control/host-origins/x.example/" + hex.EncodeToString(sum[:]) + ".json"
+ for _, tc := range []struct {
+ name string
+ target MutationTarget
+ }{
+ {"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 !errors.Is(err, ErrUnclassified) {
+ t.Fatalf("classification %+v %v", c, err)
+ }
+ })
+ }
+}
+
+func TestPlan15LifecycleGrantCannotRewriteOrigins(t *testing.T) {
+ w, o, clock := wakeResetFixture(t)
+ ctx := registerFor(t, w, o, "create2", "create", o.Claim.Holder.Incarnation, clock)
+ step := claimStep(t, o, func(c *hostclaim.Claim) error {
+ c.Origins = &hostclaim.OriginHead{Digest: strings.Repeat("a", 64), Count: 1}
+ return nil
+ })
+ if err := authorizeClaimWrite(ctx, w.raw, o.Host, step); err == nil {
+ t.Fatal("general app grant manufactured origin authority")
+ }
+ if err := authorizeClaimWrite(ctx, w.raw, o.Host, claimStep(t, o, func(*hostclaim.Claim) error { return nil })); err != nil {
+ t.Fatalf("unchanged head refused: %v", err)
+ }
+ o.Claim.Origins = &hostclaim.OriginHead{Digest: strings.Repeat("b", 64), Count: 2}
+ for _, head := range []*hostclaim.OriginHead{nil, {Digest: strings.Repeat("a", 64), Count: 1}} {
+ step = claimStep(t, o, func(c *hostclaim.Claim) error { c.Origins = head; return nil })
+ if err := authorizeClaimWrite(ctx, w.raw, o.Host, step); err == nil {
+ t.Fatal("grant erased or rolled back origin history")
+ }
+ }
+}
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 bfb7fc6f8..e97ca2eca 100644
--- a/internal/provision/plan9_b8_census_test.go
+++ b/internal/provision/plan9_b8_census_test.go
@@ -748,13 +748,15 @@ var plan9B8CensusInventory = []censusEntry{
{"internal/hostclaim/adopt.go", "Adopt", "key-site", "lit:instances", "24c06b99cce6c638acb19826985cfadd3c736fd0ace7289bcf7ac3dfe784f05a"},
{"internal/hostclaim/claim.go", "ParkScope.Validate", "key-site", "lit:instances", "80b798b0cb679d8c533ff562c0c4804c06f53e7dba126250a3c8182f7ff0c08c"},
{"internal/hostclaim/claim.go", "holderKeyMatches", "key-site", "lit:instances,lit:sidecars", "d15a6841ea6f8cbfa9c8407fd3af020ddeacdfa0743b9e17de412f113faf7979"},
- {"internal/hostclaim/claim.go", "Claim.Validate", "key-site", "lit:instances", "3278f2b4b24b6627b9dcd0863c3bef58c6694a65c8a19d05ce280ae34801bdfc"},
+ {"internal/hostclaim/claim.go", "Claim.Validate", "key-site", "lit:instances", "2debcd30dda4747b8b3b00d576998f1fdc1d364ad9bacba1e1279df0b7dfb694"},
{"internal/hostclaim/create.go", "BeginCreate", "key-site", "lit:instances", "45ee1e2b1fc9fa8ffdef4f27cf983dccfa3acf7218dccd06f0ee6d12572f4a44"},
{"internal/hostclaim/lossrecord.go", "LossRecordKey", "key-site", "key:LossRecordKey", "1b109b49a8fcdb09f01dbf908a173394072f01afb0c9e324023fb9b3f71f98b3"},
{"internal/hostclaim/lossrecord.go", "ReadLossRecord", "key-site", "key:LossRecordKey", "1237dd3fa3e79a81e6c35f371bf545c9a04841a6297c078d96852daf25b7b483"},
{"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/store.go", "Update", "nonrecord", "sel:Put", "68f97a49f9a936228cb26e49d7bed3a4ab17c3a4c4255e32c2d2af1dfccd73a6"},
+ {"internal/hostclaim/origins.go", "originTuple.valid", "key-site", "lit:instances", "7f6f20f51e51694d665588f0b986def85bda4789f24623ee1993289265b01355"},
+ {"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"},
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)
+ }
+}