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 "crypto/sha256"
6 "encoding/hex"
7 "encoding/json"
8 "errors"
9 "io"
10 "net"
11 "net/http"
12 "slices"
13 "sort"
14 "time"
15
16 "github.com/carlosframework/platform/internal/hostbox"
17 "github.com/carlosframework/platform/internal/hostguard"
18 "github.com/carlosframework/platform/internal/instancectl"
19)
20
21type ClaimStopHTTP struct {
22 Coordinator ClaimStopCoordinator
23 Authenticate func(context.Context, net.Conn) (StopPeer, error)
24}
25type stopPeerValue struct{ peer StopPeer }
26
27func (p stopPeerValue) AuthenticatedPeer(context.Context) (StopPeer, error) { return p.peer, nil }
28
29type stopFrame struct {
30 FenceID string
31 Result *StopResult
32 Failed bool
33}
34type stopStreamDrain struct {
35 w http.ResponseWriter
36 decode *json.Decoder
37 peer StopPeer
38 recheck func(context.Context) (StopPeer, error)
39}
40
41func (s stopStreamDrain) DrainClaim(ctx context.Context, req StopRequest, fence string) (DrainAcknowledgement, error) {
42 if err := json.NewEncoder(s.w).Encode(stopFrame{FenceID: fence}); err != nil {
43 return DrainAcknowledgement{}, err
44 }
45 if err := http.NewResponseController(s.w).Flush(); err != nil {
46 return DrainAcknowledgement{}, err
47 }
48 var ack DrainAcknowledgement
49 if err := s.decode.Decode(&ack); err != nil {
50 return ack, err
51 }
52 if ack.ProcessStart != s.peer.ProcessStart || ack.ControllerID != s.peer.ControllerID || ack.FenceID != fence {
53 return ack, ErrClaimStopIdentity
54 }
55 current, err := s.recheck(ctx)
56 if err != nil || current != s.peer {
57 return ack, ErrClaimStopIdentity
58 }
59 return ack, ctx.Err()
60}
61func (h ClaimStopHTTP) ServeHTTP(w http.ResponseWriter, r *http.Request) {
62 if r.Method != http.MethodPost || h.Authenticate == nil {
63 w.WriteHeader(http.StatusServiceUnavailable)
64 return
65 }
66 conn := instancectl.RequestConn(r.Context())
67 if conn == nil {
68 w.WriteHeader(http.StatusUnauthorized)
69 return
70 }
71 peer, err := h.Authenticate(r.Context(), conn)
72 if err != nil {
73 w.WriteHeader(http.StatusUnauthorized)
74 return
75 }
76 if r.URL.Path == "/claim/writers-stopped" {
77 h.serveWritersStopped(w, r, peer)
78 return
79 }
80 rc := http.NewResponseController(w)
81 if err := rc.EnableFullDuplex(); err != nil {
82 w.WriteHeader(http.StatusServiceUnavailable)
83 return
84 }
85 deadline := time.Now().Add(2 * time.Minute)
86 _ = rc.SetReadDeadline(deadline)
87 _ = rc.SetWriteDeadline(deadline)
88 decoder := json.NewDecoder(io.LimitReader(r.Body, 1<<20))
89 decoder.DisallowUnknownFields()
90 var req StopRequest
91 if err := decoder.Decode(&req); err != nil {
92 w.WriteHeader(http.StatusBadRequest)
93 return
94 }
95 w.Header().Set("Content-Type", "application/x-ndjson")
96 c := h.Coordinator
97 c.Peers = stopPeerValue{peer}
98 c.Drainer = stopStreamDrain{w: w, decode: decoder, peer: peer, recheck: func(ctx context.Context) (StopPeer, error) { return h.Authenticate(ctx, conn) }}
99 result, err := c.FenceAndStop(r.Context(), req)
100 if err != nil {
101 _ = json.NewEncoder(w).Encode(stopFrame{Failed: true})
102 return
103 }
104 _ = json.NewEncoder(w).Encode(stopFrame{Result: &result})
105}
106func digestStop(v any) string {
107 b, _ := json.Marshal(v)
108 s := sha256.Sum256(b)
109 return hex.EncodeToString(s[:])
110}
111func DrainClaimAcknowledgement(fence, controller, start string, ids []string, completed ...hostbox.ProducerCompletion) DrainAcknowledgement {
112 ids = append([]string{}, ids...)
113 sort.Strings(ids)
114 ack := DrainAcknowledgement{FenceID: fence, ControllerID: controller, ProcessStart: start, ProducerIDs: ids}
115 if len(completed) == 1 {
116 ack.Completed = completed[0]
117 }
118 ack.Digest = claimDrainDigest(ack)
119 return ack
120}
121func CollectClaimStop(ctx context.Context, path string, req StopRequest, drain StopDrainer) (StopResult, error) {
122 return collectClaimStop(ctx, path, req, drain, func(ctx context.Context, _, _ string) (net.Conn, error) {
123 return (&net.Dialer{}).DialContext(ctx, "unix", path)
124 })
125}
126func collectClaimStop(ctx context.Context, path string, req StopRequest, drain StopDrainer, dial func(context.Context, string, string) (net.Conn, error)) (StopResult, error) {
127 if drain == nil {
128 return StopResult{}, ErrClaimStopUnavailable
129 }
130 ctx, cancel := context.WithTimeout(ctx, 2*time.Minute)
131 defer cancel()
132 reader, writer := io.Pipe()
133 defer reader.Close()
134 defer writer.Close()
135 stopCancel := context.AfterFunc(ctx, func() {
136 _ = reader.CloseWithError(ctx.Err())
137 _ = writer.CloseWithError(ctx.Err())
138 })
139 defer stopCancel()
140 transport := &http.Transport{DisableKeepAlives: true, DialContext: dial}
141 defer transport.CloseIdleConnections()
142 request, err := http.NewRequestWithContext(ctx, http.MethodPost, "http://control/claim/stop", reader)
143 if err != nil {
144 return StopResult{}, err
145 }
146 written := make(chan error, 1)
147 go func() { written <- json.NewEncoder(writer).Encode(req) }()
148 response, err := (&http.Client{Transport: transport}).Do(request)
149 if err != nil {
150 return StopResult{}, err
151 }
152 defer func() {
153 _ = writer.Close()
154 _ = response.Body.Close()
155 }()
156 if response.StatusCode != http.StatusOK {
157 return StopResult{}, ErrClaimStopUnavailable
158 }
159 if err := <-written; err != nil {
160 return StopResult{}, err
161 }
162 decoder := json.NewDecoder(io.LimitReader(response.Body, 1<<20))
163 decoder.DisallowUnknownFields()
164 var frame stopFrame
165 if err := decoder.Decode(&frame); err != nil {
166 return StopResult{}, err
167 }
168 if frame.Failed {
169 return StopResult{}, ErrClaimStopUnavailable
170 }
171 if frame.Result != nil {
172 return validateTransportStop(req, *frame.Result)
173 }
174 if frame.FenceID == "" {
175 return StopResult{}, ErrClaimStopIdentity
176 }
177 ack, err := drain.DrainClaim(ctx, req, frame.FenceID)
178 if err != nil {
179 return StopResult{}, err
180 }
181 if err := json.NewEncoder(writer).Encode(ack); err != nil {
182 return StopResult{}, err
183 }
184 _ = writer.Close()
185 if err := decoder.Decode(&frame); err != nil {
186 return StopResult{}, err
187 }
188 if frame.Failed || frame.Result == nil {
189 return StopResult{}, ErrClaimStopUnavailable
190 }
191 result := *frame.Result
192 if result.Digest != digestRawStop(result.Bytes) {
193 return StopResult{}, errors.New("claim stop receipt digest mismatch")
194 }
195 return validateTransportStop(req, result)
196}
197func digestRawStop(b []byte) string { s := sha256.Sum256(b); return hex.EncodeToString(s[:]) }
198
199func validateTransportStop(req StopRequest, result StopResult) (StopResult, error) {
200 if result.Digest != digestRawStop(result.Bytes) {
201 return StopResult{}, ErrClaimStopReceipt
202 }
203 var receipt StopReceipt
204 if err := json.Unmarshal(result.Bytes, &receipt); err != nil {
205 return StopResult{}, err
206 }
207 if receipt.Region != req.Region || receipt.Host != req.Host || receipt.Incarnation != req.Incarnation || receipt.InstallationStamp != req.InstallationStamp || receipt.WriterSession != req.WriterSession || receipt.OperationID != req.OperationID || receipt.Phase != req.Phase || receipt.ObligationID != req.ObligationID || receipt.Box != req.Box || receipt.Enrollment != req.Enrollment || receipt.BootID != req.BootID || receipt.ControllerID != req.ControllerID || receipt.Binding != req.ExpectedDirectoryRevision || !result.EvidenceAt.Equal(receipt.EvidenceAt) {
208 return StopResult{}, ErrClaimStopIdentity
209 }
210 // The numeral is the fence's, not this request's: a retry after roster
211 // writes returns the original receipt with its original numeral.
212 numeral, ok := receiptRosterRevision(receipt.DirectoryRevision)
213 if !ok {
214 return StopResult{}, ErrClaimStopReceipt
215 }
216
217 expected := hostguard.FenceTarget{Region: receipt.Region, FenceID: receipt.FenceID, ClaimGen: receipt.ClaimGen, ObligationID: receipt.ObligationID, Host: receipt.Host, Incarnation: receipt.Incarnation, Stamp: receipt.InstallationStamp, Session: receipt.WriterSession, Op: receipt.OperationID, Phase: receipt.Phase, Box: receipt.Box, Enrollment: receipt.Enrollment, BootID: receipt.BootID, ControllerID: receipt.ControllerID, ClaimETag: receipt.ClaimETag, DirectoryRevision: receipt.Binding, RosterRevision: numeral}
218 if result.Target != expected {
219 return StopResult{}, ErrClaimStopIdentity
220 }
221 return result, nil
222}
223
224// validateClaimDrainProducers is shared by the kernel verifier and fixtures.
225func validateClaimDrainProducers(producers []string, ack DrainAcknowledgement, completed ...hostbox.ProducerCompletion) error {
226 expected := hostbox.ProducerCompletion{}
227 if len(completed) > 1 {
228 return ErrClaimStopIdentity
229 }
230 if len(completed) == 1 {
231 expected = completed[0]
232 }
233 ids := append([]string(nil), producers...)
234 sort.Strings(ids)
235 if !expected.Valid() || expected != ack.Completed || !slices.Equal(ids, ack.ProducerIDs) || ack.Digest != claimDrainDigest(ack) {
236 return ErrClaimStopIdentity
237 }
238 return nil
239}
240
241func claimDrainDigest(ack DrainAcknowledgement) string {
242 if ack.Completed == (hostbox.ProducerCompletion{}) {
243 return digestStop(struct {
244 FenceID, ControllerID, ProcessStart string
245 ProducerIDs []string
246 }{ack.FenceID, ack.ControllerID, ack.ProcessStart, ack.ProducerIDs})
247 }
248 return digestStop(struct {
249 FenceID, ControllerID, ProcessStart string
250 ProducerIDs []string
251 Completed hostbox.ProducerCompletion
252 }{ack.FenceID, ack.ControllerID, ack.ProcessStart, ack.ProducerIDs, ack.Completed})
253}
254