oficina / home Public

Clone
git clone https://amadan.net/oficina/home

Plain git — no account needed to clone.

Download

Download this file

1package roster
2
3import (
4 "bytes"
5 "context"
6 "encoding/base64"
7 "errors"
8 "io"
9 "net/http"
10 "strings"
11 "sync"
12 "testing"
13 "time"
14
15 "amadan.net/oficina/home/handoff/seal"
16 family "amadan.net/rastrillo/rastrillo/crypto"
17 "github.com/go-json-experiment/json"
18)
19
20type fakeClock struct {
21 mu sync.Mutex
22 wall time.Time
23 mono time.Duration
24}
25
26func (f *fakeClock) Now() (time.Time, time.Duration) {
27 f.mu.Lock()
28 defer f.mu.Unlock()
29 return f.wall, f.mono
30}
31func (f *fakeClock) step(d time.Duration) {
32 f.mu.Lock()
33 defer f.mu.Unlock()
34 f.wall = f.wall.Add(d)
35 f.mono += d
36}
37func (f *fakeClock) wallStep(d time.Duration) {
38 f.mu.Lock()
39 defer f.mu.Unlock()
40 f.wall = f.wall.Add(d)
41}
42
43type memory struct {
44 mu sync.Mutex
45 raw []byte
46 fail bool
47}
48
49func (m *memory) Load(context.Context) ([]byte, error) {
50 m.mu.Lock()
51 defer m.mu.Unlock()
52 if m.raw == nil {
53 return nil, ErrNoEvidence
54 }
55 return bytes.Clone(m.raw), nil
56}
57func (m *memory) Save(_ context.Context, b []byte) error {
58 m.mu.Lock()
59 defer m.mu.Unlock()
60 if m.fail {
61 return errors.New("secret-storage-error")
62 }
63 m.raw = bytes.Clone(b)
64 return nil
65}
66func (m *memory) failWrites(v bool) { m.mu.Lock(); defer m.mu.Unlock(); m.fail = v }
67
68type transport func(*http.Request) (*http.Response, error)
69
70func (f transport) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) }
71func fixed(v byte) string { return base64.RawURLEncoding.EncodeToString(bytes.Repeat([]byte{v}, 32)) }
72func ptr[T any](v T) *T { return &v }
73func person(n int) string {
74 s := []byte(strings.Repeat("a", 26))
75 for i := 25; n > 0; i-- {
76 s[i] = "abcdefghijklmnopqrstuvwxyz"[n%26]
77 n /= 26
78 }
79 return "pr_" + string(s)
80}
81
82type fixture struct {
83 t *testing.T
84 op, key *family.Keypair
85 record Provisioning
86 verified VerifiedRecord
87 cfg Config
88 clock *fakeClock
89 disk *memory
90 mu sync.Mutex
91 handler transport
92 calls int
93}
94
95func setup(t *testing.T) *fixture {
96 t.Helper()
97 op, e := family.Generate()
98 if e != nil {
99 t.Fatal(e)
100 }
101 key, e := family.Generate()
102 if e != nil {
103 t.Fatal(e)
104 }
105 f := &fixture{t: t, op: op, key: key, clock: &fakeClock{wall: time.Unix(1800000000, 0)}, disk: &memory{}}
106 f.record = Provisioning{Version: 1, Operation: "op1", Sequence: 1, Bindings: Bindings{Issuer: "https://id.example", Directory: "https://home.example", Consumer: "docs-A", App: "docs", Team: "TeamA", Audience: "https://docs.example"}, Generation: fixed(1), MinEpoch: 1, Keys: []Key{{"roster1", base64.RawURLEncoding.EncodeToString(key.SignPub())}}}
107 f.verified = f.verify(f.record)
108 s, e := seal.New(bytes.Repeat([]byte{9}, 32), "roster-test")
109 if e != nil {
110 t.Fatal(e)
111 }
112 f.cfg = Config{Record: f.verified, Token: fixed(7), Clock: f.clock, Durable: f.disk, Sealer: s}
113 f.handler = func(r *http.Request) (*http.Response, error) {
114 return f.response(r, 1, []MemberEntry{{person(1), "member"}}, true, ""), nil
115 }
116 f.cfg.Transport = transport(func(r *http.Request) (*http.Response, error) {
117 if r.URL.Scheme != "https" || r.URL.Host != "home.example" || r.URL.Path != "/roster" || r.Header.Get("Authorization") != "Bearer "+fixed(7) || !ValidChallenge(r.URL.Query().Get("fetch")) {
118 t.Error("outbound boundary")
119 }
120 f.mu.Lock()
121 f.calls++
122 h := f.handler
123 f.mu.Unlock()
124 return h(r)
125 })
126 return f
127}
128func (f *fixture) verify(p Provisioning) VerifiedRecord {
129 f.t.Helper()
130 raw, e := SignProvisioning(f.op, "operator", p)
131 if e != nil {
132 f.t.Fatal(e)
133 }
134 v, e := VerifyProvisioning(raw, []Key{{"operator", base64.RawURLEncoding.EncodeToString(f.op.SignPub())}}, f.record.Bindings)
135 if e != nil {
136 f.t.Fatal(e)
137 }
138 return v
139}
140func (f *fixture) cache() *Cache {
141 f.t.Helper()
142 c, e := New(context.Background(), f.cfg)
143 if e != nil {
144 f.t.Fatal(e)
145 }
146 return c
147}
148func (f *fixture) set(h transport) { f.mu.Lock(); defer f.mu.Unlock(); f.handler = h }
149func (f *fixture) page(r *http.Request, epoch int64, m []MemberEntry, complete bool, cursor string) Page {
150 q := r.URL.Query()
151 i := 0
152 if q.Get("cursor") != "" {
153 i = 1
154 }
155 p := Page{Version: 1, Type: "page", Directory: f.record.Bindings.Directory, Consumer: f.record.Bindings.Consumer, Team: f.record.Bindings.Team, Audience: f.record.Bindings.Audience, Fetch: q.Get("fetch"), RequestCursorHash: CursorHash(q.Get("cursor")), Generation: f.record.Generation, Epoch: epoch, Index: &i, Members: &m, Complete: &complete}
156 if !complete {
157 p.Cursor = &cursor
158 }
159 return p
160}
161func (f *fixture) signed(p Page, status int) *http.Response {
162 f.t.Helper()
163 b, e := SignPage(f.key, "roster1", p)
164 if e != nil {
165 f.t.Fatal(e)
166 }
167 return &http.Response{StatusCode: status, Header: http.Header{"Content-Type": {"application/json"}}, Body: io.NopCloser(bytes.NewReader(b))}
168}
169func (f *fixture) response(r *http.Request, e int64, m []MemberEntry, complete bool, cursor string) *http.Response {
170 return f.signed(f.page(r, e, m, complete, cursor), 200)
171}
172func (f *fixture) check(c *Cache, n int) Decision {
173 return c.Check(context.Background(), f.record.Bindings.Issuer, f.record.Bindings.Team, person(n))
174}
175func state(t *testing.T, d Decision, s State) {
176 t.Helper()
177 if d.State != s {
178 t.Fatalf("state %d want %d", d.State, s)
179 }
180}
181
182func TestPrivacyAndSixtySecondCeiling(t *testing.T) {
183 f := setup(t)
184 c := f.cache()
185 d := f.check(c, 1)
186 state(t, d, Member)
187 raw, _ := f.disk.Load(context.Background())
188 for _, v := range []string{fixed(7), person(1), "member", "secret-storage-error"} {
189 if bytes.Contains(raw, []byte(v)) {
190 t.Fatal("readable durable data")
191 }
192 }
193 f.set(func(*http.Request) (*http.Response, error) { return nil, errors.New(fixed(7) + person(1)) })
194 f.clock.step(30 * time.Second)
195 if e := c.Refresh(context.Background()); e != ErrUnavailable {
196 t.Fatalf("unsafe error %v", e)
197 }
198 state(t, f.check(c, 1), Member)
199 f.clock.step(29 * time.Second)
200 if !d.Lease.Valid() {
201 t.Fatal("early expiry")
202 }
203 f.clock.step(time.Second)
204 if d.Lease.Valid() {
205 t.Fatal("standing survived ceiling")
206 }
207 state(t, f.check(c, 1), Unknown)
208}
209func TestNoPartialPublishOrAgeExtension(t *testing.T) {
210 f := setup(t)
211 c := f.cache()
212 state(t, f.check(c, 1), Member)
213 old, _ := f.disk.Load(context.Background())
214 batch := make([]MemberEntry, 100)
215 for i := range batch {
216 batch[i] = MemberEntry{person(i + 100), "member"}
217 }
218 f.set(func(r *http.Request) (*http.Response, error) {
219 if r.URL.Query().Get("cursor") != "" {
220 return nil, errors.New("partial")
221 }
222 return f.response(r, 2, batch, false, fixed(4)), nil
223 })
224 f.clock.step(30 * time.Second)
225 if c.Refresh(context.Background()) == nil {
226 t.Fatal("partial accepted")
227 }
228 state(t, f.check(c, 100), NotMember)
229 state(t, f.check(c, 1), Member)
230 after, _ := f.disk.Load(context.Background())
231 if !bytes.Equal(old, after) {
232 t.Fatal("partial persisted")
233 }
234 f.clock.step(30 * time.Second)
235 state(t, f.check(c, 1), Unknown)
236}
237func TestEmptyRosterAndRestartEpochFloor(t *testing.T) {
238 f := setup(t)
239 f.set(func(r *http.Request) (*http.Response, error) { return f.response(r, 2, []MemberEntry{}, true, ""), nil })
240 c := f.cache()
241 state(t, f.check(c, 1), NotMember)
242 f.clock.step(30 * time.Second)
243 f.set(func(r *http.Request) (*http.Response, error) {
244 return f.response(r, 2, []MemberEntry{{person(1), "member"}}, true, ""), nil
245 })
246 if c.Refresh(context.Background()) == nil {
247 t.Fatal("changed empty epoch accepted")
248 }
249 state(t, f.check(c, 1), NotMember)
250 c = f.cache()
251 if c.Refresh(context.Background()) == nil {
252 t.Fatal("restart forgot digest")
253 }
254 state(t, f.check(c, 1), Unknown)
255}
256func TestGenerationRequiresOperatorApproval(t *testing.T) {
257 f := setup(t)
258 c := f.cache()
259 d := f.check(c, 1)
260 f.clock.step(30 * time.Second)
261 f.set(func(r *http.Request) (*http.Response, error) {
262 p := f.page(r, 2, []MemberEntry{}, true, "")
263 p.Generation = fixed(2)
264 return f.signed(p, 200), nil
265 })
266 if c.Refresh(context.Background()) == nil {
267 t.Fatal("generation accepted")
268 }
269 if d.Lease.Valid() {
270 t.Fatal("old lease survived")
271 }
272 f.set(func(r *http.Request) (*http.Response, error) {
273 return f.response(r, 1, []MemberEntry{{person(1), "member"}}, true, ""), nil
274 })
275 f.clock.step(5 * time.Second)
276 state(t, f.check(c, 1), Unknown)
277 c = f.cache()
278 state(t, f.check(c, 1), Unknown)
279 n := f.record
280 n.Sequence = 2
281 n.Operation = "op2"
282 n.Generation = fixed(2)
283 if c.UpdateRecord(context.Background(), f.verify(n)) == nil {
284 t.Fatal("missing confirmation accepted")
285 }
286 n.OwnerConfirmation = "owner-confirmation"
287 if e := c.UpdateRecord(context.Background(), f.verify(n)); e != nil {
288 t.Fatal(e)
289 }
290 f.set(func(r *http.Request) (*http.Response, error) {
291 p := f.page(r, 1, []MemberEntry{{person(1), "owner"}}, true, "")
292 p.Generation = fixed(2)
293 return f.signed(p, 200), nil
294 })
295 state(t, f.check(c, 1), Member)
296}
297func TestClockToleranceAndColdRestart(t *testing.T) {
298 f := setup(t)
299 c := f.cache()
300 d := f.check(c, 1)
301 f.clock.wallStep(-ClockTolerance)
302 if !d.Lease.Valid() {
303 t.Fatal("tolerated slew fenced")
304 }
305 f.clock.wallStep(-ClockTolerance - time.Nanosecond)
306 if d.Lease.Valid() {
307 t.Fatal("clock rollback missed")
308 }
309 f.clock.step(5 * time.Second)
310 state(t, f.check(c, 1), Member)
311 f.set(func(*http.Request) (*http.Response, error) { return nil, errors.New("offline") })
312 c = f.cache()
313 state(t, f.check(c, 1), Unknown)
314}
315func TestRecordUpdateFencesStagingAndWriteFailure(t *testing.T) {
316 f := setup(t)
317 c := f.cache()
318 d := f.check(c, 1)
319 old, _ := f.disk.Load(context.Background())
320 n := f.record
321 n.Sequence = 2
322 n.Operation = "op2"
323 f.disk.failWrites(true)
324 if e := c.UpdateRecord(context.Background(), f.verify(n)); e != ErrUnavailable {
325 t.Fatalf("storage error %v", e)
326 }
327 if !d.Lease.Valid() {
328 t.Fatal("failed write changed lease")
329 }
330 after, _ := f.disk.Load(context.Background())
331 if !bytes.Equal(old, after) {
332 t.Fatal("failed write persisted")
333 }
334 f.disk.failWrites(false)
335 entered := make(chan struct{})
336 release := make(chan struct{})
337 f.clock.step(30 * time.Second)
338 f.set(func(r *http.Request) (*http.Response, error) {
339 close(entered)
340 <-release
341 return f.response(r, 2, []MemberEntry{{person(2), "member"}}, true, ""), nil
342 })
343 done := make(chan error)
344 go func() { done <- c.Refresh(context.Background()) }()
345 <-entered
346 if e := c.UpdateRecord(context.Background(), f.verify(n)); e != nil {
347 t.Fatal(e)
348 }
349 close(release)
350 if <-done == nil {
351 t.Fatal("old staging published")
352 }
353 if d.Lease.Valid() {
354 t.Fatal("old lease survived")
355 }
356 if c.UpdateRecord(context.Background(), f.verified) == nil {
357 t.Fatal("old record replay")
358 }
359}
360func TestSingleFlight(t *testing.T) {
361 f := setup(t)
362 entered := make(chan struct{})
363 release := make(chan struct{})
364 var once sync.Once
365 f.set(func(r *http.Request) (*http.Response, error) {
366 once.Do(func() { close(entered) })
367 <-release
368 return f.response(r, 1, []MemberEntry{{person(1), "member"}}, true, ""), nil
369 })
370 c := f.cache()
371 var wg sync.WaitGroup
372 for i := 0; i < 24; i++ {
373 wg.Add(1)
374 go func() { defer wg.Done(); state(t, f.check(c, 1), Member) }()
375 }
376 <-entered
377 close(release)
378 wg.Wait()
379 f.mu.Lock()
380 defer f.mu.Unlock()
381 if f.calls != 1 {
382 t.Fatalf("calls %d", f.calls)
383 }
384}
385func TestFetchDeadlineAndSaveFailure(t *testing.T) {
386 for _, slow := range []bool{false, true} {
387 t.Run(strings.Repeat("slow", map[bool]int{false: 0, true: 1}[slow])+"failure", func(t *testing.T) {
388 f := setup(t)
389 c := f.cache()
390 f.set(func(r *http.Request) (*http.Response, error) {
391 if slow {
392 f.clock.step(FetchTimeout)
393 } else {
394 f.disk.failWrites(true)
395 }
396 return f.response(r, 2, []MemberEntry{{person(1), "member"}}, true, ""), nil
397 })
398 if c.Refresh(context.Background()) == nil {
399 t.Fatal("invalid refresh accepted")
400 }
401 state(t, f.check(c, 1), Unknown)
402 })
403 }
404}
405func TestRefusalsNeverRenew(t *testing.T) {
406 for _, name := range []string{"challenge", "cursor-hash", "team", "directory", "audience", "role", "subject", "index", "extra", "duplicate", "context", "signature", "oversize", "compression", "redirect", "304", "epoch"} {
407 t.Run(name, func(t *testing.T) {
408 f := setup(t)
409 c := f.cache()
410 state(t, f.check(c, 1), Member)
411 f.clock.step(30 * time.Second)
412 f.set(func(r *http.Request) (*http.Response, error) {
413 p := f.page(r, 1, []MemberEntry{{person(1), "member"}}, true, "")
414 switch name {
415 case "challenge":
416 p.Fetch = fixed(9)
417 case "cursor-hash":
418 p.RequestCursorHash = ""
419 case "team":
420 p.Team = "TeamB"
421 case "directory":
422 p.Directory = "https://foreign.example"
423 case "audience":
424 p.Audience = "https://foreign.example"
425 case "role":
426 (*p.Members)[0].Role = "admin"
427 case "subject":
428 (*p.Members)[0].Subject = "someone@example.test"
429 case "index":
430 p.Index = ptr(1)
431 case "epoch":
432 p.Epoch = 0
433 }
434 res := f.signed(p, 200)
435 switch name {
436 case "extra", "duplicate":
437 raw, _ := json.Marshal(p)
438 tail := `,"extra":1}`
439 if name == "duplicate" {
440 tail = `,"epoch":1}`
441 }
442 raw = append(raw[:len(raw)-1], []byte(tail)...)
443 sig, _ := family.Sign(f.key, PageContext, raw)
444 b, _ := json.Marshal(envelope{1, "roster1", base64.RawURLEncoding.EncodeToString(raw), base64.RawURLEncoding.EncodeToString(sig)})
445 res.Body = io.NopCloser(bytes.NewReader(b))
446 case "context":
447 b, _ := encode(f.key, "roster1", ProvisioningContext, p)
448 res.Body = io.NopCloser(bytes.NewReader(b))
449 case "signature":
450 b, _ := io.ReadAll(res.Body)
451 var e envelope
452 _ = json.Unmarshal(b, &e)
453 e.Signature = base64.RawURLEncoding.EncodeToString(make([]byte, 64))
454 b, _ = json.Marshal(e)
455 res.Body = io.NopCloser(bytes.NewReader(b))
456 case "oversize":
457 res.Body = io.NopCloser(strings.NewReader(strings.Repeat("a", BodyLimit+1)))
458 case "compression":
459 res.Header.Set("Content-Encoding", "gzip")
460 case "redirect":
461 res.StatusCode = 302
462 res.Header.Set("Location", "https://foreign.example/roster")
463 case "304":
464 res.StatusCode = 304
465 }
466 return res, nil
467 })
468 if c.Refresh(context.Background()) == nil {
469 t.Fatal("invalid response accepted")
470 }
471 f.clock.step(30 * time.Second)
472 state(t, f.check(c, 1), Unknown)
473 })
474 }
475}
476