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 "errors"
6 "fmt"
7 "github.com/carlosframework/platform/internal/registry"
8 "github.com/carlosframework/platform/internal/rootstate"
9 "net/http"
10 "os"
11 "path/filepath"
12 "strings"
13 "time"
14
15 "github.com/carlosframework/platform/internal/activator"
16 "github.com/carlosframework/platform/internal/controlsock"
17 "github.com/carlosframework/platform/internal/genselect"
18 "github.com/carlosframework/platform/internal/hostbox"
19 "github.com/carlosframework/platform/internal/hostclaim"
20 "github.com/carlosframework/platform/internal/hostguard"
21 "github.com/carlosframework/platform/internal/hostops"
22 "github.com/carlosframework/platform/internal/hostwrite"
23 "github.com/carlosframework/platform/internal/lease"
24 "github.com/carlosframework/platform/internal/park"
25 "github.com/carlosframework/platform/internal/provision"
26 "github.com/carlosframework/platform/internal/release"
27)
28
29type ClaimDependencyError struct{ Contracts []string }
30
31func (e *ClaimDependencyError) Error() string {
32 return "host claims: missing provider: " + strings.Join(e.Contracts, ", ")
33}
34func (e *ClaimDependencyError) Unwrap() error { return hostwrite.ErrUnavailable }
35
36type claimEntryKey struct{}
37type claimEntry struct {
38 Resources ClaimRuntimeResources
39 Enabled, FlagRead bool
40 Store release.Store
41 Assembly *ClaimAssembly
42}
43
44func claimStore(ctx context.Context) release.Store {
45 v, _ := ctx.Value(claimEntryKey{}).(claimEntry)
46 return v.Store
47}
48func claimEntryContext(ctx context.Context, getenv func(string) string, open func(context.Context) (release.Store, error)) (context.Context, error) {
49 entry, _ := ctx.Value(claimEntryKey{}).(claimEntry)
50 if !entry.FlagRead {
51 entry.Enabled = hostwrite.Enabled(getenv)
52 entry.FlagRead = true
53 }
54 ctx = context.WithValue(ctx, claimEntryKey{}, entry)
55 if !entry.Enabled || entry.Store != nil {
56 return ctx, nil
57 }
58 if open == nil {
59 return ctx, hostwrite.ErrUnavailable
60 }
61 startup, cancel := context.WithTimeout(ctx, 2*time.Minute)
62 defer cancel()
63 store, err := open(startup)
64 if err != nil {
65 return ctx, err
66 }
67 if store == nil {
68 return ctx, hostwrite.ErrUnavailable
69 }
70 entry.Store = store
71 return context.WithValue(ctx, claimEntryKey{}, entry), nil
72}
73func claimEntryEnv(ctx context.Context) func(string) string {
74 entry, _ := ctx.Value(claimEntryKey{}).(claimEntry)
75 return func(key string) string {
76 if key == "CARLOS_HOST_CLAIM" {
77 if entry.FlagRead && entry.Enabled {
78 return "1"
79 }
80 return ""
81 }
82 return os.Getenv(key)
83 }
84}
85
86// WithClaimAssembly is used by the deployment composition after it constructs
87// the Plan 8/8b and moves adapters. It does not activate any host itself.
88func WithClaimAssembly(ctx context.Context, a *ClaimAssembly) context.Context {
89 v, _ := ctx.Value(claimEntryKey{}).(claimEntry)
90 v.Assembly = a
91 return context.WithValue(ctx, claimEntryKey{}, v)
92}
93
94// ClaimAssembly owns the concrete wiring. Dependencies retain their contract
95// names so a missing implementation cannot silently become permissive policy.
96type ClaimAssembly struct {
97 LossAdmission *hostbox.LossAdmission
98 moveRead *claimMoveReadAccess
99 wakeUnavailable error
100 Journals *hostbox.JournalService
101 cleanupEnrollment *claimCleanupEnrollment
102 RetirementAuthority *hostops.ProductionAuthority
103 Mode string
104 Producers *ClaimProducerSet
105 Launches *hostbox.LaunchJournal
106 SharedPaths map[string][]string
107 ManagedRoots map[string]string
108 Reconcile *hostbox.ReconcileAdapter
109 RootJournal hostbox.StopKernelJournalReader
110 Controllers StopControllerReader
111 Claims release.Store
112 Guard hostguard.Guard
113 Identity hostbox.IdentityProvider
114 Local hostbox.LocalAuthority
115 Policy hostbox.Policy
116 Data lease.VersionedStore
117 Wake *ClaimWakeAssembly
118 Stop *ClaimStopCoordinator
119 Evidence hostbox.RegionalEvidence
120 LeaseSet hostbox.LeaseSet
121 View func(context.Context, hostclaim.Observed) (park.View, error)
122 Authorize func(context.Context, hostclaim.Observed, hostclaim.Obligation) error
123 AdmissionEvidence func(context.Context, hostclaim.Observed, hostclaim.Admission, *genselect.Plan) error
124 PublicationEvidence func(context.Context, hostclaim.Observed) error
125 NewID func() string
126 Provision func(*provision.ReconcileConfig)
127 Close func(context.Context) error
128 // Test seams for the local stop identity and the root transport. Nil uses
129 // the verified identity and the instance-control socket.
130 stopLocal func(context.Context) (claimStopLocal, error)
131 collectStop func(context.Context, StopRequest, StopDrainer) (StopResult, error)
132 // acquireReplicaLease is the relocation build's replica lease, a test
133 // seam for its loss. Nil uses lease.Acquire.
134 acquireReplicaLease func(ctx context.Context, bucket, prefix, region string) (replicaLease, error)
135}
136
137func (a ClaimAssembly) Build(ctx context.Context) (ClaimProviders, error) {
138 if a.Reconcile != nil && a.Reconcile.Registry != nil {
139 a.Policy.Held = hostbox.RegistryHoldPolicy(a.Reconcile.Registry)
140 }
141 if ctx == nil || a.RetirementAuthority == nil || a.Claims == nil || a.Guard.DataDir == "" || a.Data == nil || a.Local == nil || a.Policy.Held == nil || a.Wake == nil || a.Stop == nil || a.Evidence == nil || a.LeaseSet == nil || a.View == nil || a.Authorize == nil || a.NewID == nil || a.AdmissionEvidence == nil || a.PublicationEvidence == nil || a.Reconcile == nil || a.Close == nil {
142 return ClaimProviders{}, hostwrite.ErrUnavailable
143 }
144 if a.Controllers == nil || a.Launches == nil || a.Reconcile.Registry == nil || a.Reconcile.Inventory == nil || !filepath.IsAbs(a.Reconcile.Retirement.JournalDir) {
145 return ClaimProviders{}, hostwrite.ErrUnavailable
146 }
147 for _, name := range []string{"data", "opt", "units"} {
148 if !filepath.IsAbs(a.ManagedRoots[name]) {
149 return ClaimProviders{}, hostwrite.ErrUnavailable
150 }
151 }
152 if filepath.Clean(a.ManagedRoots["data"]) != filepath.Clean(a.Guard.DataDir) {
153 return ClaimProviders{}, hostbox.ErrChanged
154 }
155 // Build owns these values. Do not mutate a caller's shared adapter or accept
156 // zero-value write configuration that would reopen the legacy path.
157 reconcile := *a.Reconcile
158 a.Reconcile = &reconcile
159 actor, err := a.Identity.Verify(ctx)
160 if err != nil {
161 return ClaimProviders{}, err
162 }
163 a.Guard.Claims = hostguard.StoreReader{Store: a.Claims}
164 // Continue is the moves C1 session table, nil unless both flags are on (D1).
165 source := hostbox.WorkSource{FleetHomeRegion: actor.FleetHomeRegion, HomeRegion: actor.HomeRegion, Guard: a.Guard, Box: actor.Box, Region: park.PlacementRegion(actor.Region), Now: time.Now, Policy: a.Policy.Check, Local: a.Identity.Local(a.Local), Continue: a.Wake.Continue}
166 source.Timer = func(ctx context.Context, work hostwrite.Work, timer registry.Timer, expected string, remove bool) (*registry.Timer, error) {
167 return a.Reconcile.MutateTimer(ctx, work, timer, expected, remove)
168 }
169 a.Reconcile.Reconciler.Claims = a.Claims
170 a.Reconcile.Reconciler.Guard = a.Guard
171 a.Reconcile.Reconciler.Writes = hostwrite.Config{Enabled: true, Selector: source}
172 a.Reconcile.Reconciler.Box = actor.Box
173 a.Reconcile.Reconciler.Region = park.PlacementRegion(actor.Region)
174 a.Reconcile.Reconciler.HomeRegion = actor.HomeRegion
175 a.Reconcile.Reconciler.Now = time.Now
176 a.Reconcile.Reconciler.Local = park.Local{Guard: a.Guard, Claims: a.Claims, DataDir: a.Guard.DataDir, Box: actor.Box}
177 a.Reconcile.Box = actor.Box
178 a.Reconcile.Enrollment = actor.Enrollment
179 a.Reconcile.DirectoryRevision = actor.Binding
180 census := &hostbox.EmptyInstallationCensus{Inventory: a.Reconcile.Inventory, Registry: a.Reconcile.Registry, Launches: a.Launches, Box: actor.Box, Enrollment: actor.Enrollment, DirectoryRevision: actor.Binding, Roots: a.ManagedRoots, SharedPaths: a.SharedPaths}
181 a.Reconcile.Census = census
182 a.Reconcile.Reconciler.CheckLocal = census.CheckLocal
183 a.Reconcile.Stop = a.collectRetirementStop
184 a.Reconcile.RemoteRetire = func(ctx context.Context, target hostguard.FenceTarget) error {
185 return RetireClaimOverSocket(ctx, controlsock.Path, target)
186 }
187 check := claimBoundIdentity(a.Identity.Verify, actor)
188 data := hostbox.RequestStore{Store: a.Data, Check: check}
189 writer := hostbox.Writer{HomeRegion: actor.HomeRegion, Loss: a.LossAdmission, Claims: a.Claims, Data: data, Bucket: actor.Bucket, Region: park.PlacementRegion(actor.Region), Box: actor.Box, AdmissionEvidence: a.AdmissionEvidence, PublicationEvidence: a.PublicationEvidence, NewID: a.NewID}
190 a.bindClaimWakeStop()
191 if local, ok := a.Local.(*productionClaimLocal); ok {
192 if err := local.bind(ctx, &a); err != nil {
193 return ClaimProviders{}, err
194 }
195 a.Reconcile.Reconciler.Prepare = local.Prepare
196 source.Prepare = local.Prepare
197 a.Reconcile.Reconciler.Writes = hostwrite.Config{Enabled: true, Selector: source}
198 }
199 a.Provision = a.Reconcile.Configure
200 if a.moveRead != nil {
201 if err := a.moveRead.bind(ctx, source); err != nil {
202 return ClaimProviders{}, err
203 }
204 }
205 a.Wake.LeaseRegion = actor.Cloud.Region
206 wake := refusingRootWake(actor.Box, a.Claims)
207 if (a.Mode == "agent" || a.Mode == "") && a.wakeUnavailable == nil {
208 wake, err = a.Wake.Build(ctx, source, writer)
209 }
210 if a.wakeUnavailable != nil && (a.Mode == "agent" || a.Mode == "") {
211 wake = refusingClaimMoveWake(actor.Box, a.Claims, a.wakeUnavailable)
212 }
213 if err != nil {
214 return ClaimProviders{}, err
215 }
216 if wake == nil || wake.Read == nil || wake.Prerequisites == nil || wake.Admit == nil || wake.Prepare == nil || wake.StopForeign == nil || wake.FailedAdmission == nil {
217 return ClaimProviders{}, hostwrite.ErrUnavailable
218 }
219 if a.Mode == "agent" && a.wakeUnavailable == nil {
220 a.Wake.seedFactory = func(call context.Context, op *filesSeedOperation, exclusion *activator.SeedExclusion) (*activator.ClaimRuntime, *ClaimWakeAssembly, error) {
221 return newFilesSeedRuntime(call, a, source, writer, op, exclusion)
222 }
223 }
224 // Where claimed wakes lease (writer.Bucket, a.Wake.LeaseRegion above):
225 // the agent's build-state store reads and writes beside that lease.
226 wake.Bucket, wake.LeaseRegion = actor.Bucket, actor.Cloud.Region
227 enrollment := claimRegionalEnrollment{identity: a.Identity, actor: actor}
228 authorize := func(ctx context.Context, o hostclaim.Observed, ob hostclaim.Obligation) error {
229 if err := enrollment.VerifyRegional(ctx, actor.Region, actor.Bucket, hostbox.RequestLimit); err != nil {
230 return err
231 }
232 return a.Authorize(ctx, o, ob)
233 }
234 leases := hostbox.LeaseSet(func(ctx context.Context, bucket, scope string) ([]string, error) {
235 if err := check(ctx); err != nil {
236 return nil, err
237 }
238 if bucket != actor.Bucket {
239 return nil, hostbox.ErrChanged
240 }
241 return a.LeaseSet(ctx, bucket, scope)
242 })
243 view := func(ctx context.Context, o hostclaim.Observed) (park.View, error) {
244 if err := check(ctx); err != nil {
245 return park.View{}, err
246 }
247 return a.View(ctx, o)
248 }
249 engine := park.Engine{HomeRegion: actor.HomeRegion, Store: data, Claims: a.Claims, Bucket: actor.Bucket, Region: park.PlacementRegion(actor.Region), Leases: hostbox.AggregateLeases{Leases: hostbox.RenewingLeases{Region: actor.Cloud.Region}, Resolve: leases}, Now: time.Now, NewID: a.NewID, View: view, Authorize: authorize}
250 engine.VerifyMovesSweep = func(ctx context.Context, o hostclaim.Observed, versions []lease.ObjectVersion) (park.MoveSweepProof, error) {
251 if err := check(ctx); err != nil {
252 return park.MoveSweepProof{}, err
253 }
254 verifier, ok := a.Evidence.(park.MoveSweepVerifier)
255 if !ok {
256 return park.MoveSweepProof{}, &ClaimDependencyError{Contracts: []string{"moves Plan 5 newest retired move fence or release evidence"}}
257 }
258 return verifier.VerifyMovesSweep(ctx, o, versions)
259 }
260 regional := hostbox.RegionalWorker{Regional: hostbox.Regional{Engine: engine, Owner: func(ctx context.Context, _ hostclaim.Observed) error {
261 return enrollment.VerifyRegional(ctx, actor.Region, actor.Bucket, hostbox.RequestLimit)
262 }}, Enrollment: enrollment, Evidence: a.Evidence}
263 regional, err = hostops.NewClaimRegionalWorker(regional.Regional, enrollment, *a.RetirementAuthority, a.Evidence)
264 if err != nil {
265 return ClaimProviders{}, err
266 }
267 if local, ok := a.Local.(*productionClaimLocal); ok {
268 local.start(ctx)
269 }
270 if err := a.startWakeResetWorker(ctx); err != nil {
271 return ClaimProviders{}, err
272 }
273 return ClaimProviders{guard: a.Guard, registerLanding: a.Reconcile.RegisterLanding, moveTargetFence: claimBoxFenceVerifier(check, a.Claims, a.RetirementAuthority), moveParks: &claimMoveParks{local: claimMoveLocalAuthorizer(check, a.Authorize), engine: regional.Regional.Engine, leases: leases, check: check}, Delivery: ClaimTimerDelivery(a.Producers, a.NewID), Selector: source, Wake: wake, Regional: regional, Stop: a.Stop, Provision: a.Provision, Close: a.Close}, nil
274}
275
276type claimRegionalEnrollment struct {
277 identity hostbox.IdentityProvider
278 actor hostbox.Actor
279}
280
281func (e claimRegionalEnrollment) VerifyRegional(ctx context.Context, region, bucket string, limit time.Duration) error {
282 if limit != hostbox.RequestLimit {
283 return hostbox.ErrUnsupported
284 }
285 a, err := e.identity.Verify(ctx)
286 if err != nil {
287 return err
288 }
289 if !hostbox.SameBinding(a, e.actor) || !park.SameRegion(a.HomeRegion, park.PlacementRegion(a.Region), region) || a.Bucket != bucket {
290 return hostbox.ErrChanged
291 }
292 return nil
293}
294
295func productionClaimFactory(ctx context.Context) (ClaimProviders, error) {
296 entry, _ := ctx.Value(claimEntryKey{}).(claimEntry)
297 if entry.Store == nil {
298 return ClaimProviders{}, fmt.Errorf("deployment client must precede claim assembly: %w", hostwrite.ErrUnavailable)
299 }
300 if entry.Assembly == nil {
301 var err error
302 ctx, err = ensureProductionClaimAssembly(ctx)
303 if err != nil {
304 return ClaimProviders{}, err
305 }
306 entry, _ = ctx.Value(claimEntryKey{}).(claimEntry)
307 }
308 if entry.Assembly.Claims != entry.Store {
309 return ClaimProviders{}, errors.New("host claims: deployment store mismatch")
310 }
311 return entry.Assembly.Build(ctx)
312}
313
314func configuredClaimProvision(cfg provision.ReconcileConfig, configure []func(*provision.ReconcileConfig)) provision.ReconcileConfig {
315 for _, f := range configure {
316 if f != nil {
317 f(&cfg)
318 }
319 }
320 return cfg
321}
322
323func NewProductionRootClaimStop(ctx context.Context) (http.Handler, error) {
324 entry, _ := ctx.Value(claimEntryKey{}).(claimEntry)
325 if entry.Store == nil {
326 var err error
327 ctx, err = claimEntryContext(ctx, os.Getenv, openDeployStore)
328 if err != nil {
329 return nil, err
330 }
331 entry, _ = ctx.Value(claimEntryKey{}).(claimEntry)
332 }
333 if entry.Assembly == nil || entry.Assembly.Stop == nil || entry.Assembly.RootJournal == nil || entry.Assembly.Controllers == nil {
334 return nil, &ClaimDependencyError{Contracts: []string{"Plan 8/8b verified controller, roster, floor and quarantine root bindings"}}
335 }
336 assembly := entry.Assembly
337 if assembly.Claims != entry.Store || assembly.Guard.DataDir == "" || assembly.Reconcile == nil || assembly.Reconcile.Inventory == nil {
338 return nil, hostwrite.ErrUnavailable
339 }
340 for _, name := range []string{"data", "opt", "units"} {
341 if !filepath.IsAbs(assembly.ManagedRoots[name]) {
342 return nil, hostwrite.ErrUnavailable
343 }
344 }
345 actor, err := assembly.actorForRoot(ctx)
346 if err != nil {
347 return nil, err
348 }
349 if actor.Box == "" || actor.Enrollment == "" {
350 return nil, hostbox.ErrChanged
351 }
352 handler, err := NewProductionClaimStopHandler(assembly.rootStopCoordinator(entry.Store), assembly.RootJournal, assembly.Controllers, assembly.ManagedRoots)
353 if err != nil {
354 return nil, err
355 }
356 stop, ok := handler.(ClaimStopHTTP)
357 if !ok {
358 return nil, ErrClaimStopUnavailable
359 }
360 root := assembly.claimRootHTTP(actor, stop)
361 state, err := openClaimExecutionAuthority()
362 if err != nil {
363 return nil, err
364 }
365 if assembly.Wake == nil || assembly.Wake.Inputs == nil {
366 state.Close()
367 return nil, hostbox.ErrUnsupported
368 }
369 root, err = withServingParkRoot(root, assembly.Wake.Inputs)
370 if err != nil {
371 state.Close()
372 return nil, err
373 }
374 runtime, err := newClaimExecutionRuntime(root, state)
375 if err != nil {
376 root.servingParks.Close(ctx)
377 state.Close()
378 return nil, err
379 }
380 runtime.parks = root.servingParks
381 runtime.root = &root
382 root.Execution, root.executionRuntime = runtime.broker, runtime
383 return root, nil
384}
385
386// rootStopCoordinator is root's stop coordinator over the assembly and the
387// deployment store, before NewProductionClaimStopHandler binds its peer
388// authentication, kernel and identities. Its numeral bound is the cleanup
389// enrollment's fresh roster read.
390func (assembly *ClaimAssembly) rootStopCoordinator(store release.Store) ClaimStopCoordinator {
391 coordinator := *assembly.Stop
392 coordinator.Guard = assembly.Guard
393 coordinator.Guard.Claims = hostguard.StoreReader{Store: store}
394 coordinator.Claims = hostguard.StoreReader{Store: store}
395 coordinator.Inventory = assembly.Reconcile.Inventory
396 coordinator.Evidence = store
397 // The numeral bound is root's own fresh verified roster read (fix
398 // round 1, finding 3): the cleanup enrollment's, which still reads a
399 // quarantined box's roster. Production instance-control always has one;
400 // without it the bound stays nil and every stop refuses.
401 if assembly.cleanupEnrollment != nil {
402 coordinator.VerifiedRosterRevision = assembly.cleanupEnrollment.rosterRevision
403 }
404 return coordinator
405}
406
407// claimRootHTTP is instance-control's root handler over the assembly. Two
408// identities, kept apart (roster binding R2): actor is the historical
409// cleanup identity of root's startup read, which pins the census and
410// stamps what the cleanup service itself constructs, and keeps stops,
411// retirements and settlements of historical installations working while
412// the box is quarantined. InventoryActor is the fresh verified identity on
413// every call: every comparison that admits new work (an inventory write,
414// a serving handover, a recovery rebind, a reboot preparation) is against
415// the roster as it is now, so a quarantine refuses at once and a
416// readmission admits only the new binding, without a restart.
417func (assembly *ClaimAssembly) claimRootHTTP(actor hostbox.Actor, stop ClaimStopHTTP) ClaimRootHTTP {
418 reconcile := *assembly.Reconcile
419 reconcile.Box, reconcile.Enrollment = actor.Box, actor.Enrollment
420 reconcile.Census = &hostbox.EmptyInstallationCensus{Inventory: assembly.Reconcile.Inventory, Registry: assembly.Reconcile.Registry, Launches: assembly.Launches, Box: actor.Box, Enrollment: actor.Enrollment, DirectoryRevision: actor.Binding, Roots: assembly.ManagedRoots, SharedPaths: assembly.SharedPaths}
421 return ClaimRootHTTP{inventoryIdentity: rootstate.ObjectIdentity, InventoryActor: assembly.Identity.Verify, startupBinding: actor.Binding, Journals: assembly.Journals, RequireNewWork: func(ctx context.Context) error { _, err := assembly.Identity.Verify(ctx); return err }, Launches: assembly.Launches, Stop: stop, Reconcile: reconcile, Roots: assembly.ManagedRoots}
422}
423
424func claimMoveLocalAuthorizer(check func(context.Context) error, owner func(context.Context, hostclaim.Observed, hostclaim.Obligation) error) ClaimLocalParkAuthorizer {
425 return func(ctx context.Context, c *hostclaim.Claim, ob hostclaim.Obligation) error {
426 if check == nil || owner == nil || c == nil || ob.Kind != "park-local" {
427 return hostwrite.ErrUnavailable
428 }
429 if err := check(ctx); err != nil {
430 return err
431 }
432 // Preserve the unwrapped owner used by the assembly's own park.Local.
433 return owner(ctx, hostclaim.Observed{Host: c.Host, Claim: c}, ob)
434 }
435}
436