| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "crypto" |
| 7 | "crypto/rand" |
| 8 | "crypto/rsa" |
| 9 | "crypto/sha256" |
| 10 | "crypto/x509" |
| 11 | "encoding/base64" |
| 12 | "encoding/json" |
| 13 | "errors" |
| 14 | "fmt" |
| 15 | "net" |
| 16 | "net/http" |
| 17 | "os" |
| 18 | "path/filepath" |
| 19 | "strings" |
| 20 | "sync" |
| 21 | "testing" |
| 22 | "time" |
| 23 | |
| 24 | "github.com/carlosframework/platform/internal/boxid" |
| 25 | "github.com/carlosframework/platform/internal/hostbox" |
| 26 | "github.com/carlosframework/platform/internal/hostclaim" |
| 27 | "github.com/carlosframework/platform/internal/hostguard" |
| 28 | "github.com/carlosframework/platform/internal/provision" |
| 29 | "github.com/carlosframework/platform/internal/registry" |
| 30 | ) |
| 31 | |
| 32 | type handoverRecoveryRoster struct{ actor hostbox.Actor } |
| 33 | |
| 34 | func (r handoverRecoveryRoster) Actor(context.Context, string) (hostbox.Actor, error) { |
| 35 | return r.actor, nil |
| 36 | } |
| 37 | |
| 38 | type handoverRecoveryEvidence []byte |
| 39 | |
| 40 | func (e handoverRecoveryEvidence) Load(context.Context, string) ([]byte, error) { |
| 41 | return e, nil |
| 42 | } |
| 43 | |
| 44 | // handoverRecoveryKey is the RSA key every test identity signs its document |
| 45 | // with. Generating a 2048-bit key costs tens of milliseconds and this helper |
| 46 | // has dozens of callers, many in loops, so the process makes one. Each |
| 47 | // provider still carries its own document, signature and certificate map, and |
| 48 | // no test verifies one provider's evidence against another's certificates. |
| 49 | var handoverRecoveryKey = sync.OnceValues(func() (*rsa.PrivateKey, error) { |
| 50 | return rsa.GenerateKey(rand.Reader, 2048) |
| 51 | }) |
| 52 | |
| 53 | func handoverRecoveryIdentity(t *testing.T, actor hostbox.Actor) hostbox.IdentityProvider { |
| 54 | t.Helper() |
| 55 | key, err := handoverRecoveryKey() |
| 56 | if err != nil { |
| 57 | t.Fatal(err) |
| 58 | } |
| 59 | actor.Bucket, actor.Namespace, actor.Eligible = "data", "platform", true |
| 60 | if actor.Binding == "" { |
| 61 | // Fixture shorthand: these fixtures stamp installed state with the |
| 62 | // actor's numeral, so the opaque binding defaults to that string. The |
| 63 | // production roster computes hostactors.BoxBinding; the binding |
| 64 | // witnesses (claim_binding_*_test.go) run on it. |
| 65 | actor.Binding = fmt.Sprint(actor.Revision) |
| 66 | } |
| 67 | actor.Cloud = hostbox.CloudIdentity{Provider: "aws", Account: "111122223333", Region: "eu-west-1", InstanceID: "i-0abc"} |
| 68 | doc, err := json.Marshal(map[string]string{"accountId": actor.Cloud.Account, "region": actor.Cloud.Region, "instanceId": actor.Cloud.InstanceID}) |
| 69 | if err != nil { |
| 70 | t.Fatal(err) |
| 71 | } |
| 72 | sum := sha256.Sum256(doc) |
| 73 | signature, err := rsa.SignPKCS1v15(rand.Reader, key, crypto.SHA256, sum[:]) |
| 74 | if err != nil { |
| 75 | t.Fatal(err) |
| 76 | } |
| 77 | raw, err := json.Marshal(boxid.Record{V: 1, Box: actor.Box, IdentityDocument: base64.StdEncoding.EncodeToString(doc), IdentitySignature: base64.StdEncoding.EncodeToString(signature)}) |
| 78 | if err != nil { |
| 79 | t.Fatal(err) |
| 80 | } |
| 81 | return hostbox.IdentityProvider{Account: actor.Cloud.Account, Box: actor.Box, Region: actor.Region, AWSRegion: actor.Cloud.Region, Bucket: actor.Bucket, Namespace: actor.Namespace, Roster: handoverRecoveryRoster{actor}, Evidence: handoverRecoveryEvidence(raw), Certs: map[string]*x509.Certificate{actor.Cloud.Region: {PublicKey: &key.PublicKey, PublicKeyAlgorithm: x509.RSA}}} |
| 82 | } |
| 83 | |
| 84 | func postRecoveryHandover(t *testing.T, socket string, request coldHandoverRequest) int { |
| 85 | t.Helper() |
| 86 | body, err := json.Marshal(request) |
| 87 | if err != nil { |
| 88 | t.Fatal(err) |
| 89 | } |
| 90 | transport := &http.Transport{DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { |
| 91 | return (&net.Dialer{}).DialContext(ctx, "unix", socket) |
| 92 | }} |
| 93 | defer transport.CloseIdleConnections() |
| 94 | req, err := http.NewRequestWithContext(t.Context(), http.MethodPost, "http://instance-control/claim/handover", bytes.NewReader(body)) |
| 95 | if err != nil { |
| 96 | t.Fatal(err) |
| 97 | } |
| 98 | response, err := (&http.Client{Transport: transport}).Do(req) |
| 99 | if err != nil { |
| 100 | t.Fatal(err) |
| 101 | } |
| 102 | defer response.Body.Close() |
| 103 | return response.StatusCode |
| 104 | } |
| 105 | |
| 106 | // The kernel and lease are fixtures. Recovery, persisted receipts, the Unix |
| 107 | // root endpoint, cold handover, writer admission and next restore are real. |
| 108 | func TestClaimColdHandoverAfterControllerRecoveryReachesNextWake(t *testing.T) { |
| 109 | t.Parallel() |
| 110 | for _, boot := range []string{"boot-1", "boot-2"} { |
| 111 | t.Run(boot, func(t *testing.T) { |
| 112 | ctx := t.Context() |
| 113 | f := newRecoveryFixture(t) |
| 114 | // Production reads claims and publishes evidence in the same bucket. |
| 115 | f.evidence = f.claims |
| 116 | recordedBoot, parkRuntime := servingRecoveryFilesystem(t, f.guard.DataDir) |
| 117 | if boot == "boot-1" { |
| 118 | boot = recordedBoot |
| 119 | } |
| 120 | former := f.controller(recordedBoot, "controller-1", 4101, "start-1") |
| 121 | admitted := former.admit(t, f) |
| 122 | f.revision = 8 |
| 123 | restarted := f.controller(boot, "controller-2", 4202, "start-2") |
| 124 | route := registry.Route{Host: recoveryHost, Kind: "instance", Account: admitted.Claim.Holder.Account, App: admitted.Claim.Holder.App, DBPath: provision.NestedDBPath(f.guard.DataDir, recoveryHost), Socket: provision.NestedSocket(recoveryHost), ClaimOwnership: registry.ClaimOwnership{ClaimIncarnation: admitted.Claim.Holder.Incarnation, ClaimStamp: admitted.Claim.Holder.Stamp, ClaimOwner: "old-installation", ClaimRowID: "old-route"}} |
| 125 | f.kernel.alive = true |
| 126 | if err := restarted.runtime.Prerequisites(ctx, route, f.read(), recoveryPolicy); !errors.Is(err, ErrClaimStopControllerAlive) { |
| 127 | t.Fatalf("live predecessor accepted: %v", err) |
| 128 | } |
| 129 | f.kernel.alive = false |
| 130 | if err := restarted.runtime.Prerequisites(ctx, route, f.read(), recoveryPolicy); !errors.Is(err, hostbox.ErrChanged) { |
| 131 | t.Fatalf("restart recovery: %v", err) |
| 132 | } |
| 133 | if retry := f.retry(); retry.State != "open" || retry.AttemptsReserved != 1 || retry.Attempt != nil { |
| 134 | t.Fatalf("recovery refunded or stranded attempt: %+v", retry) |
| 135 | } |
| 136 | prior := f.read() |
| 137 | receipt, err := f.guard.CompletedColdPredecessor(ctx, prior.Host, hostguard.Captured{Incarnation: prior.Claim.Holder.Incarnation, Stamp: prior.Claim.Holder.Stamp}, prior.Claim.Holder.Session, prior.Claim.Op.ID, prior.Claim.Op.Phase) |
| 138 | if err != nil { |
| 139 | t.Fatal(err) |
| 140 | } |
| 141 | if receipt.Target.ControllerID != "controller-1" || receipt.Target.BootID != recordedBoot || receipt.Target.DirectoryRevision != "7" { |
| 142 | t.Fatalf("recovered tuple was rewritten: %+v", receipt.Target) |
| 143 | } |
| 144 | base := t.TempDir() |
| 145 | inventory, err := openLegacyInventory(t, filepath.Join(base, "inventory.db")) |
| 146 | if err != nil { |
| 147 | t.Fatal(err) |
| 148 | } |
| 149 | t.Cleanup(func() { inventory.Close() }) |
| 150 | reg, err := registry.Open(filepath.Join(base, "registry.db")) |
| 151 | if err != nil { |
| 152 | t.Fatal(err) |
| 153 | } |
| 154 | t.Cleanup(func() { reg.Close() }) |
| 155 | if err = reg.Upsert(route); err != nil { |
| 156 | t.Fatal(err) |
| 157 | } |
| 158 | owner := route.ClaimOwnership |
| 159 | in := hostbox.StopInventory{Complete: true, Host: route.Host, Incarnation: owner.ClaimIncarnation, InstallationStamp: owner.ClaimStamp, Box: "box-1", Enrollment: "enrollment-1", DirectoryRevision: "7", Entries: []hostbox.StopInventoryEntry{{Kind: "route", ArtifactID: owner.ClaimRowID, ManagedRoot: "registry", RelativePath: route.Host, RowIdentity: owner.ClaimRowID, Ownership: owner.ClaimOwner, Materialization: "materialized"}}} |
| 160 | if err = os.MkdirAll(filepath.Dir(route.DBPath), 0700); err != nil { |
| 161 | t.Fatal(err) |
| 162 | } |
| 163 | rel, err := filepath.Rel(f.guard.DataDir, filepath.Dir(route.DBPath)) |
| 164 | if err != nil { |
| 165 | t.Fatal(err) |
| 166 | } |
| 167 | roots := map[string]string{"data": f.guard.DataDir} |
| 168 | a, err := (hostbox.TypedParker{Roots: roots, ParkedRoots: roots, Incarnation: owner.ClaimIncarnation, Host: route.Host, Retirement: "fixture"}).CaptureArtifact("data", rel, owner.ClaimOwner, hostbox.ArtifactPath) |
| 169 | if err != nil { |
| 170 | t.Fatal(err) |
| 171 | } |
| 172 | in.Entries = append(in.Entries, hostbox.StopInventoryEntry{Kind: "path", ArtifactID: "state", ManagedRoot: "data", RelativePath: rel, Ownership: owner.ClaimOwner, Materialization: "materialized", Device: a.Dev, Inode: a.Ino, Type: a.Type}) |
| 173 | holder := prior.Claim.Holder |
| 174 | body, err := json.Marshal(provision.Record{Schema: 1, Host: prior.Host, Account: holder.Account, App: holder.App, CreatedAt: holder.CreatedAt, Channel: "edge"}) |
| 175 | if err != nil { |
| 176 | t.Fatal(err) |
| 177 | } |
| 178 | if _, err = f.claims.Put(ctx, holder.Key, body, ""); err != nil { |
| 179 | t.Fatal(err) |
| 180 | } |
| 181 | if err = inventory.WriteStopInventory(ctx, in); err != nil { |
| 182 | t.Fatal(err) |
| 183 | } |
| 184 | |
| 185 | rootGuard := f.guard |
| 186 | rootGuard.Root = hostguard.NewRootAuthorityWithInventory(&rollbackDisk{dir: t.TempDir()}, func(ctx context.Context, host string, c hostguard.Captured) (uint64, error) { |
| 187 | _, revision, err := inventory.Snapshot(ctx, host, c) |
| 188 | return revision, err |
| 189 | }) |
| 190 | held := prior.Claim.Holder |
| 191 | if err := rootGuard.EstablishRoot(ctx, hostguard.RootInstallation{Host: prior.Host, Incarnation: held.Incarnation, Stamp: held.Stamp, Account: held.Account, App: held.App, Record: held.Key, Box: "box-1", Enrollment: "enrollment-1", Migration: 1, InventoryRevision: 1}, func(*hostclaim.Claim) error { return nil }); err != nil { |
| 192 | t.Fatal(err) |
| 193 | } |
| 194 | observedFence := prior |
| 195 | observedFence.ETag = receipt.Target.ClaimETag |
| 196 | rootGuard.Claims = rootViewClaimReader{obs: observedFence} |
| 197 | if _, err := rootGuard.BeginFence(ctx, receipt.Target); err != nil { |
| 198 | t.Fatal("root fence setup", err) |
| 199 | } |
| 200 | if err := rootGuard.CompleteFence(ctx, receipt.Target, receipt.Bytes, receipt.EvidenceAt, func(_ context.Context, target hostguard.FenceTarget, body []byte, at time.Time) error { |
| 201 | return verifyRetirementReceipt(ctx, hostguard.FenceReceipt{Target: target, Bytes: body, EvidenceAt: at}) |
| 202 | }); err != nil { |
| 203 | t.Fatal("root receipt setup", err) |
| 204 | } |
| 205 | rootGuard.Claims = f.guard.Claims |
| 206 | coordinator := f.root(recoveryStopDrain{}) |
| 207 | coordinator.Guard = rootGuard |
| 208 | root := ClaimRootHTTP{Launches: f.journal, Stop: ClaimStopHTTP{Coordinator: coordinator, Authenticate: func(context.Context, net.Conn) (StopPeer, error) { return f.peer, nil }}, Roots: map[string]string{"data": f.guard.DataDir}, Reconcile: hostbox.ReconcileAdapter{Box: "box-1", Enrollment: "enrollment-1", Registry: reg, Inventory: inventory, Retirement: hostbox.TypedRetirement{JournalDir: filepath.Join(base, "retire")}}} |
| 209 | root.servingParks = parkRuntime |
| 210 | root.RequireNewWork = func(context.Context) error { return nil } |
| 211 | root.InventoryActor = func(context.Context) (hostbox.Actor, error) { |
| 212 | actor := restarted.local.Actor |
| 213 | actor.Eligible = true |
| 214 | return actor, nil |
| 215 | } |
| 216 | socket := startRetirementRecoveryTransport(t, root) |
| 217 | unauthenticated := root |
| 218 | unauthenticated.Stop.Authenticate = func(context.Context, net.Conn) (StopPeer, error) { return StopPeer{}, errors.New("unknown requester") } |
| 219 | refusedSocket := startRetirementRecoveryTransport(t, unauthenticated) |
| 220 | hook := coldAdmissionHook(f.guard, handoverRecoveryIdentity(t, restarted.local.Actor), socket) |
| 221 | restarted.wake.AfterAdmission = func(ctx context.Context, before, after hostclaim.Observed) error { |
| 222 | h := after.Claim.Holder |
| 223 | request := coldHandoverRequest{Prior: receipt.Target, Next: hostbox.ColdSuccessor{Host: after.Host, Incarnation: h.Incarnation, Stamp: h.Stamp, Session: h.Session, Generation: h.Generation, Box: "box-1", Enrollment: "enrollment-1", DirectoryRevision: recoveryFixtureBinding, ETag: after.ETag, Op: after.Claim.Op.ID, Phase: after.Claim.Op.Phase}} |
| 224 | if status := postRecoveryHandover(t, refusedSocket, request); status != http.StatusForbidden { |
| 225 | t.Fatalf("receipt authenticated an unknown requester: %d", status) |
| 226 | } |
| 227 | for _, field := range []string{"controller", "boot", "revision", "box", "enrollment"} { |
| 228 | forged := request |
| 229 | switch field { |
| 230 | case "controller": |
| 231 | forged.Prior.ControllerID = "unknown-controller" |
| 232 | case "boot": |
| 233 | forged.Prior.BootID = "unknown-boot" |
| 234 | case "revision": |
| 235 | forged.Prior.DirectoryRevision = "8" |
| 236 | case "box": |
| 237 | forged.Prior.Box = "other-box" |
| 238 | case "enrollment": |
| 239 | forged.Prior.Enrollment = "other-enrollment" |
| 240 | } |
| 241 | if status := postRecoveryHandover(t, socket, forged); status != http.StatusBadRequest { |
| 242 | t.Fatalf("forged %s accepted: %d", field, status) |
| 243 | } |
| 244 | } |
| 245 | if row, err := reg.Get(route.Host); err != nil || row == nil || row.ClaimRowID != route.ClaimRowID { |
| 246 | t.Fatalf("refusal changed predecessor row: %+v %v", row, err) |
| 247 | } |
| 248 | if status := postRecoveryHandover(t, socket, request); status != http.StatusNoContent { |
| 249 | t.Fatalf("exact completed predecessor receipt refused: HTTP %d", status) |
| 250 | } |
| 251 | return hook(ctx, before, after) |
| 252 | } |
| 253 | // The fixture supplies the lease; the real Admit invokes root's cold |
| 254 | // handover before activation, prelaunch persistence and retry binding. |
| 255 | f.now = f.retry().NotBefore |
| 256 | next := restarted.admit(t, f) |
| 257 | if next.Claim.Holder.Session <= admitted.Claim.Holder.Session { |
| 258 | t.Fatal("next wake reused predecessor session") |
| 259 | } |
| 260 | if retry := f.retry(); retry.AttemptsReserved != 2 || retry.Attempt == nil || retry.Attempt.ControllerID != "controller-2" || retry.Attempt.BootID != boot { |
| 261 | t.Fatalf("next attempt not charged to current requester: %+v", retry) |
| 262 | } |
| 263 | currentRow, err := reg.Get(route.Host) |
| 264 | if err != nil || currentRow == nil || currentRow.ClaimStamp != next.Claim.Holder.Stamp || currentRow.ClaimRowID == route.ClaimRowID { |
| 265 | t.Fatalf("next wake ownership: %+v %v", currentRow, err) |
| 266 | } |
| 267 | prepared, err := restarted.runtime.Prepare(ctx, *currentRow, next, recoveryPolicy) |
| 268 | if err != nil { |
| 269 | t.Fatal(err) |
| 270 | } |
| 271 | if err = prepared.Restore(ctx); err != nil { |
| 272 | t.Fatalf("next wake restore: %v", err) |
| 273 | } |
| 274 | marker := filepath.Join(f.guard.DataDir, "state", "data.db.gen") |
| 275 | if body, err := os.ReadFile(marker); err != nil || strings.TrimSpace(string(body)) != next.Claim.Holder.Generation { |
| 276 | t.Fatalf("next wake generation: %q %v", body, err) |
| 277 | } |
| 278 | replayed, err := rootGuard.FenceReceiptFor(ctx, receipt.Target) |
| 279 | if err != nil || !bytes.Equal(replayed.Bytes, receipt.Bytes) || !replayed.EvidenceAt.Equal(receipt.EvidenceAt) || replayed.Target != receipt.Target { |
| 280 | t.Fatalf("handover rewrote predecessor evidence: %+v %v", replayed, err) |
| 281 | } |
| 282 | launch, err := f.journal.Read(ctx, recoveryTuple(next, restarted.local)) |
| 283 | if err != nil || launch.ControllerPID != 4202 || launch.ProcessStart != "start-2" || launch.Target.DirectoryRevision != recoveryFixtureBinding { |
| 284 | t.Fatalf("next wake launch tuple: %+v %v", launch, err) |
| 285 | } |
| 286 | }) |
| 287 | } |
| 288 | } |
| 289 | |
amadan