| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "strings" |
| 7 | "sync" |
| 8 | "time" |
| 9 | |
| 10 | "github.com/carlosframework/platform/internal/hostactors" |
| 11 | "github.com/carlosframework/platform/internal/hostbox" |
| 12 | "github.com/carlosframework/platform/internal/hostevidence" |
| 13 | "github.com/carlosframework/platform/internal/release" |
| 14 | ) |
| 15 | |
| 16 | // One closure is one immutable migration batch. Its verified roster and each |
| 17 | // evidence namespace are loaded once; both successful reads and errors are |
| 18 | // sticky until a new root startup creates a fresh closure. No cache is global |
| 19 | // or survives migration. Only signatures, never copied counters, admit rows. |
| 20 | func legacyJournalProof(store release.Store, directory hostactors.Directory) func(context.Context, hostbox.LaunchInventory, hostbox.StopInventory, hostbox.StopKernelJournal) error { |
| 21 | store = legacyProofReadStore{Store: store} |
| 22 | directory.Store = legacyProofReadStore{Store: directory.Store} |
| 23 | type proofSet struct { |
| 24 | matches map[string]struct{} |
| 25 | err error |
| 26 | } |
| 27 | var mu sync.Mutex |
| 28 | var roster hostactors.Observed |
| 29 | var rosterErr error |
| 30 | loadedRoster := false |
| 31 | namespaces := map[string]proofSet{} |
| 32 | return func(ctx context.Context, in hostbox.LaunchInventory, inventory hostbox.StopInventory, journal hostbox.StopKernelJournal) error { |
| 33 | if err := ctx.Err(); err != nil { |
| 34 | return err |
| 35 | } |
| 36 | mu.Lock() |
| 37 | defer mu.Unlock() |
| 38 | if err := ctx.Err(); err != nil { |
| 39 | return err |
| 40 | } |
| 41 | if !loadedRoster { |
| 42 | roster, rosterErr = directory.Read(ctx) |
| 43 | loadedRoster = true |
| 44 | } |
| 45 | if rosterErr != nil { |
| 46 | return rosterErr |
| 47 | } |
| 48 | actor, ok := roster.Roster.Boxes[in.Target.Box] |
| 49 | if !ok || actor.Enrollment != in.Target.Enrollment || roster.Roster.RequestLimitNS != int64(hostbox.RequestLimit) { |
| 50 | return hostbox.ErrLegacyJournalProof |
| 51 | } |
| 52 | // A serving writer has no operation; its stop is published under |
| 53 | // serving-evidence, exactly where stopEvidenceKey puts it. |
| 54 | prefix := strings.TrimSuffix(stopEvidenceKey(in.Target.Host, in.Target.Op, in.Target.Box, ""), ".json") |
| 55 | cached, ok := namespaces[prefix] |
| 56 | if !ok { |
| 57 | cached = proofSet{matches: map[string]struct{}{}} |
| 58 | keys, err := store.List(ctx, prefix) |
| 59 | cached.err = err |
| 60 | if err == nil { |
| 61 | for _, key := range keys { |
| 62 | if err = ctx.Err(); err != nil { |
| 63 | cached.err = err |
| 64 | break |
| 65 | } |
| 66 | if !strings.HasPrefix(key, prefix) || !strings.HasSuffix(key, ".json") { |
| 67 | continue |
| 68 | } |
| 69 | raw, _, err := store.Get(ctx, key) |
| 70 | if err != nil { |
| 71 | cached.err = err |
| 72 | break |
| 73 | } |
| 74 | signature, _, err := store.Get(ctx, key+".sig") |
| 75 | if err != nil { |
| 76 | if ctx.Err() != nil { |
| 77 | cached.err = ctx.Err() |
| 78 | break |
| 79 | } |
| 80 | continue |
| 81 | } |
| 82 | if hostevidence.Verify(actor.CensusPublicKey, key, raw, signature) != nil { |
| 83 | continue |
| 84 | } |
| 85 | var proof StopReceipt |
| 86 | if json.Unmarshal(raw, &proof) != nil || proof.V != 1 || proof.Kind != "stopped" || proof.ID == "" || key != prefix+proof.ID+".json" || proof.RequestLimitNS != int64(hostbox.RequestLimit) || proof.EvidenceAt.IsZero() || proof.FenceID == "" || proof.ClaimGen == 0 || proof.ObservedBootID == "" || len(proof.CgroupIdentities) == 0 { |
| 87 | continue |
| 88 | } |
| 89 | // Cache only the exact match digest, not every decoded receipt and its |
| 90 | // cgroup vectors. Each later row is one map lookup, not a receipt scan. |
| 91 | cached.matches[legacyJournalProofMatch(proof)] = struct{}{} |
| 92 | } |
| 93 | } |
| 94 | if cached.err == nil { |
| 95 | cached.err = ctx.Err() |
| 96 | } |
| 97 | namespaces[prefix] = cached |
| 98 | } |
| 99 | if cached.err != nil { |
| 100 | return cached.err |
| 101 | } |
| 102 | if err := ctx.Err(); err != nil { |
| 103 | return err |
| 104 | } |
| 105 | target := in.Target |
| 106 | wanted := StopReceipt{Host: target.Host, Incarnation: target.Incarnation, InstallationStamp: target.Stamp, WriterSession: target.Session, Box: target.Box, Enrollment: target.Enrollment, BootID: target.BootID, ControllerID: target.ControllerID, OperationID: target.Op, Region: target.Region, Binding: inventory.DirectoryRevision, InventorySHA256: digestStop(inventory), LocalJournalSHA256: digestStop(journal)} |
| 107 | if _, ok := cached.matches[legacyJournalProofMatch(wanted)]; ok { |
| 108 | return nil |
| 109 | } |
| 110 | return hostbox.ErrLegacyJournalProof |
| 111 | } |
| 112 | } |
| 113 | func legacyJournalProofMatch(p StopReceipt) string { |
| 114 | return digestStop([]any{p.Host, p.Incarnation, p.InstallationStamp, p.WriterSession, p.Box, p.Enrollment, p.BootID, p.ControllerID, p.OperationID, p.Region, p.Binding, p.InventorySHA256, p.LocalJournalSHA256}) |
| 115 | } |
| 116 | |
| 117 | // A shared batch can cover many rows, but no individual network request may |
| 118 | // consume more than one read allowance. Neither wrapper outlives its closure. |
| 119 | type legacyProofReadStore struct{ release.Store } |
| 120 | |
| 121 | func (s legacyProofReadStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 122 | ctx, cancel := context.WithTimeout(ctx, 30*time.Second) |
| 123 | defer cancel() |
| 124 | body, etag, err := s.Store.Get(ctx, key) |
| 125 | if err == nil { |
| 126 | err = ctx.Err() |
| 127 | } |
| 128 | return body, etag, err |
| 129 | } |
| 130 | func (s legacyProofReadStore) List(ctx context.Context, prefix string) ([]string, error) { |
| 131 | ctx, cancel := context.WithTimeout(ctx, 30*time.Second) |
| 132 | defer cancel() |
| 133 | keys, err := s.Store.List(ctx, prefix) |
| 134 | if err == nil { |
| 135 | err = ctx.Err() |
| 136 | } |
| 137 | return keys, err |
| 138 | } |
| 139 | |
amadan