amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package host
2
3import (
4 "context"
5 "crypto/ed25519"
6 "encoding/json"
7 "errors"
8 "os"
9 "path/filepath"
10 "reflect"
11 "strings"
12 "testing"
13 "time"
14
15 "github.com/carlosframework/platform/internal/hostclaim"
16 "github.com/carlosframework/platform/internal/hostguard"
17 "github.com/carlosframework/platform/internal/release"
18)
19
20type stopPeerFunc func(context.Context) (StopPeer, error)
21
22func (f stopPeerFunc) AuthenticatedPeer(ctx context.Context) (StopPeer, error) { return f(ctx) }
23
24type stopInventoryFunc func(context.Context, string) (StopInventory, error)
25
26func (f stopInventoryFunc) ReadStopInventory(ctx context.Context, host string) (StopInventory, error) {
27 return f(ctx, host)
28}
29
30type stopDrainFunc func(context.Context, StopRequest, string) (DrainAcknowledgement, error)
31
32func (f stopDrainFunc) DrainClaim(ctx context.Context, req StopRequest, fence string) (DrainAcknowledgement, error) {
33 return f(ctx, req, fence)
34}
35
36type stopKernelFunc func(context.Context, hostguard.FenceTarget, StopInventory, DrainAcknowledgement) (StopKernelEvidence, error)
37
38func (f stopKernelFunc) VerifyStopped(ctx context.Context, target hostguard.FenceTarget, inv StopInventory, ack DrainAcknowledgement) (StopKernelEvidence, error) {
39 return f(ctx, target, inv, ack)
40}
41
42type stopClaimReader struct{ obs hostclaim.Observed }
43
44func (r stopClaimReader) Read(context.Context, string) (hostclaim.Observed, error) { return r.obs, nil }
45
46type failPutStore struct {
47 release.Store
48 err error
49}
50
51func (s *failPutStore) Put(ctx context.Context, key string, body []byte, match string) (string, error) {
52 if s.err != nil {
53 err := s.err
54 s.err = nil
55 return "", err
56 }
57 return s.Store.Put(ctx, key, body, match)
58}
59
60func claimStopFixture(t *testing.T) (ClaimStopCoordinator, StopRequest, *failPutStore) {
61 t.Helper()
62 host := "shop.example"
63 data, locks := t.TempDir(), t.TempDir()
64 if err := os.Mkdir(filepath.Join(data, "hostguard"), 0770); err != nil {
65 t.Fatal(err)
66 }
67 claim := &hostclaim.Claim{V: 2, Host: host, Gen: 19,
68 Holder: &hostclaim.Holder{Incarnation: "inc-1", Stamp: 11, Session: 13},
69 Op: hostclaim.Op{Kind: hostclaim.KindMoving, ID: "operation-1", Phase: "stopping", Exec: "box-1"}}
70 obs := hostclaim.Observed{Host: host, Claim: claim, ETag: "etag-1"}
71 guard := hostguard.Guard{DataDir: data, LockDir: locks, Claims: stopClaimReader{obs}}
72 if err := guard.Activate(t.Context(), host, "box-1", nil); err != nil {
73 t.Fatal(err)
74 }
75 req := StopRequest{Host: host, Incarnation: "inc-1", InstallationStamp: 11, WriterSession: 13,
76 OperationID: "operation-1", Phase: "stopping", ObligationID: "obligation-1", Box: "box-1",
77 Enrollment: "enrollment-1", BootID: "boot-1", ControllerID: "controller-1",
78 ExpectedClaimETag: "etag-1", ExpectedDirectoryRevision: "directory-1", RosterRevision: 7}
79 base := release.NewDirStore(t.TempDir())
80 store := &failPutStore{Store: base}
81 ids := []string{"fence-1", "receipt-1", "fence-retry", "receipt-retry"}
82 now := time.Date(2026, 10, 6, 17, 0, 0, 123, time.UTC)
83 c := ClaimStopCoordinator{VerifiedRosterRevision: fixedRosterRevision(7), EvidenceSigner: ed25519.NewKeyFromSeed(make([]byte, ed25519.SeedSize)), RequestLimit: 120 * time.Second, Guard: guard, Claims: stopClaimReader{obs}, CarlosUID: 1001, Evidence: store,
84 Peers: stopPeerFunc(func(context.Context) (StopPeer, error) {
85 return StopPeer{UID: 1001, PID: 42, ProcessStart: "start-42", BootID: "boot-1", ControllerID: "controller-1"}, nil
86 }),
87 Inventory: stopInventoryFunc(func(_ context.Context, got string) (StopInventory, error) {
88 return StopInventory{Host: got, Incarnation: "inc-1", InstallationStamp: 11, Box: "box-1", Enrollment: "enrollment-1", DirectoryRevision: "directory-1", Entries: []StopInventoryEntry{{Kind: "path", ArtifactID: "artifact-1", ManagedRoot: "data", RelativePath: "derived/by/root"}}}, nil
89 }),
90 Drainer: stopDrainFunc(func(_ context.Context, _ StopRequest, fence string) (DrainAcknowledgement, error) {
91 return DrainAcknowledgement{FenceID: fence, ControllerID: "controller-1", ProcessStart: "start-42", Digest: strings.Repeat("a", 64)}, nil
92 }),
93 Kernel: stopKernelFunc(func(_ context.Context, _ hostguard.FenceTarget, _ StopInventory, _ DrainAcknowledgement) (StopKernelEvidence, error) {
94 return StopKernelEvidence{InventorySHA256: strings.Repeat("b", 64), LocalJournalSHA256: strings.Repeat("c", 64), CgroupIdentities: []string{"cg-1"}, UnitInvocationIDs: []string{"unit-1"}}, nil
95 }),
96 NewID: func() string { id := ids[0]; ids = ids[1:]; return id }, Now: func() time.Time { return now }}
97 return c, req, store
98}
99
100func TestClaimStopCollectsAndPublishesExactReceipt(t *testing.T) {
101 c, req, _ := claimStopFixture(t)
102 result, err := c.FenceAndStop(t.Context(), req)
103 if err != nil {
104 t.Fatal(err)
105 }
106 wantKey := "control/hosts/shop.example/evidence/operation-1/box-1/receipt-1.json"
107 if result.Key != wantKey || result.Digest == "" {
108 t.Fatalf("result = %+v", result)
109 }
110 stored, _, err := c.Evidence.Get(t.Context(), result.Key)
111 if err != nil || !reflect.DeepEqual(stored, result.Bytes) {
112 t.Fatalf("stored exact bytes = %v, %v", reflect.DeepEqual(stored, result.Bytes), err)
113 }
114 var receipt StopReceipt
115 if err := json.Unmarshal(stored, &receipt); err != nil {
116 t.Fatal(err)
117 }
118 if receipt.FenceID != "fence-1" || receipt.ID != "receipt-1" || receipt.EvidenceAt != result.EvidenceAt ||
119 receipt.ControllerID != req.ControllerID || receipt.Phase != req.Phase || receipt.ProducerDrainDigest != strings.Repeat("a", 64) ||
120 !reflect.DeepEqual(receipt.CgroupIdentities, []string{"cg-1"}) || !reflect.DeepEqual(receipt.UnitInvocationIDs, []string{"unit-1"}) {
121 t.Fatalf("receipt = %+v", receipt)
122 }
123}
124
125func TestClaimStopAuthenticatesBeforeFenceOrDrain(t *testing.T) {
126 c, req, _ := claimStopFixture(t)
127 drained := false
128 c.Peers = stopPeerFunc(func(context.Context) (StopPeer, error) {
129 return StopPeer{UID: 1001, PID: 42, ProcessStart: "start-42", BootID: "wrong", ControllerID: "controller-1"}, nil
130 })
131 c.Drainer = stopDrainFunc(func(context.Context, StopRequest, string) (DrainAcknowledgement, error) {
132 drained = true
133 return DrainAcknowledgement{}, nil
134 })
135 if _, err := c.FenceAndStop(t.Context(), req); !errors.Is(err, ErrClaimStopIdentity) || drained {
136 t.Fatalf("unauthenticated stop = %v drained=%v", err, drained)
137 }
138 called := false
139 err := c.Guard.With(t.Context(), req.Host, hostguard.Captured{Incarnation: req.Incarnation, Stamp: req.InstallationStamp}, func(*hostclaim.Claim) error { return nil }, func() error { called = true; return nil })
140 if err != nil || !called {
141 t.Fatalf("authentication installed fence: %v called=%v", err, called)
142 }
143}
144
145func TestClaimStopPublicationRetryKeepsBytesAndEvidenceTime(t *testing.T) {
146 c, req, store := claimStopFixture(t)
147 transport := errors.New("temporary publication failure")
148 store.err = transport
149 result, err := c.FenceAndStop(t.Context(), req)
150 if !errors.Is(err, transport) || len(result.Bytes) == 0 || result.EvidenceAt.IsZero() {
151 t.Fatalf("first publication = %+v, %v", result, err)
152 }
153 original := append([]byte(nil), result.Bytes...)
154 at := result.EvidenceAt
155 if err := c.Publish(t.Context(), result); err != nil {
156 t.Fatal(err)
157 }
158 if !reflect.DeepEqual(result.Bytes, original) || result.EvidenceAt != at {
159 t.Fatal("retry changed durable receipt")
160 }
161 if err := c.Publish(t.Context(), result); err != nil {
162 t.Fatalf("create-only exact replay = %v", err)
163 }
164}
165
166// The fence was persisted with roster numeral 5 before the crash; the retry
167// arrives after two roster writes with numeral 7. The retry resumes the
168// durable fence, and its receipt carries the persisted numeral and the
169// installation's binding (roster binding B2, N1).
170func TestClaimStopCrashAfterBeginReusesDurableFenceID(t *testing.T) {
171 c, req, _ := claimStopFixture(t)
172 original := hostguard.FenceTarget{ObligationID: req.ObligationID, ClaimGen: 19, FenceID: "fence-before-crash", Host: req.Host, Incarnation: req.Incarnation,
173 Stamp: req.InstallationStamp, Session: req.WriterSession, Op: req.OperationID, Phase: req.Phase,
174 Box: req.Box, Enrollment: req.Enrollment, BootID: req.BootID, ControllerID: req.ControllerID,
175 ClaimETag: req.ExpectedClaimETag, DirectoryRevision: req.ExpectedDirectoryRevision, RosterRevision: req.RosterRevision - 2}
176 if _, err := c.Guard.BeginFence(t.Context(), original); err != nil {
177 t.Fatal(err)
178 }
179 seen := ""
180 c.Drainer = stopDrainFunc(func(_ context.Context, _ StopRequest, fence string) (DrainAcknowledgement, error) {
181 seen = fence
182 return DrainAcknowledgement{FenceID: fence, ControllerID: req.ControllerID, ProcessStart: "start-42", Digest: strings.Repeat("a", 64)}, nil
183 })
184 result, err := c.FenceAndStop(t.Context(), req)
185 if err != nil {
186 t.Fatal(err)
187 }
188 var receipt StopReceipt
189 if err := json.Unmarshal(result.Bytes, &receipt); err != nil {
190 t.Fatal(err)
191 }
192 if seen != original.FenceID || receipt.FenceID != original.FenceID {
193 t.Fatalf("retry used fence %q, receipt %q, want durable %q", seen, receipt.FenceID, original.FenceID)
194 }
195 if result.Target != original || receipt.DirectoryRevision != "5" || receipt.Binding != req.ExpectedDirectoryRevision {
196 t.Fatalf("retry target %+v, receipt numeral %q binding %q; want the persisted fence's", result.Target, receipt.DirectoryRevision, receipt.Binding)
197 }
198 if _, err := validateTransportStop(req, result); err != nil {
199 t.Fatal("the agent refused the original numeral:", err)
200 }
201}
202
203// A fence persisted before roster binding has no numeral: a retry resumes
204// it, and it is refused before any drain, with the deny kept.
205func TestClaimStopRefusesFenceWithoutRosterNumeral(t *testing.T) {
206 c, req, _ := claimStopFixture(t)
207 legacy := hostguard.FenceTarget{ObligationID: req.ObligationID, ClaimGen: 19, FenceID: "legacy-fence", Host: req.Host, Incarnation: req.Incarnation,
208 Stamp: req.InstallationStamp, Session: req.WriterSession, Op: req.OperationID, Phase: req.Phase,
209 Box: req.Box, Enrollment: req.Enrollment, BootID: req.BootID, ControllerID: req.ControllerID,
210 ClaimETag: req.ExpectedClaimETag, DirectoryRevision: req.ExpectedDirectoryRevision}
211 if _, err := c.Guard.BeginFence(t.Context(), legacy); err != nil {
212 t.Fatal(err)
213 }
214 c.Drainer = stopDrainFunc(func(context.Context, StopRequest, string) (DrainAcknowledgement, error) {
215 t.Fatal("a fence without a numeral was drained")
216 return DrainAcknowledgement{}, nil
217 })
218 if _, err := c.FenceAndStop(t.Context(), req); !errors.Is(err, ErrClaimStopReceipt) {
219 t.Fatalf("legacy fence: %v", err)
220 }
221 if pending, err := c.Guard.PendingFences(t.Context()); err != nil || len(pending) != 1 || pending[0] != legacy {
222 t.Fatalf("legacy fence not kept: %+v %v", pending, err)
223 }
224 zero := req
225 zero.RosterRevision = 0
226 if _, err := c.FenceAndStop(t.Context(), zero); !errors.Is(err, ErrClaimStopUnavailable) {
227 t.Fatalf("a request without a numeral: %v", err)
228 }
229}
230
231func TestClaimStopCompletedRetryPublishesOriginalEvidence(t *testing.T) {
232 c, req, _ := claimStopFixture(t)
233 drains := 0
234 c.Drainer = stopDrainFunc(func(_ context.Context, _ StopRequest, fence string) (DrainAcknowledgement, error) {
235 drains++
236 return DrainAcknowledgement{FenceID: fence, ControllerID: req.ControllerID, ProcessStart: "start-42", Digest: strings.Repeat("a", 64)}, nil
237 })
238 first, err := c.FenceAndStop(t.Context(), req)
239 if err != nil {
240 t.Fatal(err)
241 }
242 second, err := c.FenceAndStop(t.Context(), req)
243 if err != nil {
244 t.Fatal(err)
245 }
246 if drains != 1 || !reflect.DeepEqual(first.Bytes, second.Bytes) || first.Key != second.Key || !first.EvidenceAt.Equal(second.EvidenceAt) {
247 t.Fatalf("completed retry recollected evidence: drains=%d first=%+v second=%+v", drains, first, second)
248 }
249}
250
251func TestClaimStopRefusesIncompleteProvidersAndEvidence(t *testing.T) {
252 c, req, _ := claimStopFixture(t)
253 c.Kernel = nil
254 if _, err := c.FenceAndStop(t.Context(), req); !errors.Is(err, ErrClaimStopUnavailable) {
255 t.Fatalf("nil kernel = %v", err)
256 }
257 c, req, _ = claimStopFixture(t)
258 c.Kernel = stopKernelFunc(func(context.Context, hostguard.FenceTarget, StopInventory, DrainAcknowledgement) (StopKernelEvidence, error) {
259 return StopKernelEvidence{}, nil
260 })
261 if _, err := c.FenceAndStop(t.Context(), req); !errors.Is(err, ErrClaimStopReceipt) {
262 t.Fatalf("incomplete evidence = %v", err)
263 }
264}
265
266// fixedRosterRevision is a fixture's verified roster read: the roster it
267// stands for is at revision n.
268func fixedRosterRevision(n uint64) func(context.Context) (uint64, error) {
269 return func(context.Context) (uint64, error) { return n, nil }
270}
271