amadan

carlos / platform Public

hostfence, host: the regional termination-evidence producer (claim Plan 8b Task 4)

Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
codex pushed by claude 264381ed26c119ddb1ccd13a7609365a0037bc49 parent 1f7151c
4 files changed, +454 −0
  • internal/host/agent.go +2 −0
  • internal/host/hostfencepass.go +75 −0
  • internal/hostfence/termination.go +185 −0
  • internal/hostfence/termination_test.go +192 −0
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 328c83bef..2045cc082 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -201,6 +201,7 @@ func dataDir() string {
// edge.RunWithActivator, which owns the serving loop and blocks until
// SIGTERM/SIGINT.
func RunAgent(args []string) {
+ go runHostFencePass(context.Background())
dd := dataDir()
regPath := peekRegistryPath(args, dd)
// A second handle on the same registry file the edge opens — the registry
@@ -2936,6 +2937,7 @@ func statusLoop(reg *registry.Registry) {
}
// The hosts this box is refusing because it cannot provide their
// config (amadan #213): private Metrics only, nothing public.
+ claimReport(&p)
p.ConfigRefusals, p.ConfigRefusalsTotal = configRefusalProbe()
p.ConfigGuard = configGuardProbe()
diff --git a/internal/host/hostfencepass.go b/internal/host/hostfencepass.go
new file mode 100644
index 000000000..755a94d64
--- /dev/null
+++ b/internal/host/hostfencepass.go
@@ -0,0 +1,75 @@
+package host
+
+import (
+ "context"
+ "github.com/carlosframework/platform/internal/awsinv"
+ "github.com/carlosframework/platform/internal/hostactors"
+ "github.com/carlosframework/platform/internal/hostfence"
+ "github.com/carlosframework/platform/internal/hostfloor"
+ "github.com/carlosframework/platform/internal/release"
+ "log"
+ "os"
+ "strings"
+ "time"
+)
+
+func runHostFencePass(ctx context.Context) {
+ if strings.TrimSpace(os.Getenv("CARLOS_HOST_CLAIM")) == "" || os.Getenv("CARLOS_FLEET") != "" {
+ return
+ }
+ store, pub, err := openSystemStore(ctx)
+ if err != nil {
+ log.Printf("host fences: %v", err)
+ return
+ }
+ enrollment, err := hostfloor.ReadEnrollment("/etc/carlos/host-actor.json")
+ if err != nil {
+ log.Printf("host fences: %v", err)
+ return
+ }
+ d := hostactors.Directory{Store: store, PublicKey: pub, Bucket: os.Getenv("CARLOS_DEPLOYMENT_BUCKET")}
+ worker := hostfence.TerminationWorker{Enabled: true, Directory: d, Self: hostfence.Worker{Box: enrollment.Box, Enrollment: enrollment.Enrollment}, Reader: awsinv.NewEC2Reader, Now: time.Now}
+ ticker := time.NewTicker(10 * time.Minute)
+ defer ticker.Stop()
+ for {
+ pass, cancel := context.WithTimeout(ctx, 2*time.Minute)
+ floor, err := hostfloor.Read(pass, store, pub, d.Bucket)
+ if err == nil {
+ local, loadErr := floorDisk(release.OptDir())
+ if loadErr != nil {
+ err = loadErr
+ } else {
+ _, armed, e := local.Load()
+ if e != nil {
+ err = e
+ } else if armed.Digest != floor.Digest {
+ err = hostfloor.ErrFloor
+ }
+ }
+ }
+ if err == nil {
+ roster, e := d.Read(pass)
+ err = e
+ if e == nil {
+ self := roster.Roster.Boxes[enrollment.Box]
+ for ref, box := range roster.Roster.Boxes {
+ if ref == enrollment.Box || box.Region != self.Region || box.State == "retired" {
+ continue
+ }
+ if _, fenceErr := worker.Describe(pass, ref); fenceErr != nil {
+ log.Printf("host fence %s: awaiting verified termination: %v", ref, fenceErr)
+ }
+ }
+ }
+ }
+ if err != nil {
+ log.Printf("host fence pass: %v", err)
+ }
+ cancel()
+ select {
+ case <-ctx.Done():
+ return
+ case <-ticker.C:
+ }
+ }
+}
diff --git a/internal/hostfence/termination.go b/internal/hostfence/termination.go
new file mode 100644
index 000000000..38692a4d6
--- /dev/null
+++ b/internal/hostfence/termination.go
@@ -0,0 +1,185 @@
+package hostfence
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "github.com/carlosframework/platform/internal/awsinv"
+ "github.com/carlosframework/platform/internal/hostactors"
+ "github.com/carlosframework/platform/internal/hostfloor"
+ "time"
+)
+
+type Worker struct {
+ Box string `json:"box"`
+ Enrollment string `json:"enrollment"`
+}
+type Termination struct {
+ V int `json:"v"`
+ Box string `json:"box"`
+ Enrollment string `json:"enrollment"`
+ Cloud hostactors.Cloud `json:"cloud"`
+ State string `json:"state"`
+ ObservedAt time.Time `json:"observed_at"`
+ Via string `json:"via"`
+ RequestID string `json:"request_id,omitempty"`
+ EventID string `json:"event_id,omitempty"`
+ WrittenBy Worker `json:"written_by"`
+}
+type EvidenceRef struct {
+ Enrollment string `json:"enrollment"`
+ Key string `json:"key"`
+ SHA256 string `json:"sha256"`
+}
+type TerminationWorker struct {
+ Enabled bool
+ Directory hostactors.Directory
+ Self Worker
+ Reader func(context.Context, string) (awsinv.ExactEC2API, error)
+ Now func() time.Time
+}
+
+func FenceKey(enrollment string) string { return "control/box-fences/" + enrollment + ".json" }
+func (w TerminationWorker) binding(ctx context.Context, box string) (hostactors.Box, error) {
+ if !w.Enabled || w.Now == nil {
+ return hostactors.Box{}, ErrObservation
+ }
+ o, err := w.Directory.Read(ctx)
+ if err != nil {
+ return hostactors.Box{}, err
+ }
+ self, ok := o.Roster.Boxes[w.Self.Box]
+ if !ok || self.State != "active" || self.Enrollment != w.Self.Enrollment {
+ return hostactors.Box{}, ErrObservation
+ }
+ eligible, err := w.Directory.Lookup(ctx, w.Self.Box)
+ if err != nil || eligible.State != "active" || eligible.Enrollment != w.Self.Enrollment {
+ return hostactors.Box{}, ErrObservation
+ }
+ target, ok := o.Roster.Boxes[box]
+ if !ok || target.Region != self.Region || target.Cloud.Region != self.Cloud.Region || target.Cloud.Account != self.Cloud.Account {
+ return hostactors.Box{}, ErrObservation
+ }
+ return target, nil
+}
+func (w TerminationWorker) Describe(ctx context.Context, box string) (EvidenceRef, error) {
+ b, err := w.binding(ctx, box)
+ if err != nil {
+ return EvidenceRef{}, err
+ }
+ if w.Reader == nil {
+ return EvidenceRef{}, ErrObservation
+ }
+ api, err := w.Reader(ctx, b.Cloud.Region)
+ if err != nil {
+ return EvidenceRef{}, err
+ }
+ observation, err := awsinv.VerifyExactStoppedInstance(ctx, api, b.Cloud.Account, b.Cloud.Region, b.Cloud.InstanceID)
+ if err != nil {
+ return EvidenceRef{}, err
+ }
+ if observation.State != "terminated" {
+ return EvidenceRef{}, ErrObservation
+ }
+ return w.persist(ctx, Termination{V: 1, Box: box, Enrollment: b.Enrollment, Cloud: b.Cloud, State: "terminated", ObservedAt: w.Now().UTC(), Via: "describe", RequestID: observation.RequestID, WrittenBy: w.Self})
+}
+
+// Event is called only by the regional queue consumer, after the existing
+// queue URL and rule-ARN policy validation. It is never an HTTP request body.
+func (w TerminationWorker) Event(ctx context.Context, box string, raw []byte) (EvidenceRef, error) {
+ b, err := w.binding(ctx, box)
+ if err != nil {
+ return EvidenceRef{}, err
+ }
+ var event struct {
+ ID string `json:"id"`
+ Source string `json:"source"`
+ Type string `json:"detail-type"`
+ Account string `json:"account"`
+ Region string `json:"region"`
+ Time time.Time `json:"time"`
+ Detail struct {
+ ID string `json:"instance-id"`
+ State string `json:"state"`
+ } `json:"detail"`
+ }
+ if json.Unmarshal(raw, &event) != nil || event.ID == "" || event.Source != "aws.ec2" || event.Type != "EC2 Instance State-change Notification" || event.Account != b.Cloud.Account || event.Region != b.Cloud.Region || event.Detail.ID != b.Cloud.InstanceID || event.Detail.State != "terminated" || event.Time.IsZero() || event.Time.After(w.Now().Add(5*time.Second)) {
+ return EvidenceRef{}, ErrObservation
+ }
+ // Receipt time is conservative for the late completion horizon. Retries use
+ // the first accepted record, so they cannot move that horizon forward.
+ return w.persist(ctx, Termination{V: 1, Box: box, Enrollment: b.Enrollment, Cloud: b.Cloud, State: "terminated", ObservedAt: w.Now().UTC(), Via: "event", EventID: event.ID, WrittenBy: w.Self})
+}
+func (w TerminationWorker) persist(ctx context.Context, f Termination) (EvidenceRef, error) {
+ raw, err := json.Marshal(f)
+ if err != nil {
+ return EvidenceRef{}, err
+ }
+ key := FenceKey(f.Enrollment)
+ _, err = w.Directory.Store.Put(ctx, key, raw, "")
+ if err != nil {
+ saved, _, readErr := w.Directory.Store.Get(ctx, key)
+ if readErr != nil {
+ return EvidenceRef{}, errors.Join(err, readErr)
+ }
+ var old Termination
+ if json.Unmarshal(saved, &old) != nil || old.V != 1 || old.Box != f.Box || old.Enrollment != f.Enrollment || old.Cloud != f.Cloud || old.State != "terminated" || old.ObservedAt.IsZero() || (old.Via != "describe" && old.Via != "event") || (old.RequestID == "" && old.EventID == "") {
+ return EvidenceRef{}, ErrObservation
+ }
+ raw = saved
+ }
+ return EvidenceRef{Enrollment: f.Enrollment, Key: key, SHA256: hostfloor.Digest(raw)}, nil
+}
+
+type BlockDevice struct {
+ Device string `json:"device"`
+ VolumeID string `json:"volume_id"`
+ DeleteOnTermination bool `json:"delete_on_termination"`
+ Status string `json:"status"`
+}
+type VolumeRecord struct {
+ V int `json:"v"`
+ Box string `json:"box"`
+ Enrollment string `json:"enrollment"`
+ Cloud hostactors.Cloud `json:"cloud"`
+ InstanceState string `json:"instance_state"`
+ BlockDevices []BlockDevice `json:"block_devices"`
+ RemovedVolumes []string `json:"removed_volumes"`
+ ObservedAt time.Time `json:"observed_at"`
+ RequestID string `json:"request_id"`
+ WrittenBy Worker `json:"written_by"`
+}
+
+// DestroyedLocal never participates in execution-fence acceptance. The caller
+// supplies the complete paginated volume history, with enrollment-time first
+// evidence and a bounded final running observation. Unknown history is false.
+func DestroyedLocal(f Termination, records []VolumeRecord, complete bool, enrolledAt time.Time, maxAge time.Duration) bool {
+ if !complete || len(records) == 0 || enrolledAt.IsZero() || maxAge <= 0 {
+ return false
+ }
+ if records[0].ObservedAt.After(enrolledAt) {
+ return false
+ }
+ previous := map[string]bool{}
+ last := time.Time{}
+ for _, r := range records {
+ if r.V != 1 || r.Box != f.Box || r.Enrollment != f.Enrollment || r.Cloud != f.Cloud || r.RequestID == "" || r.ObservedAt.IsZero() || !r.ObservedAt.After(last) || r.ObservedAt.After(f.ObservedAt) || (r.InstanceState != "running" && r.InstanceState != "shutting-down") || len(r.RemovedVolumes) > 0 || len(r.BlockDevices) == 0 {
+ return false
+ }
+ current := map[string]bool{}
+ for _, v := range r.BlockDevices {
+ if v.VolumeID == "" || v.Device == "" || !v.DeleteOnTermination || v.Status != "attached" || current[v.VolumeID] {
+ return false
+ }
+ current[v.VolumeID] = true
+ }
+ for id := range previous {
+ if !current[id] {
+ return false
+ }
+ }
+ previous = current
+ last = r.ObservedAt
+ }
+ return f.ObservedAt.Sub(last) <= maxAge
+}
diff --git a/internal/hostfence/termination_test.go b/internal/hostfence/termination_test.go
new file mode 100644
index 000000000..e61795337
--- /dev/null
+++ b/internal/hostfence/termination_test.go
@@ -0,0 +1,192 @@
+package hostfence
+
+import (
+ "context"
+ "crypto/ed25519"
+ "crypto/rand"
+ "encoding/json"
+ "github.com/aws/aws-sdk-go-v2/aws"
+ awsmiddleware "github.com/aws/aws-sdk-go-v2/aws/middleware"
+ "github.com/aws/aws-sdk-go-v2/service/ec2"
+ "github.com/aws/aws-sdk-go-v2/service/ec2/types"
+ "github.com/carlosframework/platform/internal/awsinv"
+ "github.com/carlosframework/platform/internal/hostactors"
+ "github.com/carlosframework/platform/internal/hostfloor"
+ "github.com/carlosframework/platform/internal/release"
+ "strings"
+ "testing"
+ "time"
+)
+
+type terminatedEC2 struct {
+ state types.InstanceStateName
+ missing bool
+ owner string
+ after func()
+}
+
+func (f terminatedEC2) DescribeInstances(_ context.Context, in *ec2.DescribeInstancesInput, _ ...func(*ec2.Options)) (*ec2.DescribeInstancesOutput, error) {
+ out := &ec2.DescribeInstancesOutput{}
+ if !f.missing {
+ out.Reservations = []types.Reservation{{OwnerId: aws.String(f.owner), Instances: []types.Instance{{InstanceId: aws.String(in.InstanceIds[0]), State: &types.InstanceState{Name: f.state}}}}}
+ }
+ if f.after != nil {
+ f.after()
+ }
+ awsmiddleware.SetRequestIDMetadata(&out.ResultMetadata, "request")
+ return out, nil
+}
+func terminationFixture(t *testing.T) (TerminationWorker, release.Store) {
+ t.Helper()
+ ctx := context.Background()
+ pub, key, _ := ed25519.GenerateKey(rand.Reader)
+ s := release.NewDirStore(t.TempDir())
+ now := time.Now().UTC()
+ r := hostactors.Roster{V: 1, Revision: 1, DeploymentBucket: "bucket", Boxes: map[string]hostactors.Box{}, Consoles: map[string]hostactors.Console{}, Holds: map[string]hostactors.Reservation{}, Activations: map[string]hostactors.Reservation{}}
+ for i, ref := range []string{"worker", "dead"} {
+ en := strings.Repeat(string(rune('a'+i)), 32)
+ r.Boxes[ref] = hostactors.Box{Enrollment: en, Namespace: "platform", BoxID: ref, Label: ref, Region: "ie", Pool: "default", Cloud: hostactors.Cloud{Provider: "aws", Account: "123", Region: "eu-west-1", InstanceID: "i-" + ref}, EvidenceSHA256: strings.Repeat("e", 64), State: "active"}
+ }
+ raw, _ := release.SignRootEnvelope(release.RootSigner(key), r)
+ if _, err := s.Put(ctx, hostactors.Key, raw, ""); err != nil {
+ t.Fatal(err)
+ }
+ return TerminationWorker{Enabled: true, Directory: hostactors.Directory{Store: s, PublicKey: pub, Bucket: "bucket"}, Self: Worker{Box: "worker", Enrollment: r.Boxes["worker"].Enrollment}, Now: func() time.Time { return now }}, s
+}
+func TestTerminationAloneFencesAndAbsenceNeverDoes(t *testing.T) {
+ for _, test := range []struct {
+ name string
+ state types.InstanceStateName
+ missing bool
+ owner string
+ ok bool
+ }{
+ {"terminated-no-volumes", types.InstanceStateNameTerminated, false, "123", true},
+ {"stopped", types.InstanceStateNameStopped, false, "123", false},
+ {"shutting-down", types.InstanceStateNameShuttingDown, false, "123", false},
+ {"absent", types.InstanceStateNameTerminated, true, "123", false},
+ {"wrong-account", types.InstanceStateNameTerminated, false, "456", false},
+ } {
+ t.Run(test.name, func(t *testing.T) {
+ w, s := terminationFixture(t)
+ w.Reader = func(context.Context, string) (awsinv.ExactEC2API, error) {
+ return terminatedEC2{state: test.state, missing: test.missing, owner: test.owner}, nil
+ }
+ ref, err := w.Describe(context.Background(), "dead")
+ if (err == nil) != test.ok {
+ t.Fatal(ref, err)
+ }
+ if test.ok {
+ raw, _, err := s.Get(context.Background(), ref.Key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var f Termination
+ _ = json.Unmarshal(raw, &f)
+ old := f.ObservedAt
+ w.Now = func() time.Time { return old.Add(time.Hour) }
+ again, err := w.Describe(context.Background(), "dead")
+ if err != nil || again != ref {
+ t.Fatal("retry moved evidence", again, err)
+ }
+ } else {
+ keys, _ := s.List(context.Background(), "control/box-fences/")
+ if len(keys) != 0 {
+ t.Fatal("refusal wrote a fence")
+ }
+ }
+ })
+ }
+}
+func TestDestroyedLocalIsIndependentAndNeedsHistory(t *testing.T) {
+ now := time.Now().UTC()
+ f := Termination{Box: "dead", Enrollment: "enrollment", ObservedAt: now}
+ if DestroyedLocal(f, nil, false, now.Add(-time.Hour), time.Hour) {
+ t.Fatal("empty terminated mappings proved destroyed storage")
+ }
+ r := VolumeRecord{V: 1, Box: f.Box, Enrollment: f.Enrollment, InstanceState: "running", ObservedAt: now.Add(-time.Minute), RequestID: "request", BlockDevices: []BlockDevice{{Device: "root", VolumeID: "volume", DeleteOnTermination: true, Status: "attached"}}}
+ if !DestroyedLocal(f, []VolumeRecord{r}, true, r.ObservedAt, time.Hour) {
+ t.Fatal("complete true history refused")
+ }
+ r.BlockDevices[0].DeleteOnTermination = false
+ if DestroyedLocal(f, []VolumeRecord{r}, true, r.ObservedAt, time.Hour) {
+ t.Fatal("surviving volume discharged")
+ }
+}
+
+func TestFencedWorkerCannotProduceNewEvidence(t *testing.T) {
+ w, s := terminationFixture(t)
+ o, err := w.Directory.Read(context.Background())
+ if err != nil {
+ t.Fatal(err)
+ }
+ self := o.Roster.Boxes["worker"]
+ dead := o.Roster.Boxes["dead"]
+ fence := Termination{V: 1, Box: "worker", Enrollment: self.Enrollment, Cloud: self.Cloud, State: "terminated", ObservedAt: w.Now(), Via: "describe", RequestID: "request", WrittenBy: Worker{Box: "dead", Enrollment: dead.Enrollment}}
+ raw, _ := json.Marshal(fence)
+ if _, err := s.Put(context.Background(), FenceKey(self.Enrollment), raw, ""); err != nil {
+ t.Fatal(err)
+ }
+ w.Reader = func(context.Context, string) (awsinv.ExactEC2API, error) {
+ return terminatedEC2{state: types.InstanceStateNameTerminated, owner: "123"}, nil
+ }
+ if _, err := w.Describe(context.Background(), "dead"); err == nil {
+ t.Fatal("fenced worker produced new execution authority")
+ }
+ keys, err := s.List(context.Background(), "control/box-fences/")
+ if err != nil || len(keys) != 1 {
+ t.Fatal("refused worker wrote another fence", keys, err)
+ }
+}
+
+func TestTerminationReferenceIncludesExactEnrollmentAndBytes(t *testing.T) {
+ w, s := terminationFixture(t)
+ w.Reader = func(context.Context, string) (awsinv.ExactEC2API, error) {
+ return terminatedEC2{state: types.InstanceStateNameTerminated, owner: "123"}, nil
+ }
+ ref, err := w.Describe(context.Background(), "dead")
+ if err != nil {
+ t.Fatal(err)
+ }
+ raw, _, err := s.Get(context.Background(), ref.Key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var fence Termination
+ if json.Unmarshal(raw, &fence) != nil || ref.Enrollment != fence.Enrollment || ref.Key != FenceKey(fence.Enrollment) || ref.SHA256 != hostfloor.Digest(raw) {
+ t.Fatal("reference is not bound to the verified enrollment and bytes", ref)
+ }
+ wire, err := json.Marshal(ref)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var fields map[string]string
+ if json.Unmarshal(wire, &fields) != nil || len(fields) != 3 || fields["enrollment"] != fence.Enrollment || fields["key"] != ref.Key || fields["sha256"] != ref.SHA256 {
+ t.Fatal("journal reference lacks the adopted D2 wire contract", string(wire))
+ }
+}
+
+func TestTerminationFenceTimeIsAfterTheVerifiedRead(t *testing.T) {
+ w, s := terminationFixture(t)
+ start := w.Now()
+ now := start
+ w.Now = func() time.Time { return now }
+ w.Reader = func(context.Context, string) (awsinv.ExactEC2API, error) {
+ return terminatedEC2{state: types.InstanceStateNameTerminated, owner: "123", after: func() { now = start.Add(time.Minute) }}, nil
+ }
+ ref, err := w.Describe(context.Background(), "dead")
+ if err != nil {
+ t.Fatal(err)
+ }
+ raw, _, err := s.Get(context.Background(), ref.Key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var fence Termination
+ if err = json.Unmarshal(raw, &fence); err != nil {
+ t.Fatal(err)
+ }
+ if !fence.ObservedAt.Equal(now) {
+ t.Fatal("fence time predates completion of the terminal-state read", fence.ObservedAt, now)
+ }
+}