carlos / platform Public

Download this file

1	package host
2
3	import (
4	"bytes"
5	"context"
6	"crypto/sha256"
7	"encoding/hex"
8	"encoding/json"
9	"fmt"
10	"io"
11	"net"
12	"net/http"
13	"reflect"
14	"slices"
15	"strconv"
16	"time"
17
18	"github.com/carlosframework/platform/internal/hostbox"
19	"github.com/carlosframework/platform/internal/hostclaim"
20	"github.com/carlosframework/platform/internal/hostguard"
21	)
22
23	type recoveryRebindRequest struct {
24	V int
25	Prior hostguard.Captured
26	Revision uint64
27	RecordKey, RecordETag string
28	Next hostbox.ColdSuccessor
29	}
30
31	type recoveryRebindIntent struct{ Recovery *recoveryRebindRequest }
32
33	func recoveryInventory(in hostbox.StopInventory, next hostbox.ColdSuccessor) hostbox.StopInventory {
34	in.Entries = append([]hostbox.StopInventoryEntry(nil), in.Entries...)
35	in.InstallationStamp, in.DirectoryRevision = next.Stamp, next.DirectoryRevision
36	sum := sha256.Sum256([]byte(fmt.Sprintf("%s\x00%s\x00%d\x00%s", next.Host, next.Incarnation, next.Stamp, next.Box)))
37	owner := hex.EncodeToString(sum[:])
38	for i := range in.Entries {
39	e := &in.Entries[i]
40	e.Ownership = owner
41	if e.Kind == "route" || e.Kind == "timer" {
42	name := e.RelativePath
43	if e.Kind == "timer" {
44	name = name[len(next.Host)+1:]
45	}
46	sum := sha256.Sum256([]byte(owner + "\x00" + e.Kind + "\x00" + name))
47	e.RowIdentity = hex.EncodeToString(sum[:])
48	}
49	}
50	return in
51	}
52
53	func (h ClaimRootHTTP) rebindRecoveryInventory(ctx context.Context, q recoveryRebindRequest) error {
54	guard := h.Stop.Coordinator.Guard
55	if guard.Root == nil || h.Reconcile.Inventory == nil || h.InventoryActor == nil || q.V != 1 || q.Revision == 0 || q.RecordKey == "" || q.RecordETag == "" {
56	return hostbox.ErrUnsupported
57	}
58	if h.RequireNewWork != nil {
59	if err := h.RequireNewWork(ctx); err != nil {
60	return err
61	}
62	}
63	actor, err := h.InventoryActor(ctx)
64	if err != nil {
65	return err
66	}
67	n := q.Next
68	if n.Box != actor.Box || n.Enrollment != actor.Enrollment || actor.Binding == "" || n.DirectoryRevision != actor.Binding || n.Incarnation != q.Prior.Incarnation || n.Stamp <= q.Prior.Stamp || n.Session != n.Stamp || n.Generation == "" || n.ETag == "" {
69	return hostbox.ErrChanged
70	}
71	source, revision, err := h.Reconcile.Inventory.Snapshot(ctx, n.Host, q.Prior)
72	if err != nil {
73	return err
74	}
75	// The predecessor is reused only under today's binding (roster binding
76	// R3): an installation stamped under another binding (before a
77	// quarantine, or a numeral) is refused before any reuse, so the rebound
78	// inventory copies a verified value.
79	if revision != q.Revision || !source.Complete || source.Host != n.Host || source.Incarnation != q.Prior.Incarnation || source.InstallationStamp != q.Prior.Stamp || source.Box != actor.Box || source.Enrollment != actor.Enrollment || source.DirectoryRevision != actor.Binding || len(source.Entries) == 0 {
80	return hostbox.ErrTypedInventory
81	}
82	observed, err := hostclaim.Read(ctx, h.Stop.Coordinator.Evidence, n.Host)
83	if err != nil {
84	return err
85	}
86	if observed.Claim == nil || observed.Claim.Holder == nil {
87	return hostbox.ErrChanged
88	}
89	holder := observed.Claim.Holder
90	next := hostguard.RootInstallation{Host: n.Host, Incarnation: n.Incarnation, Stamp: n.Stamp, Account: holder.Account, App: holder.App, Record: q.RecordKey, Box: n.Box, Enrollment: n.Enrollment, Migration: 1, InventoryRevision: 1}
91	var rebound hostbox.StopInventory
92	check := func(c *hostclaim.Claim) error {
93	if slices.Contains(c.Lost, n.Box) {
94	return hostbox.ErrChanged
95	}
96	fresh, err := hostclaim.Read(ctx, h.Stop.Coordinator.Evidence, n.Host)
97	if err != nil {
98	return err
99	}
100	if fresh.Absent || fresh.ETag != n.ETag || !reflect.DeepEqual(c, fresh.Claim) || c.Holder == nil {
101	return hostbox.ErrChanged
102	}
103	holder := c.Holder
104	stamp, err := strconv.ParseUint(c.Op.Journal[hostclaim.JournalSourceStamp], 10, 64)
105	if err != nil || stamp != q.Prior.Stamp || !claimRecoveryKind(c.Op.Kind) || c.Op.Phase != hostclaim.PhaseProbing || c.Op.ID != n.Op || c.Op.Phase != n.Phase || c.Op.Exec != n.Box || c.Op.Intent["source"] != n.Box || holder.Key != q.RecordKey || holder.Incarnation != n.Incarnation || holder.Stamp != n.Stamp || holder.Session != n.Session || holder.Generation != n.Generation || holder.WriterBox != n.Box {
106	return hostbox.ErrChanged
107	}
108	if len(holder.Admissions) == 0 {
109	return hostbox.ErrChanged
110	}
111	a := holder.Admissions[len(holder.Admissions)-1]
112	if a.Mode != "cold" || a.Session != n.Session || a.Generation != n.Generation || a.Box != n.Box {
113	return hostbox.ErrChanged
114	}
115	canonical, etag, err := rootClaimRecord(ctx, h.Stop.Coordinator.Evidence, c, h.Roots["data"])
116	if err != nil {
117	return err
118	}
119	if etag != q.RecordETag || canonical.Host != n.Host || canonical.Account != next.Account || canonical.App != next.App {
120	return hostbox.ErrChanged
121	}
122	canonical.ClaimIncarnation, canonical.ClaimStamp = q.Prior.Incarnation, q.Prior.Stamp
123	routes := 0
124	for _, e := range source.Entries {
125	if e.Kind == "route" {
126	routes++
127	canonical.ClaimOwner, canonical.ClaimRowID = e.Ownership, e.RowIdentity
128	}
129	}
130	if routes != 1 || hostbox.ValidateAgentInventory(source, &canonical, h.Roots["data"]) != nil {
131	return hostbox.ErrTypedInventory
132	}
133	current, rev, err := h.Reconcile.Inventory.Snapshot(ctx, n.Host, q.Prior)
134	if err != nil {
135	return err
136	}
137	if rev != q.Revision || !reflect.DeepEqual(source, current) {
138	return hostbox.ErrChanged
139	}
140	rebound = recoveryInventory(source, n)
141	return nil
142	}
143	raw, err := json.Marshal(recoveryRebindIntent{Recovery: &q})
144	if err != nil {
145	return err
146	}
147	return guard.RebindRootInventory(ctx, q.Prior, q.Revision, next, raw, check, func() error {
148	current, rev, err := h.Reconcile.Inventory.Snapshot(ctx, n.Host, hostguard.Captured{Incarnation: n.Incarnation, Stamp: n.Stamp})
149	if err != nil {
150	return err
151	}
152	if rev != 0 {
153	if rev != 1 || !reflect.DeepEqual(current, rebound) {
154	return hostbox.ErrChanged
155	}
156	return nil
157	}
158	return h.Reconcile.Inventory.WriteStopInventory(ctx, rebound)
159	})
160	}
161
162	func (h ClaimRootHTTP) serveRecoveryRebind(w http.ResponseWriter, r *http.Request) {
163	var q recoveryRebindRequest
164	dec := json.NewDecoder(io.LimitReader(r.Body, 32<<10))
165	dec.DisallowUnknownFields()
166	if dec.Decode(&q) != nil || dec.Decode(new(any)) != io.EOF {
167	w.WriteHeader(http.StatusBadRequest)
168	return
169	}
170	if err := h.rebindRecoveryInventory(r.Context(), q); err != nil {
171	w.WriteHeader(http.StatusConflict)
172	return
173	}
174	w.WriteHeader(http.StatusNoContent)
175	}
176	func recoveryRebindClient(socket string) func(context.Context, recoveryRebindRequest) error {
177	return func(ctx context.Context, q recoveryRebindRequest) error {
178	body, err := json.Marshal(q)
179	if err != nil {
180	return err
181	}
182	transport := &http.Transport{DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
183	return (&net.Dialer{}).DialContext(ctx, "unix", socket)
184	}}
185	defer transport.CloseIdleConnections()
186	client := &http.Client{Transport: transport, Timeout: 30 * time.Second}
187	for attempt := 0; attempt < 2; attempt++ {
188	req, e := http.NewRequestWithContext(ctx, http.MethodPost, "http://instance-control/claim/recovery-rebind", bytes.NewReader(body))
189	if e != nil {
190	return e
191	}
192	response, e := client.Do(req)
193	if e != nil {
194	err = e
195	continue
196	}
197	response.Body.Close()
198	if response.StatusCode != http.StatusNoContent {
199	return hostbox.ErrChanged
200	}
201	return nil
202	}
203	return err
204	}
205	}
206