amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package hostops
2
3import (
4 "context"
5 "crypto/ed25519"
6 "encoding/json"
7 "errors"
8 "os"
9 "strings"
10 "testing"
11 "time"
12
13 "github.com/carlosframework/platform/internal/hostactors"
14 "github.com/carlosframework/platform/internal/hostbox"
15 "github.com/carlosframework/platform/internal/hostclaim"
16 "github.com/carlosframework/platform/internal/hostevidence"
17 "github.com/carlosframework/platform/internal/release"
18)
19
20func resetTransportFixture(t *testing.T) (WakeResetTransport, WakeResetRequest, hostclaim.Observed, ed25519.PrivateKey) {
21 t.Helper()
22 w, o, clock := wakeResetFixture(t)
23 key := ed25519.NewKeyFromSeed(make([]byte, 32))
24 boxkey := ed25519.NewKeyFromSeed([]byte(strings.Repeat("b", 32)))
25 h := o.Claim.Holder
26 h.WakeRetry.Attempt.Box = "box"
27 h.WakeRetry.Attempt.Enrollment = strings.Repeat("a", 32)
28 h.WakeRetry.Attempt.BootID = "boot"
29 h.WakeRetry.Attempt.ControllerID = "controller"
30 h.WakeRetry.Attempt.Session = h.Session
31 raw, _ := json.Marshal(o.Claim)
32 etag, err := w.raw.Put(t.Context(), "control/hosts/"+o.Host+".json", raw, o.ETag)
33 if err != nil {
34 t.Fatal(err)
35 }
36 o.ETag = etag
37 r := hostactors.Roster{V: 1, DeploymentBucket: "bucket", Revision: 1, RequestLimitNS: int64(hostbox.RequestLimit), Boxes: map[string]hostactors.Box{"box": {BoxID: "box", Enrollment: strings.Repeat("a", 32), Namespace: "platform", Label: "box", Region: "ie", Pool: "default", Cloud: hostactors.Cloud{Provider: "aws", Account: "123", Region: "eu-west-1", InstanceID: "i-box"}, EvidenceSHA256: strings.Repeat("c", 64), State: "active", CensusPublicKey: boxkey.Public().(ed25519.PublicKey)}}, Consoles: map[string]hostactors.Console{"console": {InstanceID: "console", Enrollment: strings.Repeat("d", 32), Namespaces: []string{"platform"}, DeploymentBucket: "bucket", Compute: &hostactors.ComputeBinding{Box: "box", Enrollment: strings.Repeat("a", 32)}}}, Holds: map[string]hostactors.Reservation{}, Activations: map[string]hostactors.Reservation{}}
38 wire, err := release.SignRootEnvelope(release.RootSigner(key), r)
39 if err != nil {
40 t.Fatal(err)
41 }
42 if _, err = w.raw.Put(t.Context(), hostactors.Key, wire, ""); err != nil {
43 t.Fatal(err)
44 }
45 transport := WakeResetTransport{Store: w.raw, Directory: hostactors.Directory{Store: w.raw, PublicKey: key.Public().(ed25519.PublicKey), Bucket: "bucket"}, Now: clock}
46 req := WakeResetRequest{V: 1, ID: strings.Repeat("e", 64), Bucket: "bucket", Console: "console", ConsoleEnrollment: strings.Repeat("d", 32), Namespace: "platform", Host: o.Host, ClaimETag: o.ETag, ClaimGen: o.Claim.Gen, Incarnation: h.Incarnation, Stamp: h.Stamp, Session: h.Session, Operation: o.Claim.Op.ID, Phase: o.Claim.Op.Phase, Episode: h.WakeRetry.EpisodeID, Attempt: *h.WakeRetry.Attempt, IssuedAt: clock(), Expires: clock().Add(time.Minute)}
47 owner := RegistrationOwner{Console: req.Console, Enrollment: req.ConsoleEnrollment, Controller: "controller"}
48 _, err = w.RegisterStart(t.Context(), h.Account, h.App, "app", "reset", Registration{Kind: RegistrationWakeRetryReset, Host: o.Host, Owner: owner, AppID: "app", Epoch: 1, Claim: SavedObservation(o), Incarnation: h.Incarnation, CreatedAt: clock()}, clock())
49 if err != nil {
50 t.Fatal(err)
51 }
52 req.Admission = WakeResetAdmission{Account: h.Account, App: h.App, AppID: "app", Registration: "reset", Epoch: 1, Owner: owner}
53 req.ID = wakeResetID(req)
54 return transport, req, o, boxkey
55}
56
57func TestPlan9ResetRequestRejectsChangedAuthorityAndClaim(t *testing.T) {
58 tr, r, o, _ := resetTransportFixture(t)
59 if err := tr.Authorize(t.Context(), r); err != nil {
60 t.Fatal(err)
61 }
62 for _, change := range []func(*WakeResetRequest){func(r *WakeResetRequest) { r.ConsoleEnrollment = "other" }, func(r *WakeResetRequest) { r.ClaimETag = "other" }, func(r *WakeResetRequest) { r.Incarnation = "other" }, func(r *WakeResetRequest) { r.Attempt.ID = "other" }, func(r *WakeResetRequest) { r.Expires = r.IssuedAt }, func(r *WakeResetRequest) { r.Namespace = "other" }} {
63 copy := r
64 change(&copy)
65 if tr.Authorize(t.Context(), copy) == nil {
66 t.Fatal("accepted changed binding")
67 }
68 }
69 o.Claim.Holder.WakeRetry.Attempt.ID = "replacement"
70 raw, _ := json.Marshal(o.Claim)
71 if _, err := tr.Store.Put(t.Context(), "control/hosts/"+o.Host+".json", raw, o.ETag); err != nil {
72 t.Fatal(err)
73 }
74 if tr.Authorize(t.Context(), r) == nil {
75 t.Fatal("accepted replaced attempt")
76 }
77}
78
79func TestPlan9ResetLostResponseIsUncertain(t *testing.T) {
80 tr, r, _, _ := resetTransportFixture(t)
81 if err := tr.VerifyReceipt(t.Context(), r); !errors.Is(err, ErrWakeResetUncertain) {
82 t.Fatalf("absence = %v", err)
83 }
84 ctx, cancel := context.WithCancel(t.Context())
85 cancel()
86 if err := tr.Wait(ctx, r, time.Millisecond); !errors.Is(err, ErrWakeResetUncertain) {
87 t.Fatalf("cancel = %v", err)
88 }
89}
90
91func publishResetProof(t *testing.T, tr WakeResetTransport, r WakeResetRequest, key ed25519.PrivateKey) string {
92 t.Helper()
93 k := "control/hosts/" + r.Host + "/serving-evidence/box/proof.json"
94 proof := StopReceipt{Region: "ie", V: 1, ID: "proof", Host: r.Host, Incarnation: r.Incarnation, InstallationStamp: r.Stamp, WriterSession: r.Session, OperationID: r.Operation, Phase: r.Phase, Box: r.Attempt.Box, Enrollment: r.Attempt.Enrollment, BootID: r.Attempt.BootID, ControllerID: r.Attempt.ControllerID, ClaimETag: r.ClaimETag, ClaimGen: r.ClaimGen, Kind: "stopped", EvidenceAt: tr.Now(), FenceID: "fence", DirectoryRevision: "1", RequestLimitNS: int64(hostbox.RequestLimit), InventorySHA256: strings.Repeat("1", 64), LocalJournalSHA256: strings.Repeat("2", 64), ProducerDrainDigest: strings.Repeat("3", 64), CgroupIdentities: []string{"group"}}
95 raw, _ := json.Marshal(proof)
96 if err := hostevidence.Publish(t.Context(), tr.Store, key, k, raw); err != nil {
97 t.Fatal(err)
98 }
99 return k
100}
101func TestPlan9ResetReceiptNeedsRootSignatureAndExactTuple(t *testing.T) {
102 tr, r, _, key := resetTransportFixture(t)
103 k := publishResetProof(t, tr, r, key)
104 if err := tr.VerifyReceipt(t.Context(), r); err != nil {
105 t.Fatal(err)
106 }
107 changed := r
108 changed.Attempt.ControllerID = "other"
109 if tr.VerifyReceipt(t.Context(), changed) == nil {
110 t.Fatal("accepted other controller")
111 }
112 raw, etag, err := tr.Store.Get(t.Context(), k)
113 if err != nil {
114 t.Fatal(err)
115 }
116 raw = append(raw, ' ')
117 if _, err = tr.Store.Put(t.Context(), k, raw, etag); err != nil {
118 t.Fatal(err)
119 }
120 if tr.VerifyReceipt(t.Context(), r) == nil {
121 t.Fatal("accepted forged proof")
122 }
123}
124
125func signedResetRequest(t *testing.T, tr WakeResetTransport, r WakeResetRequest) {
126 t.Helper()
127 key := ed25519.NewKeyFromSeed(make([]byte, 32))
128 wire, err := release.SignRootEnvelope(release.RootSigner(key), r)
129 if err != nil {
130 t.Fatal(err)
131 }
132 if _, err = tr.Store.Put(t.Context(), r.Key(), wire, ""); err != nil {
133 t.Fatal(err)
134 }
135}
136func TestPlan9ResetWorkerReplayCancellationAndExpiry(t *testing.T) {
137 for _, mode := range []string{"dispatch", "cancel", "expiry", "invalid-signature", "completed", "lost-response"} {
138 t.Run(mode, func(t *testing.T) {
139 tr, r, _, key := resetTransportFixture(t)
140 if mode == "expiry" {
141 r.IssuedAt = tr.Now().Add(-2 * time.Minute)
142 r.Expires = tr.Now().Add(-time.Minute)
143 }
144 signedResetRequest(t, tr, r)
145 if mode == "cancel" {
146 raw, _ := json.Marshal(r)
147 wire, err := release.SignRootEnvelope(release.RootSigner(ed25519.NewKeyFromSeed(make([]byte, 32))), WakeResetCancellation{V: 1, RequestSHA256: evidenceDigest(raw)})
148 if err != nil {
149 t.Fatal(err)
150 }
151 if _, err = tr.Store.Put(t.Context(), r.Key()+".cancel", wire, ""); err != nil {
152 t.Fatal(err)
153 }
154 }
155 if mode == "invalid-signature" {
156 raw, tag, _ := tr.Store.Get(t.Context(), r.Key())
157 raw[0] = '['
158 if _, err := tr.Store.Put(t.Context(), r.Key(), raw, tag); err != nil {
159 t.Fatal(err)
160 }
161 }
162 if mode == "completed" {
163 publishResetProof(t, tr, r, key)
164 }
165 calls := 0
166 stop := func(ctx context.Context, o hostclaim.Observed, a hostclaim.WakeRetryAttempt) error {
167 calls++
168 if !r.matches(o) || a != r.Attempt {
169 t.Fatal("wrong attempt")
170 }
171 publishResetProof(t, tr, r, key)
172 if mode == "lost-response" {
173 return errors.New("connection lost")
174 }
175 return nil
176 }
177 err := tr.Pass(t.Context(), "box", stop)
178 want := 0
179 if mode == "dispatch" || mode == "lost-response" {
180 want = 1
181 }
182 if calls != want {
183 t.Fatalf("calls=%d want=%d err=%v", calls, want, err)
184 }
185 if mode == "invalid-signature" && err == nil {
186 t.Fatal("accepted unsigned request")
187 }
188 if mode == "lost-response" && !errors.Is(err, ErrWakeResetUncertain) {
189 t.Fatalf("lost stop = %v", err)
190 }
191 if mode == "dispatch" || mode == "lost-response" || mode == "completed" {
192 if err := tr.Pass(t.Context(), "box", stop); err != nil {
193 t.Fatal(err)
194 }
195 if calls != want {
196 t.Fatal("replayed completed stop")
197 }
198 if err := tr.VerifyReceipt(t.Context(), r); err != nil {
199 t.Fatal(err)
200 }
201 }
202 })
203 }
204}
205
206func TestPlan9ResetTypedWriterRequiresLiveExactRegistration(t *testing.T) {
207 tr, r, o, _ := resetTransportFixture(t)
208 w, err := NewControlWriter(tr.Store, tr.Directory.PublicKey, tr.Now)
209 if err != nil {
210 t.Fatal(err)
211 }
212 signer := release.RootSigner(ed25519.NewKeyFromSeed(make([]byte, 32)))
213 issue := func(ctx context.Context) (WakeResetRequest, error) {
214 return w.RequestWakeReset(ctx, tr, signer, r.Console, r.ConsoleEnrollment, r.Namespace, o, r.Attempt)
215 }
216 if _, err := issue(t.Context()); !errors.Is(err, ErrIdentity) {
217 t.Fatalf("unregistered issue=%v", err)
218 }
219 b := r.Admission
220 grant := lifecycleGrant{kind: "app", account: b.Account, app: b.App, appID: b.AppID, operation: b.Registration, owner: b.Owner, epoch: b.Epoch, w: w, registration: RegistrationWakeRetryReset, incarnation: r.Incarnation}
221 ctx := withLifecycleGrant(t.Context(), grant)
222 first, err := issue(ctx)
223 if err != nil {
224 t.Fatal(err)
225 }
226 replay, err := issue(ctx)
227 if err != nil || replay != first {
228 t.Fatalf("replay=%+v err=%v", replay, err)
229 }
230 if err = w.CancelWakeReset(t.Context(), tr, signer, first); err != nil {
231 t.Fatal(err)
232 }
233 cancelled, err := tr.cancelled(t.Context(), first)
234 if err != nil || !cancelled {
235 t.Fatalf("cancel=%t %v", cancelled, err)
236 }
237 wrong := grant
238 wrong.incarnation = "another"
239 if _, err := issue(withLifecycleGrant(t.Context(), wrong)); !errors.Is(err, ErrIdentity) {
240 t.Fatalf("other incarnation=%v", err)
241 }
242}
243
244func TestPlan9ResetNewControlObjectsNeverPassDataBoundary(t *testing.T) {
245 for _, key := range []string{WakeResetPrefix + "box/id.json", WakeResetPrefix + "box/id.json.cancel"} {
246 if !IsControlOnlyKey(key) {
247 t.Fatalf("not control-only: %s", key)
248 }
249 }
250}
251
252func TestPlan9ResetArchivePreventsResurrectionAndKeepsProof(t *testing.T) {
253 tr, r, _, boxkey := resetTransportFixture(t)
254 root := ed25519.NewKeyFromSeed(make([]byte, 32))
255 raw, err := release.SignRootEnvelope(release.RootSigner(root), r)
256 if err != nil {
257 t.Fatal(err)
258 }
259 if _, err = tr.Store.Put(t.Context(), r.Key(), raw, ""); err != nil {
260 t.Fatal(err)
261 }
262 publishResetProof(t, tr, r, boxkey)
263 now := r.Expires.Add(hostbox.RequestLimit + time.Second)
264 tr.Now = func() time.Time { return now }
265 writer := &ControlWriter{raw: tr.Store, pub: root.Public().(ed25519.PublicKey), now: tr.Now}
266 if err = writer.SweepWakeResets(t.Context(), tr); err != nil {
267 t.Fatal(err)
268 }
269 if _, _, err = tr.Store.Get(t.Context(), r.Key()); !errors.Is(err, os.ErrNotExist) {
270 t.Fatalf("expired request retained: %v", err)
271 }
272 if _, _, err = tr.Store.Get(t.Context(), wakeResetArchiveKey(r)); err != nil {
273 t.Fatal(err)
274 }
275 if err = tr.VerifyReceipt(t.Context(), r); err != nil {
276 t.Fatalf("archive lost proof: %v", err)
277 }
278 // A delayed reissue that passed its first archive read cannot dispatch even
279 // if it recreated the original request key with a later expiry.
280 revived := r
281 revived.IssuedAt = now
282 revived.Expires = now.Add(time.Minute)
283 if revived.ID != wakeResetID(revived) {
284 t.Fatal("fixture did not reuse deterministic id")
285 }
286 if err = tr.Authorize(t.Context(), revived); err == nil {
287 t.Fatal("archived request resurrected")
288 }
289 if !IsControlOnlyKey(wakeResetArchiveKey(r)) {
290 t.Fatal("archive not control-only")
291 }
292}
293func TestPlan9ResetStopDeadlineIsFairAndExpiryBounded(t *testing.T) {
294 now := time.Now()
295 if got := minResetDeadline(now, now.Add(time.Minute)); !got.Equal(now.Add(hostbox.RequestLimit / 4)) {
296 t.Fatal(got)
297 }
298 if got := minResetDeadline(now, now.Add(time.Second)); !got.Equal(now.Add(time.Second)) {
299 t.Fatal(got)
300 }
301}
302
303func TestPlan9ArchivedResetRejectsLateCancellation(t *testing.T) {
304 tr, r, _, _ := resetTransportFixture(t)
305 root := ed25519.NewKeyFromSeed(make([]byte, 32))
306 raw, _ := release.SignRootEnvelope(release.RootSigner(root), r)
307 if _, err := tr.Store.Put(t.Context(), wakeResetArchiveKey(r), raw, ""); err != nil {
308 t.Fatal(err)
309 }
310 writer := &ControlWriter{raw: tr.Store, pub: root.Public().(ed25519.PublicKey), now: tr.Now}
311 if err := writer.CancelWakeReset(t.Context(), tr, release.RootSigner(root), r); err == nil {
312 t.Fatal("late cancellation recreated an orphan")
313 }
314 if _, _, err := tr.Store.Get(t.Context(), r.Key()+".cancel"); !errors.Is(err, os.ErrNotExist) {
315 t.Fatal(err)
316 }
317}
318
319type plan9CancelDeleteStore struct {
320 release.Store
321 fail bool
322 deleted []string
323}
324
325func (s *plan9CancelDeleteStore) DeleteIfMatch(ctx context.Context, key, etag string) error {
326 s.deleted = append(s.deleted, key)
327 if s.fail && strings.HasSuffix(key, ".cancel") {
328 return errors.New("cancel delete unavailable")
329 }
330 return release.DeleteIfMatch(ctx, s.Store, key, etag)
331}
332func TestPlan9ResetSweepRetainsRequestUntilCancellationRemoved(t *testing.T) {
333 tr, r, _, _ := resetTransportFixture(t)
334 root := ed25519.NewKeyFromSeed(make([]byte, 32))
335 raw, _ := release.SignRootEnvelope(release.RootSigner(root), r)
336 if _, err := tr.Store.Put(t.Context(), r.Key(), raw, ""); err != nil {
337 t.Fatal(err)
338 }
339 writer := &ControlWriter{raw: tr.Store, pub: root.Public().(ed25519.PublicKey), now: tr.Now}
340 if err := writer.CancelWakeReset(t.Context(), tr, release.RootSigner(root), r); err != nil {
341 t.Fatal(err)
342 }
343 now := r.Expires.Add(hostbox.RequestLimit + time.Second)
344 tr.Now = func() time.Time { return now }
345 writer.now = tr.Now
346 store := &plan9CancelDeleteStore{Store: tr.Store, fail: true}
347 writer.raw = store
348 if err := writer.SweepWakeResets(t.Context(), tr); err == nil {
349 t.Fatal("ignored cancellation deletion failure")
350 }
351 if _, _, err := tr.Store.Get(t.Context(), r.Key()); err != nil {
352 t.Fatalf("removed retry discovery: %v", err)
353 }
354 if len(store.deleted) != 1 || store.deleted[0] != r.Key()+".cancel" {
355 t.Fatalf("wrong deletion order: %v", store.deleted)
356 }
357 store.fail = false
358 store.deleted = nil
359 if err := writer.SweepWakeResets(t.Context(), tr); err != nil {
360 t.Fatal(err)
361 }
362 if len(store.deleted) != 2 || store.deleted[0] != r.Key()+".cancel" || store.deleted[1] != r.Key() {
363 t.Fatalf("wrong successful order: %v", store.deleted)
364 }
365}
366