| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "github.com/carlosframework/platform/internal/rootstate" |
| 10 | "io" |
| 11 | "log" |
| 12 | "net" |
| 13 | "net/http" |
| 14 | "os" |
| 15 | "path/filepath" |
| 16 | "time" |
| 17 | |
| 18 | "github.com/carlosframework/platform/internal/hostbox" |
| 19 | "github.com/carlosframework/platform/internal/hostclaim" |
| 20 | execlease "github.com/carlosframework/platform/internal/hostexec" |
| 21 | "github.com/carlosframework/platform/internal/hostguard" |
| 22 | "github.com/carlosframework/platform/internal/instancectl" |
| 23 | "github.com/carlosframework/platform/internal/registry" |
| 24 | "github.com/carlosframework/platform/internal/release" |
| 25 | ) |
| 26 | |
| 27 | // ClaimRootHTTP is instance-control's root claim handler. startupBinding is |
| 28 | // this box's binding in root's startup roster read (the cleanup identity's): |
| 29 | // a pending inventory or rebind intent stamped with another binding is |
| 30 | // obsolete and is settled, never replayed (roster binding N3). Empty skips |
| 31 | // that check. |
| 32 | type ClaimRootHTTP struct { |
| 33 | servingParks *servingParkRuntime |
| 34 | inventoryIdentity func(*os.File) (rootstate.FileIdentity, error) |
| 35 | retireBoundUnit func(context.Context, hostguard.FenceTarget) error |
| 36 | history rootHistoryAccess |
| 37 | executionRuntime *claimExecutionRuntime |
| 38 | Execution *execlease.RootBroker |
| 39 | InventoryActor func(context.Context) (hostbox.Actor, error) |
| 40 | RequireNewWork func(context.Context) error |
| 41 | Journals *hostbox.JournalService |
| 42 | Launches *hostbox.LaunchJournal |
| 43 | Stop ClaimStopHTTP |
| 44 | Reconcile hostbox.ReconcileAdapter |
| 45 | Roots map[string]string |
| 46 | startupBinding string |
| 47 | } |
| 48 | |
| 49 | func (h ClaimRootHTTP) Start(ctx context.Context) { |
| 50 | if h.executionRuntime != nil { |
| 51 | h.executionRuntime.Start(ctx) |
| 52 | } |
| 53 | h.servingParks.Start(ctx) |
| 54 | } |
| 55 | |
| 56 | func (h ClaimRootHTTP) RecoverPending(ctx context.Context) (recoveryErr error) { |
| 57 | // Controller recovery is independent of per-host retained journals and parks. |
| 58 | if h.Execution != nil { |
| 59 | recoveryErr = h.Execution.Recover(ctx) |
| 60 | } |
| 61 | recoveryErr = errors.Join(recoveryErr, h.Stop.RecoverPending(ctx)) |
| 62 | recoveryErr = errors.Join(recoveryErr, h.servingHandover().Recover(ctx)) |
| 63 | // Startup resumes a journaled cold handover only through the fresh |
| 64 | // successor admission (B1). One whose successor was stamped under a |
| 65 | // binding other than the startup binding is settled instead (B3). |
| 66 | cold := h.coldHandover() |
| 67 | cold.Obsolete = func(next hostbox.ColdSuccessor) bool { |
| 68 | if !h.obsoleteIntent(next.DirectoryRevision) { |
| 69 | return false |
| 70 | } |
| 71 | log.Printf("instance-control: %s: obsolete cold handover: stamped %s current %s", next.Host, next.DirectoryRevision, h.startupBinding) |
| 72 | return true |
| 73 | } |
| 74 | recoveryErr = errors.Join(recoveryErr, cold.Recover(ctx)) |
| 75 | journals, err := hostbox.SavedRetirementJournals(h.Reconcile.Retirement.JournalDir) |
| 76 | recoveryErr = errors.Join(recoveryErr, err) |
| 77 | for _, journal := range journals { |
| 78 | recoveryErr = errors.Join(recoveryErr, h.retire(ctx, *journal.Target)) |
| 79 | } |
| 80 | recoveryErr = errors.Join(recoveryErr, h.recoverRootInventory(ctx)) |
| 81 | recoveryErr = errors.Join(recoveryErr, h.servingParks.recoverPending(ctx)) |
| 82 | return errors.Join(recoveryErr, ctx.Err()) |
| 83 | } |
| 84 | func (h ClaimRootHTTP) ServeHTTP(w http.ResponseWriter, r *http.Request) { |
| 85 | if r.URL.Path != "/claim/execution" && r.URL.Path != "/claim/recovery-rebind" && r.URL.Path != "/claim/retire" && r.URL.Path != "/claim/handover" && r.URL.Path != "/claim/journal" { |
| 86 | h.Stop.ServeHTTP(w, r) |
| 87 | return |
| 88 | } |
| 89 | if r.Method != http.MethodPost || h.Stop.Authenticate == nil { |
| 90 | http.Error(w, "refused", http.StatusServiceUnavailable) |
| 91 | return |
| 92 | } |
| 93 | conn := instancectl.RequestConn(r.Context()) |
| 94 | if conn == nil { |
| 95 | http.Error(w, "refused", http.StatusForbidden) |
| 96 | return |
| 97 | } |
| 98 | peer, err := h.Stop.Authenticate(r.Context(), conn) |
| 99 | if err != nil { |
| 100 | http.Error(w, "refused", http.StatusForbidden) |
| 101 | return |
| 102 | } |
| 103 | if peer.UID == 0 || peer.PID <= 0 || peer.ProcessStart == "" || peer.BootID == "" || peer.ControllerID == "" { |
| 104 | http.Error(w, "refused", http.StatusForbidden) |
| 105 | return |
| 106 | } |
| 107 | if r.URL.Path == "/claim/execution" { |
| 108 | h.serveExecution(w, r, peer) |
| 109 | return |
| 110 | } |
| 111 | if r.URL.Path == "/claim/recovery-rebind" { |
| 112 | h.serveRecoveryRebind(w, r) |
| 113 | return |
| 114 | } |
| 115 | if r.URL.Path == "/claim/journal" { |
| 116 | h.serveJournal(w, r, peer) |
| 117 | return |
| 118 | } |
| 119 | if r.URL.Path == "/claim/handover" { |
| 120 | if h.RequireNewWork != nil { |
| 121 | if err := h.RequireNewWork(r.Context()); err != nil { |
| 122 | http.Error(w, "refused", http.StatusForbidden) |
| 123 | return |
| 124 | } |
| 125 | } |
| 126 | h.serveColdHandover(w, r, peer) |
| 127 | return |
| 128 | } |
| 129 | var target hostguard.FenceTarget |
| 130 | decoder := json.NewDecoder(io.LimitReader(r.Body, 32<<10)) |
| 131 | decoder.DisallowUnknownFields() |
| 132 | if decoder.Decode(&target) != nil || decoder.Decode(new(any)) != io.EOF || target.Box != h.Reconcile.Box || target.Enrollment != h.Reconcile.Enrollment { |
| 133 | http.Error(w, "refused", http.StatusBadRequest) |
| 134 | return |
| 135 | } |
| 136 | if target.BootID != peer.BootID || target.ControllerID != peer.ControllerID { |
| 137 | // A restarted controller may carry the exact completed target produced |
| 138 | // for its predecessor. Authenticate still proves the live caller; the |
| 139 | // local root's immutable receipt proves the old tuple. Never rewrite the |
| 140 | // target to the caller's tuple. |
| 141 | receipt, e := h.Stop.Coordinator.Guard.FenceReceiptFor(r.Context(), target) |
| 142 | if e != nil || verifyRetirementReceipt(r.Context(), receipt) != nil { |
| 143 | http.Error(w, "refused", http.StatusBadRequest) |
| 144 | return |
| 145 | } |
| 146 | } |
| 147 | if err = h.retire(r.Context(), target); err != nil { |
| 148 | http.Error(w, "pending", http.StatusConflict) |
| 149 | return |
| 150 | } |
| 151 | w.WriteHeader(http.StatusNoContent) |
| 152 | } |
| 153 | func (h ClaimRootHTTP) retire(ctx context.Context, target hostguard.FenceTarget) error { |
| 154 | a := h.Reconcile |
| 155 | if a.Inventory == nil || a.Registry == nil || !filepath.IsAbs(a.Retirement.JournalDir) { |
| 156 | return hostbox.ErrUnsupported |
| 157 | } |
| 158 | in, err := a.Inventory.ReadExactStopInventory(ctx, target) |
| 159 | if err != nil { |
| 160 | return err |
| 161 | } |
| 162 | if !in.Complete || in.Host != target.Host || in.Incarnation != target.Incarnation || in.InstallationStamp != target.Stamp || in.Box != target.Box || in.Enrollment != target.Enrollment || in.DirectoryRevision != target.DirectoryRevision { |
| 163 | return hostbox.ErrTypedInventory |
| 164 | } |
| 165 | var row registry.Route |
| 166 | for _, e := range in.Entries { |
| 167 | if e.Kind == "route" && e.ManagedRoot == "registry" && e.RelativePath == target.Host && e.Materialization == "materialized" { |
| 168 | if row.Host != "" { |
| 169 | return hostbox.ErrTypedInventory |
| 170 | } |
| 171 | row = registry.Route{Host: target.Host, ClaimOwnership: registry.ClaimOwnership{ClaimIncarnation: target.Incarnation, ClaimStamp: target.Stamp, ClaimOwner: e.Ownership, ClaimRowID: e.RowIdentity}} |
| 172 | } |
| 173 | } |
| 174 | if row.Host == "" || !row.ClaimOwnership.Valid() { |
| 175 | return hostbox.ErrTypedInventory |
| 176 | } |
| 177 | a.RemoteRetire = nil |
| 178 | a.DataRoot = h.Roots["data"] |
| 179 | a.Box = target.Box |
| 180 | a.Reconciler.Guard = h.Stop.Coordinator.Guard |
| 181 | a.RequireTimerInventory = a.RequireTimerInventory || a.Reconciler.Guard.Root != nil |
| 182 | a.Stop = func(context.Context, registry.Route) (hostguard.FenceTarget, error) { return target, nil } |
| 183 | a.Retirement.Filesystem = hostbox.TypedParker{CapturePersistent: h.inventoryIdentity != nil, Roots: h.Roots, ParkedRoots: h.Roots, Incarnation: target.Incarnation, Host: target.Host, Retirement: target.FenceID} |
| 184 | a.Retirement.Authority = hostbox.InstallationCleanup{Guard: a.Reconciler.Guard, Verify: verifyRetirementReceipt} |
| 185 | if h.retireBoundUnit != nil { |
| 186 | a.Retirement.Authority = claimBoundUnitCleanup{inner: a.Retirement.Authority, retire: h.retireBoundUnit} |
| 187 | } |
| 188 | a.Retirement.Ack = retirementAcknowledgement(h.Stop.Coordinator.Evidence) |
| 189 | return a.Retire(ctx, row) |
| 190 | } |
| 191 | func verifyRetirementReceipt(_ context.Context, receipt hostguard.FenceReceipt) error { |
| 192 | var s StopReceipt |
| 193 | if json.Unmarshal(receipt.Bytes, &s) != nil || !bytes.Equal(receipt.Bytes, mustMarshalStopReceipt(s)) { |
| 194 | return ErrClaimStopReceipt |
| 195 | } |
| 196 | t := receipt.Target |
| 197 | if s.Kind != "stopped" || s.Region != t.Region || s.ClaimETag != t.ClaimETag || s.ClaimGen != t.ClaimGen || s.ObligationID != t.ObligationID || s.FenceID != t.FenceID || s.Host != t.Host || s.Incarnation != t.Incarnation || s.InstallationStamp != t.Stamp || s.WriterSession != t.Session || s.OperationID != t.Op || s.Phase != t.Phase || s.Box != t.Box || s.Enrollment != t.Enrollment || s.BootID != t.BootID || s.ControllerID != t.ControllerID || s.Binding != t.DirectoryRevision || !s.EvidenceAt.Equal(receipt.EvidenceAt) { |
| 198 | return ErrClaimStopReceipt |
| 199 | } |
| 200 | if numeral, err := stopReceiptNumeral(t); err != nil || s.DirectoryRevision != numeral { |
| 201 | return ErrClaimStopReceipt |
| 202 | } |
| 203 | return nil |
| 204 | } |
| 205 | func retirementAcknowledgement(store release.Store) func(context.Context, hostbox.RetirementJournal) error { |
| 206 | return func(ctx context.Context, j hostbox.RetirementJournal) error { |
| 207 | if store == nil || !j.Complete || j.Target == nil || len(j.StopReceipt) == 0 { |
| 208 | return ErrClaimStopReceipt |
| 209 | } |
| 210 | var receipt StopReceipt |
| 211 | if json.Unmarshal(j.StopReceipt, &receipt) != nil || receipt.ObligationID == "" || receipt.OperationID == "" { |
| 212 | return ErrClaimStopReceipt |
| 213 | } |
| 214 | if err := verifyRetirementReceipt(ctx, hostguard.FenceReceipt{Target: *j.Target, Bytes: j.StopReceipt, EvidenceAt: receipt.EvidenceAt}); err != nil { |
| 215 | return err |
| 216 | } |
| 217 | body, err := json.Marshal(j) |
| 218 | if err != nil { |
| 219 | return err |
| 220 | } |
| 221 | key := fmt.Sprintf("control/hosts/%s/evidence/%s/%s/%s-retirement.json", receipt.Host, receipt.OperationID, receipt.Box, receipt.ID) |
| 222 | if _, err = store.Put(ctx, key, body, ""); err != nil { |
| 223 | if !errors.Is(err, release.ErrCASConflict) { |
| 224 | return err |
| 225 | } |
| 226 | previous, _, e := store.Get(ctx, key) |
| 227 | if e != nil { |
| 228 | return e |
| 229 | } |
| 230 | if !bytes.Equal(previous, body) { |
| 231 | return ErrClaimStopReceipt |
| 232 | } |
| 233 | } |
| 234 | for attempt := 0; attempt < 8; attempt++ { |
| 235 | observed, err := hostclaim.Read(ctx, store, receipt.Host) |
| 236 | if err != nil { |
| 237 | return err |
| 238 | } |
| 239 | if observed.Absent || observed.Claim == nil { |
| 240 | return ErrClaimStopIdentity |
| 241 | } |
| 242 | found := false |
| 243 | for _, ob := range observed.Claim.Obligations { |
| 244 | if ob.ID == receipt.ObligationID && ob.Op == receipt.OperationID { |
| 245 | if ob.Kind != "park-local" || ob.Box != receipt.Box || ob.Owner != "box:"+receipt.Box { |
| 246 | return ErrClaimStopIdentity |
| 247 | } |
| 248 | found = true |
| 249 | } |
| 250 | } |
| 251 | if !found { |
| 252 | return nil |
| 253 | } |
| 254 | _, err = hostclaim.Update(ctx, store, observed, hostclaim.CompleteObligation(receipt.OperationID, receipt.ObligationID)) |
| 255 | if err == nil { |
| 256 | return nil |
| 257 | } |
| 258 | if !errors.Is(err, release.ErrCASConflict) { |
| 259 | return err |
| 260 | } |
| 261 | } |
| 262 | return release.ErrCASConflict |
| 263 | } |
| 264 | } |
| 265 | func RetireClaimOverSocket(ctx context.Context, socket string, target hostguard.FenceTarget) error { |
| 266 | if socket == "" || !filepath.IsAbs(socket) { |
| 267 | return ErrClaimStopUnavailable |
| 268 | } |
| 269 | body, err := json.Marshal(target) |
| 270 | if err != nil { |
| 271 | return err |
| 272 | } |
| 273 | transport := &http.Transport{DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { |
| 274 | return (&net.Dialer{}).DialContext(ctx, "unix", socket) |
| 275 | }} |
| 276 | defer transport.CloseIdleConnections() |
| 277 | req, err := http.NewRequestWithContext(ctx, http.MethodPost, "http://instance-control/claim/retire", bytes.NewReader(body)) |
| 278 | if err != nil { |
| 279 | return err |
| 280 | } |
| 281 | response, err := (&http.Client{Transport: transport, Timeout: 2 * time.Minute}).Do(req) |
| 282 | if err != nil { |
| 283 | return err |
| 284 | } |
| 285 | defer response.Body.Close() |
| 286 | if response.StatusCode != http.StatusNoContent { |
| 287 | return ErrClaimStopReceipt |
| 288 | } |
| 289 | return nil |
| 290 | } |
| 291 | |
amadan