amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package hostclaim
2
3import (
4 "bytes"
5 "context"
6 "crypto/sha256"
7 "encoding/hex"
8 "encoding/json"
9 "errors"
10 "io"
11 "os"
12 "sort"
13 "strings"
14 "sync"
15
16 "github.com/carlosframework/platform/internal/hostwrite"
17 "github.com/carlosframework/platform/internal/release"
18)
19
20const originNodeLimit = 4096
21const originCountLimit = 1 << 20
22
23// OriginHead authenticates only nodes reachable from this claim's committed head.
24type OriginHead struct {
25 Digest string `json:"digest"`
26 Count uint64 `json:"count"`
27}
28
29type OriginTuple struct {
30 Incarnation string `json:"incarnation"`
31 Location string `json:"location"`
32 Region string `json:"region"`
33 Generation string `json:"generation"`
34}
35
36type originTuple = OriginTuple
37
38type originNode struct {
39 V int `json:"v"`
40 Host string `json:"host"`
41 Count uint64 `json:"count"`
42 Previous string `json:"previous,omitempty"`
43 Origin originTuple `json:"origin"`
44}
45
46func originEnabled() bool {
47 return hostwrite.Enabled(os.Getenv) && hostwrite.Enabled(func(string) string {
48 return os.Getenv("CARLOS_HOST_CLAIM_ORIGINS")
49 })
50}
51
52func originDigest(b []byte) string {
53 h := sha256.New()
54 h.Write([]byte("carlos-admitted-origin-v1\x00"))
55 h.Write(b)
56 return hex.EncodeToString(h.Sum(nil))
57}
58
59func originKey(host, digest string) string {
60 return "control/host-origins/" + host + "/" + digest + ".json"
61}
62
63func originHeadValid(h *OriginHead) bool {
64 return h == nil || h.Count > 0 && h.Count <= originCountLimit && isHex64(h.Digest) && h.Digest == strings.ToLower(h.Digest)
65}
66
67func (o originTuple) valid(host string) bool {
68 if !isHex64(o.Incarnation) || o.Incarnation != strings.ToLower(o.Incarnation) ||
69 o.Region == "" || len(o.Region) > 64 || strings.ContainsAny(o.Region, "/\\\x00") ||
70 o.Generation == "" || len(o.Generation) > 256 || o.Generation == RootGeneration ||
71 o.Generation == "." || o.Generation == ".." || strings.ContainsAny(o.Generation, "/\\\x00") {
72 return false
73 }
74 return o.Location == "instances/"+host || o.Location == "instances/"+host+"/"+o.Incarnation
75}
76
77// ValidateOriginObject classifies inert, content-addressed candidates. Creating a
78// candidate grants nothing; only the claim CAS can make one reachable.
79func ValidateOriginObject(key string, body []byte) error {
80 if len(body) == 0 || len(body) > originNodeLimit {
81 return ErrMalformed
82 }
83 d := json.NewDecoder(bytes.NewReader(body))
84 d.DisallowUnknownFields()
85 var n originNode
86 if err := d.Decode(&n); err != nil {
87 return ErrMalformed
88 }
89 if _, err := d.Token(); err != io.EOF {
90 return ErrMalformed
91 }
92 if _, err := Key(n.Host); err != nil {
93 return ErrMalformed
94 }
95 if n.V != 1 || n.Count == 0 || n.Count > originCountLimit || !n.Origin.valid(n.Host) ||
96 (n.Count == 1) != (n.Previous == "") ||
97 n.Previous != "" && (!isHex64(n.Previous) || n.Previous != strings.ToLower(n.Previous)) ||
98 key != originKey(n.Host, originDigest(body)) {
99 return ErrMalformed
100 }
101 return nil
102}
103
104func readOrigin(ctx context.Context, s release.Store, host string, h OriginHead) (originNode, error) {
105 var n originNode
106 if err := ctx.Err(); err != nil {
107 return n, err
108 }
109 key := originKey(host, h.Digest)
110 b, err := readOriginBytes(ctx, s, key)
111 if err != nil {
112 return n, err
113 }
114 if err = ValidateOriginObject(key, b); err != nil {
115 return n, err
116 }
117 if err = json.Unmarshal(b, &n); err != nil {
118 return n, err
119 }
120 if n.Host != host || n.Count != h.Count {
121 return n, ErrMalformed
122 }
123 return n, nil
124}
125
126func pendingOrigin(c *Claim, entry AdmissionRecord) error {
127 if !originEnabled() {
128 // Once written, provenance cannot silently stop collecting admissions.
129 if c.Origins != nil {
130 return ErrWrongState
131 }
132 return nil
133 }
134 if c.originAdmission != nil || c.Holder == nil {
135 return ErrWrongState
136 }
137 o := originTuple{c.Holder.Incarnation, c.Holder.Location, entry.Region, entry.Generation}
138 // The empty legacy root has no files/g-<generation>/ origin.
139 if entry.Generation == "" {
140 return nil
141 }
142 if !o.valid(c.Host) {
143 return ErrWrongState
144 }
145 c.originAdmission = &o
146 return nil
147}
148
149// commitOrigins stages immutable candidates before Update's existing claim CAS.
150// A failed or uncertain CAS leaves candidates without granting them authority.
151func commitOrigins(ctx context.Context, s release.Store, obs Observed, next *Claim) error {
152 var old *OriginHead
153 if obs.Claim != nil {
154 old = obs.Claim.Origins
155 }
156 if !sameOriginHead(old, next.Origins) {
157 return ErrWrongState
158 }
159 if next.originAdmission == nil {
160 return nil
161 }
162 if obs.Absent || obs.Claim == nil || obs.Claim.Holder == nil || next.Holder == nil {
163 return ErrWrongState
164 }
165 h := next.Holder
166 want := *next.originAdmission
167 if want.Incarnation != h.Incarnation || want.Location != h.Location || want.Generation != h.Generation ||
168 h.Session != next.Gen || len(h.Admissions) == 0 {
169 return ErrWrongState
170 }
171 a := h.Admissions[len(h.Admissions)-1]
172 if a.Session != h.Session || a.Generation != want.Generation || a.Region != want.Region {
173 return ErrWrongState
174 }
175 // Bind admission to the stored observation, never a caller-edited admission list.
176 fresh, err := Read(ctx, s, obs.Host)
177 if err != nil {
178 return err
179 }
180 before, _ := json.Marshal(obs.Claim)
181 stored, _ := json.Marshal(fresh.Claim)
182 if fresh.Absent || fresh.ETag != obs.ETag || !bytes.Equal(before, stored) {
183 return ErrConflict
184 }
185 // The chain starts empty at the independent writer boundary. Retained
186 // admissions may have been forged before that boundary; never seed them.
187 // Activation must re-copy every live origin through the #250 files backfill
188 // into a prefix admitted after the boundary.
189 head := OriginHead{}
190 if old != nil {
191 head = *old
192 last, err := readOrigin(ctx, s, obs.Host, head)
193 if err != nil {
194 return err
195 }
196 if last.Origin == want {
197 next.originAdmission = nil
198 return nil
199 }
200 }
201 if !want.valid(obs.Host) || head.Count == originCountLimit {
202 return ErrWrongState
203 }
204 node := originNode{V: 1, Host: obs.Host, Count: head.Count + 1, Previous: head.Digest, Origin: want}
205 body, err := json.Marshal(node)
206 if err != nil {
207 return err
208 }
209 key := originKey(obs.Host, originDigest(body))
210 if err = ValidateOriginObject(key, body); err != nil {
211 return err
212 }
213 if _, err = s.Put(ctx, key, body, ""); errors.Is(err, release.ErrCASConflict) {
214 existing, readErr := readOriginBytes(ctx, s, key)
215 if readErr != nil {
216 return readErr
217 }
218 if !bytes.Equal(existing, body) {
219 return ErrMalformed
220 }
221 } else if err != nil {
222 return err
223 }
224 head = OriginHead{Digest: originDigest(body), Count: node.Count}
225 next.Origins = &head
226 next.originAdmission = nil
227 return nil
228}
229
230func sameOriginHead(a, b *OriginHead) bool {
231 if a == nil || b == nil {
232 return a == nil && b == nil
233 }
234 return *a == *b
235}
236
237// OriginProofCache retains one exact batch, after authenticating its observation.
238// Construct it with NewOriginProofCache; its store cannot be replaced.
239type OriginProofCache struct {
240 mu sync.Mutex
241 store release.Store
242 key string
243 result map[OriginTuple]bool
244}
245
246// NewOriginProofCache binds proof and reuse to one authoritative store. Queries
247// accept no store argument, so a cache cannot carry answers to another store.
248func NewOriginProofCache(s release.Store) *OriginProofCache {
249 return &OriginProofCache{store: s}
250}
251
252func readOriginBytes(ctx context.Context, s release.Store, key string) ([]byte, error) {
253 if err := ctx.Err(); err != nil {
254 return nil, err
255 }
256 // Require streaming through every OpenObject wrapper in the chain.
257 reader, _, err := release.OpenObjectStreaming(ctx, s, key)
258 if errors.Is(err, errors.ErrUnsupported) {
259 return nil, ErrMalformed
260 }
261 if err != nil {
262 return nil, err
263 }
264 b, readErr := io.ReadAll(io.LimitReader(reader, originNodeLimit+1))
265 closeErr := reader.Close()
266 if readErr != nil {
267 return nil, readErr
268 }
269 if len(b) > originNodeLimit {
270 return nil, ErrMalformed
271 }
272 if closeErr != nil {
273 return nil, closeErr
274 }
275 if err := ctx.Err(); err != nil {
276 return nil, err
277 }
278 return b, nil
279}
280
281// AdmittedOrigins authenticates once and proves the complete requested set in
282// one walk, stopping once every tuple is proven. A malformed or unreadable node
283// needed by the walk returns an error without partial results.
284func AdmittedOrigins(ctx context.Context, s release.Store, obs Observed, set []OriginTuple) (map[OriginTuple]bool, error) {
285 return NewOriginProofCache(s).Proven(ctx, obs, set)
286}
287
288func (cache *OriginProofCache) Proven(ctx context.Context, obs Observed, set []OriginTuple) (map[OriginTuple]bool, error) {
289 if cache == nil {
290 return nil, ErrMalformed
291 }
292 s := cache.store
293 cache.mu.Lock()
294 defer cache.mu.Unlock()
295 invalidate := func() { cache.key, cache.result = "", nil }
296 if err := ctx.Err(); err != nil {
297 invalidate()
298 return nil, err
299 }
300 if s == nil || obs.Absent || obs.Claim == nil || obs.ETag == "" || obs.Host != obs.Claim.Host {
301 invalidate()
302 return nil, ErrMalformed
303 }
304 wanted := make(map[OriginTuple]bool, len(set))
305 for _, tuple := range set {
306 if !tuple.valid(obs.Host) {
307 invalidate()
308 return nil, ErrMalformed
309 }
310 wanted[tuple] = false
311 }
312 fresh, err := Read(ctx, s, obs.Host)
313 if err != nil {
314 invalidate()
315 return nil, err
316 }
317 before, marshalErr := json.Marshal(obs.Claim)
318 stored, storedErr := json.Marshal(fresh.Claim)
319 if fresh.Absent || fresh.ETag != obs.ETag || marshalErr != nil || storedErr != nil || !bytes.Equal(before, stored) {
320 invalidate()
321 return nil, ErrConflict
322 }
323 if err := ctx.Err(); err != nil {
324 invalidate()
325 return nil, err
326 }
327 // The store is fixed at construction. The key binds host, head, exact tuple
328 // set, ETag and full claim body, including legacy admissions.
329 tuples := make([]string, 0, len(wanted))
330 for tuple := range wanted {
331 b, _ := json.Marshal(tuple)
332 tuples = append(tuples, string(b))
333 }
334 sort.Strings(tuples)
335 keyBytes, _ := json.Marshal(struct {
336 Host, ETag, Claim string
337 Set []string
338 }{obs.Host, obs.ETag, string(stored), tuples})
339 key := string(keyBytes)
340 if cache.key == key {
341 for tuple := range wanted {
342 wanted[tuple] = cache.result[tuple]
343 }
344 return wanted, nil
345 }
346 invalidate()
347 c := fresh.Claim
348 // The chain starts empty: a claim with no recorded head proves nothing.
349 // Retained admissions are never evidence (a box can write them), so live
350 // origins are proven only through admissions recorded after the boundary
351 // or the #250 files backfill into an admitted prefix.
352 if c.Origins != nil {
353 remaining := len(wanted)
354 for head := *c.Origins; head.Count > 0 && remaining > 0; {
355 node, err := readOrigin(ctx, s, obs.Host, head)
356 if err != nil {
357 return nil, err
358 }
359 if proven, ok := wanted[node.Origin]; ok && !proven {
360 remaining--
361 wanted[node.Origin] = true
362 }
363 head = OriginHead{Digest: node.Previous, Count: head.Count - 1}
364 }
365 }
366 if err := ctx.Err(); err != nil {
367 return nil, err
368 }
369 cache.key = key
370 cache.result = make(map[OriginTuple]bool, len(wanted))
371 for tuple, proven := range wanted {
372 cache.result[tuple] = proven
373 }
374 return wanted, nil
375}
376
377// AdmittedOrigin is the singleton convenience form; multi-origin callers use
378// AdmittedOrigins or one OriginProofCache.Proven call for their entire set.
379func AdmittedOrigin(ctx context.Context, s release.Store, obs Observed, incarnation, location, region, generation string) (bool, error) {
380 tuple := OriginTuple{incarnation, location, region, generation}
381 proven, err := AdmittedOrigins(ctx, s, obs, []OriginTuple{tuple})
382 return proven[tuple], err
383}
384