amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package lease_test
2
3import (
4 "context"
5 "crypto/hmac"
6 "crypto/md5"
7 "crypto/sha256"
8 "encoding/hex"
9 "errors"
10 "fmt"
11 "io"
12 "net/http"
13 "net/http/httptest"
14 "strings"
15 "sync"
16 "testing"
17 "time"
18
19 "github.com/aws/aws-sdk-go-v2/aws"
20
21 "github.com/carlosframework/platform/internal/lease"
22 "github.com/carlosframework/platform/internal/lease/leasetest"
23)
24
25const (
26 stagingAK = "AKIATEST"
27 stagingSecret = "test-secret"
28 stagingRegion = "eu-west-2"
29)
30
31var stagingCreds = aws.CredentialsProviderFunc(func(context.Context) (aws.Credentials, error) {
32 return aws.Credentials{AccessKeyID: stagingAK, SecretAccessKey: stagingSecret}, nil
33})
34
35func hmacSHA(key []byte, s string) []byte {
36 h := hmac.New(sha256.New, key)
37 h.Write([]byte(s))
38 return h.Sum(nil)
39}
40
41// verifySigV4 recomputes the signature of a received request from first
42// principles (the AWS spec's canonical request and key derivation, not the
43// SDK's code), so the hand-built request is checked against an independent
44// reading of what S3 would check. It returns the X-Amz-Date it saw.
45func verifySigV4(r *http.Request) error {
46 auth := r.Header.Get("Authorization")
47 rest, ok := strings.CutPrefix(auth, "AWS4-HMAC-SHA256 ")
48 if !ok {
49 return fmt.Errorf("authorization %q", auth)
50 }
51 f := map[string]string{}
52 for _, part := range strings.Split(rest, ", ") {
53 k, v, _ := strings.Cut(part, "=")
54 f[k] = v
55 }
56 date := r.Header.Get("X-Amz-Date")
57 scope := date[:8] + "/" + stagingRegion + "/s3/aws4_request"
58 if f["Credential"] != stagingAK+"/"+scope {
59 return fmt.Errorf("credential scope %q, want %q", f["Credential"], scope)
60 }
61 var canon strings.Builder
62 canon.WriteString(r.Method + "\n")
63 canon.WriteString(strings.SplitN(r.RequestURI, "?", 2)[0] + "\n") // the path as sent: S3 canonicalizes it once-encoded
64 canon.WriteString(r.URL.RawQuery + "\n")
65 for _, h := range strings.Split(f["SignedHeaders"], ";") {
66 v := r.Header.Get(h)
67 if h == "host" {
68 v = r.Host
69 }
70 if h == "content-length" {
71 v = fmt.Sprint(r.ContentLength)
72 }
73 canon.WriteString(h + ":" + strings.TrimSpace(v) + "\n")
74 }
75 canon.WriteString("\n" + f["SignedHeaders"] + "\n" + r.Header.Get("X-Amz-Content-Sha256"))
76 sum := sha256.Sum256([]byte(canon.String()))
77 sts := "AWS4-HMAC-SHA256\n" + date + "\n" + scope + "\n" + hex.EncodeToString(sum[:])
78 k := hmacSHA([]byte("AWS4"+stagingSecret), date[:8])
79 k = hmacSHA(k, stagingRegion)
80 k = hmacSHA(k, "s3")
81 k = hmacSHA(k, "aws4_request")
82 if got := hex.EncodeToString(hmacSHA(k, sts)); got != f["Signature"] {
83 return fmt.Errorf("signature mismatch\ncanonical request:\n%s", canon.String())
84 }
85 return nil
86}
87
88type recorded struct {
89 method, uri, date, payload, contentType, copySource, ifMatch, ifNoneMatch, body string
90 sigErr error
91}
92
93// stagingFixture is a staging mirror over leasetest's local S3, with every
94// request recorded and its signature verified on arrival. Optionally an
95// override answers instead of the model (a failing S3).
96func stagingFixture(t *testing.T, override func(w http.ResponseWriter, r *http.Request) bool) (*lease.StagingMirror, *leasetest.VersionedBucket, func() []recorded) {
97 t.Helper()
98 bucket := leasetest.NewVersionedBucket()
99 bucket.AddBucket("data-uk")
100 bucket.AddBucket("data-ie")
101 handler, _ := leasetest.NewS3Server(bucket)
102 var mu sync.Mutex
103 var reqs []recorded
104 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
105 body, _ := io.ReadAll(r.Body)
106 mu.Lock()
107 reqs = append(reqs, recorded{r.Method, r.RequestURI, r.Header.Get("X-Amz-Date"), r.Header.Get("X-Amz-Content-Sha256"),
108 r.Header.Get("Content-Type"), r.Header.Get("X-Amz-Copy-Source"), r.Header.Get("X-Amz-Copy-Source-If-Match"),
109 r.Header.Get("If-None-Match"), string(body), verifySigV4(r)})
110 mu.Unlock()
111 r.Body = io.NopCloser(strings.NewReader(string(body)))
112 if override != nil && override(w, r) {
113 return
114 }
115 handler.ServeHTTP(w, r)
116 }))
117 t.Cleanup(srv.Close)
118 return lease.NewStagingMirrorAt(srv.URL, stagingRegion, stagingCreds), bucket, func() []recorded {
119 mu.Lock()
120 defer mu.Unlock()
121 return append([]recorded(nil), reqs...)
122 }
123}
124
125var signAt = time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
126
127func TestStagingMirrorCopyVersioned(t *testing.T) {
128 m, bucket, reqs := stagingFixture(t, nil)
129 bucket.Put("data-ie", "instances/h/generations/0 1+a/lvl.ltx", []byte("body"))
130 etag := fmt.Sprintf("%x", md5sum("body"))
131 ctx := context.Background()
132 gotETag, ver, err := m.CopyVersionedAt(ctx, signAt, "data-ie", "instances/h/generations/0 1+a/lvl.ltx", etag, "data-uk", "moves/h/op/replica/x y.ltx")
133 if err != nil || gotETag != etag || ver == "" {
134 t.Fatalf("copy: etag %q ver %q err %v", gotETag, ver, err)
135 }
136 if got, _ := bucket.Current("data-uk", "moves/h/op/replica/x y.ltx"); string(got) != "body" {
137 t.Fatalf("destination %q", got)
138 }
139 r := reqs()[0]
140 if r.sigErr != nil {
141 t.Fatal(r.sigErr)
142 }
143 if r.date != "20261006T090000Z" || r.payload != "UNSIGNED-PAYLOAD" || r.method != "PUT" {
144 t.Fatalf("request %+v", r)
145 }
146 if r.copySource != "/data-ie/instances/h/generations/0%201%2Ba/lvl.ltx" || r.ifMatch != `"`+etag+`"` {
147 t.Fatalf("copy headers: source %q if-match %q", r.copySource, r.ifMatch)
148 }
149 if r.uri != "/data-uk/moves/h/op/replica/x%20y.ltx" {
150 t.Fatalf("destination path %q", r.uri)
151 }
152 // no source ETag: no precondition header
153 if _, _, err := m.CopyVersionedAt(ctx, signAt, "data-ie", "instances/h/generations/0 1+a/lvl.ltx", "", "data-uk", "moves/h/op/replica/z"); err != nil {
154 t.Fatal(err)
155 }
156 if rs := reqs(); rs[1].ifMatch != "" || rs[1].sigErr != nil {
157 t.Fatalf("unconditional copy: %+v", rs[1])
158 }
159}
160
161func md5sum(s string) [16]byte { return md5.Sum([]byte(s)) }
162
163func TestStagingMirrorCopyFailures(t *testing.T) {
164 m, bucket, reqs := stagingFixture(t, func(w http.ResponseWriter, r *http.Request) bool {
165 switch {
166 case strings.Contains(r.URL.Path, "late"):
167 w.WriteHeader(200)
168 io.WriteString(w, "<Error><Code>InternalError</Code><Message>x</Message></Error>")
169 return true
170 case strings.Contains(r.URL.Path, "boom"):
171 w.WriteHeader(503)
172 io.WriteString(w, "<Error><Code>SlowDown</Code><Message>x</Message></Error>")
173 return true
174 case strings.Contains(r.URL.Path, "empty"):
175 w.WriteHeader(200)
176 return true
177 }
178 return false
179 })
180 bucket.Put("data-ie", "k", []byte("body"))
181 ctx := context.Background()
182 // the source changed since it was listed
183 if _, _, err := m.CopyVersionedAt(ctx, signAt, "data-ie", "k", "0000", "data-uk", "d"); !errors.Is(err, lease.ErrCopySourceChanged) {
184 t.Fatalf("stale source: %v", err)
185 }
186 if _, ok := bucket.Current("data-uk", "d"); ok {
187 t.Fatal("a stale-source copy landed")
188 }
189 // a 200 carrying an <Error> body is a failure
190 if _, _, err := m.CopyVersionedAt(ctx, signAt, "data-ie", "k", "", "data-uk", "late"); err == nil || !strings.Contains(err.Error(), "InternalError") {
191 t.Fatalf("late failure: %v", err)
192 }
193 if _, _, err := m.CopyVersionedAt(ctx, signAt, "data-ie", "k", "", "data-uk", "empty"); err == nil {
194 t.Fatal("a 200 with no result is not a success")
195 }
196 // any other status names the code, and is sent once (no retry)
197 before := len(reqs())
198 _, _, err := m.CopyVersionedAt(ctx, signAt, "data-ie", "k", "", "data-uk", "boom")
199 if err == nil || !strings.Contains(err.Error(), "SlowDown") || errors.Is(err, lease.ErrCopySourceChanged) {
200 t.Fatalf("503: %v", err)
201 }
202 if n := len(reqs()) - before; n != 1 {
203 t.Fatalf("a failing request was sent %d times", n)
204 }
205}
206
207func TestStagingMirrorPutAndDelete(t *testing.T) {
208 m, bucket, reqs := stagingFixture(t, nil)
209 ctx := context.Background()
210 v1, err := m.PutVersionedAt(ctx, signAt, "data-uk", "moves/h/op/timers/t.json", []byte(`{"a":1}`))
211 if err != nil || v1 == "" {
212 t.Fatalf("put: %q %v", v1, err)
213 }
214 v2, err := m.PutVersionedAt(ctx, signAt, "data-uk", "moves/h/op/timers/t.json", []byte(`{"a":2}`))
215 if err != nil || v2 == "" || v2 == v1 {
216 t.Fatalf("overwrite: %q %v", v2, err)
217 }
218 sum := sha256.Sum256([]byte(`{"a":1}`))
219 r := reqs()[0]
220 if r.sigErr != nil || r.payload != hex.EncodeToString(sum[:]) || r.contentType != "application/json" || r.ifNoneMatch != "" || r.body != `{"a":1}` {
221 t.Fatalf("put request %+v", r)
222 }
223 if _, err := m.PutIfAbsentVersionedAt(ctx, signAt, "data-uk", "moves/h/op/move.json", []byte("{}")); err != nil {
224 t.Fatal(err)
225 }
226 if rs := reqs(); rs[2].ifNoneMatch != "*" || rs[2].sigErr != nil {
227 t.Fatalf("create-only request %+v", rs[2])
228 }
229 if _, err := m.PutIfAbsentVersionedAt(ctx, signAt, "data-uk", "moves/h/op/move.json", []byte("{}")); !errors.Is(err, lease.ErrExists) {
230 t.Fatalf("second create-only: %v", err)
231 }
232 // an empty body still signs and lands
233 if _, err := m.PutVersionedAt(ctx, signAt, "data-uk", "empty", nil); err != nil {
234 t.Fatal(err)
235 }
236 if err := m.DeleteAt(ctx, signAt, "data-uk", "moves/h/op/timers/t.json"); err != nil {
237 t.Fatal(err)
238 }
239 if _, ok := bucket.Current("data-uk", "moves/h/op/timers/t.json"); ok {
240 t.Fatal("still current after DeleteAt")
241 }
242 for i, r := range reqs() {
243 if r.sigErr != nil {
244 t.Fatalf("request %d: %v", i, r.sigErr)
245 }
246 }
247 // a missing bucket is an error naming the code
248 if _, err := m.PutVersionedAt(ctx, signAt, "nope", "k", []byte("x")); err == nil || !strings.Contains(err.Error(), "NoSuchBucket") {
249 t.Fatalf("missing bucket: %v", err)
250 }
251}
252
253// The signature carries the timestamp it was given and nothing else: the
254// same inputs sign identically however much later they are sent, and a
255// signAt in the far past is used as is (the mirror reads no clock).
256func TestStagingMirrorSignsOnlyAtSignAt(t *testing.T) {
257 var auths []string
258 m, _, reqs := stagingFixture(t, func(w http.ResponseWriter, r *http.Request) bool {
259 auths = append(auths, r.Header.Get("Authorization"))
260 return false
261 })
262 ctx := context.Background()
263 for i := 0; i < 2; i++ {
264 if _, err := m.PutVersionedAt(ctx, signAt, "data-uk", "k", []byte("x")); err != nil {
265 t.Fatal(err)
266 }
267 time.Sleep(1100 * time.Millisecond) // a second boundary passes between sends
268 }
269 if len(auths) != 2 || auths[0] != auths[1] {
270 t.Fatalf("same signAt, different signatures: %q", auths)
271 }
272 old := time.Date(2001, 1, 2, 3, 4, 5, 0, time.UTC)
273 if _, err := m.PutVersionedAt(ctx, old, "data-uk", "k", []byte("x")); err != nil {
274 t.Fatal(err)
275 }
276 if rs := reqs(); rs[2].date != "20010102T030405Z" {
277 t.Fatalf("date %q", rs[2].date)
278 }
279}
280
281// A mutation without a signing time (a caller that bypassed the guard)
282// sends nothing.
283func TestStagingMirrorRefusesAZeroSignTime(t *testing.T) {
284 m, _, reqs := stagingFixture(t, nil)
285 ctx := context.Background()
286 if _, err := m.PutVersionedAt(ctx, time.Time{}, "data-uk", "k", []byte("x")); err == nil {
287 t.Fatal("put signed at the zero time")
288 }
289 if _, _, err := m.CopyVersionedAt(ctx, time.Time{}, "data-ie", "k", "", "data-uk", "d"); err == nil {
290 t.Fatal("copy signed at the zero time")
291 }
292 if err := m.DeleteAt(ctx, time.Time{}, "data-uk", "k"); err == nil {
293 t.Fatal("delete signed at the zero time")
294 }
295 if n := len(reqs()); n != 0 {
296 t.Fatalf("%d requests sent", n)
297 }
298}
299
300// Get is the ordinary read, delegated.
301func TestStagingMirrorGetDelegates(t *testing.T) {
302 m, bucket, _ := stagingFixture(t, nil)
303 bucket.Put("data-uk", "r", []byte("receipts"))
304 got, err := m.Get(context.Background(), "data-uk", "r")
305 if err != nil || string(got) != "receipts" {
306 t.Fatalf("get: %q %v", got, err)
307 }
308 if _, err := m.Get(context.Background(), "data-uk", "absent"); !errors.Is(err, lease.ErrMirrorNotFound) {
309 t.Fatalf("get of an absent key: %v", err)
310 }
311}
312
313// A staging bucket must be versioned: a 2xx with no x-amz-version-id is a
314// failure for every mutation that is sealed by version, never an empty id.
315func TestStagingMirrorRefusesAMissingVersionID(t *testing.T) {
316 m, bucket, _ := stagingFixture(t, func(w http.ResponseWriter, r *http.Request) bool {
317 if !strings.Contains(r.URL.Path, "unversioned") {
318 return false
319 }
320 if r.Header.Get("X-Amz-Copy-Source") != "" {
321 io.WriteString(w, `<CopyObjectResult><ETag>"abc"</ETag></CopyObjectResult>`)
322 }
323 w.Header().Set("ETag", `"abc"`)
324 return true // a 200 with no x-amz-version-id
325 })
326 bucket.Put("data-ie", "k", []byte("body"))
327 ctx := context.Background()
328 for name, call := range map[string]func() error{
329 "copy": func() error {
330 _, _, err := m.CopyVersionedAt(ctx, signAt, "data-ie", "k", "", "data-uk", "unversioned/c")
331 return err
332 },
333 "put": func() error {
334 _, err := m.PutVersionedAt(ctx, signAt, "data-uk", "unversioned/p", []byte("x"))
335 return err
336 },
337 "put-if-absent": func() error {
338 _, err := m.PutIfAbsentVersionedAt(ctx, signAt, "data-uk", "unversioned/pia", []byte("x"))
339 return err
340 },
341 } {
342 if err := call(); err == nil || !strings.Contains(err.Error(), "version id") {
343 t.Errorf("%s with no version id: %v", name, err)
344 }
345 }
346}
347
348// Final review finding 4: a versioning-suspended bucket answers every write
349// with the version id "null", and the next write to the key overwrites that
350// same version. It names no immutable object, so it is refused like a
351// missing id, for every mutation that returns one.
352func TestStagingMirrorRefusesTheNullVersionID(t *testing.T) {
353 m, bucket, _ := stagingFixture(t, func(w http.ResponseWriter, r *http.Request) bool {
354 if !strings.Contains(r.URL.Path, "suspended") {
355 return false
356 }
357 // Headers first: a body written before them sends them.
358 w.Header().Set("ETag", `"abc"`)
359 w.Header().Set("X-Amz-Version-Id", "null")
360 if r.Header.Get("X-Amz-Copy-Source") != "" {
361 io.WriteString(w, `<CopyObjectResult><ETag>"abc"</ETag></CopyObjectResult>`)
362 }
363 return true
364 })
365 bucket.Put("data-ie", "k", []byte("body"))
366 ctx := context.Background()
367 body := "sealed"
368 sum := sha256.Sum256([]byte(body))
369 for name, call := range map[string]func() (string, error){
370 "copy": func() (string, error) {
371 _, v, err := m.CopyVersionedAt(ctx, signAt, "data-ie", "k", "", "data-uk", "suspended/c")
372 return v, err
373 },
374 "put": func() (string, error) {
375 return m.PutVersionedAt(ctx, signAt, "data-uk", "suspended/p", []byte("x"))
376 },
377 "put-reader": func() (string, error) {
378 return m.PutReaderVersionedAt(ctx, signAt, "data-uk", "suspended/r", strings.NewReader(body), int64(len(body)), hex.EncodeToString(sum[:]))
379 },
380 "put-if-absent": func() (string, error) {
381 return m.PutIfAbsentVersionedAt(ctx, signAt, "data-uk", "suspended/pia", []byte("x"))
382 },
383 } {
384 v, err := call()
385 if err == nil || v != "" || !strings.Contains(err.Error(), "version id") {
386 t.Errorf("%s answered with the null version: %q, %v", name, v, err)
387 }
388 }
389 for _, v := range []string{"", "null"} {
390 if lease.UsableVersionID(v) {
391 t.Errorf("UsableVersionID(%q) = true", v)
392 }
393 }
394 if !lease.UsableVersionID("3HL4kqtJlcpXroDTDmJ+rmSpXd3dIbrHY") {
395 t.Error("a real version id is not usable")
396 }
397}
398
399// A redirect is an error, and it is never followed: no second request,
400// whatever the status (a 307/308 would resend the mutation, a 301/302/303
401// would turn it into a GET).
402func TestStagingMirrorNeverFollowsARedirect(t *testing.T) {
403 for _, code := range []int{301, 302, 303, 307, 308} {
404 m, bucket, reqs := stagingFixture(t, func(w http.ResponseWriter, r *http.Request) bool {
405 if r.URL.Path == "/elsewhere" {
406 t.Errorf("a redirect was followed: %s %s", r.Method, r.URL.Path)
407 }
408 w.Header().Set("Location", "http://"+r.Host+"/elsewhere")
409 w.WriteHeader(code)
410 return true
411 })
412 bucket.Put("data-ie", "k", []byte("body"))
413 ctx := context.Background()
414 _, cerr := m.PutVersionedAt(ctx, signAt, "data-uk", "k", []byte("x"))
415 _, _, kerr := m.CopyVersionedAt(ctx, signAt, "data-ie", "k", "", "data-uk", "d")
416 _, perr := m.PutIfAbsentVersionedAt(ctx, signAt, "data-uk", "m", []byte("x"))
417 derr := m.DeleteAt(ctx, signAt, "data-uk", "k")
418 for name, err := range map[string]error{"put": cerr, "copy": kerr, "put-if-absent": perr, "delete": derr} {
419 if err == nil || !strings.Contains(err.Error(), fmt.Sprint(code)) {
420 t.Errorf("%d on %s: %v", code, name, err)
421 }
422 }
423 if n := len(reqs()); n != 4 {
424 t.Errorf("%d: %d requests for 4 calls", code, n)
425 }
426 }
427}
428
429// PutReaderVersionedAt streams a body of a known size and sha256 (the
430// sealed archive, kept on disk) under the same single signature: the
431// payload hash it signs is the one given, the body lands whole, and a
432// reader that ends early is never sent as a complete object.
433func TestStagingMirrorPutReader(t *testing.T) {
434 m, bucket, reqs := stagingFixture(t, nil)
435 ctx := context.Background()
436 body := strings.Repeat("sealed ", 1000)
437 sum := sha256.Sum256([]byte(body))
438 ver, err := m.PutReaderVersionedAt(ctx, signAt, "data-uk", "moves/h/op/files/files-x.sealed", strings.NewReader(body), int64(len(body)), hex.EncodeToString(sum[:]))
439 if err != nil || ver == "" {
440 t.Fatalf("put: %q %v", ver, err)
441 }
442 r := reqs()[0]
443 if r.sigErr != nil || r.payload != hex.EncodeToString(sum[:]) || r.body != body || r.ifNoneMatch != "" {
444 t.Fatalf("request %+v", r.sigErr)
445 }
446 if got, ok := bucket.Current("data-uk", "moves/h/op/files/files-x.sealed"); !ok || string(got) != body {
447 t.Fatalf("not stored: %q", got)
448 }
449 // A hash that is not 64 hex digits, a negative size, or a zero signing
450 // time sends nothing.
451 n := len(reqs())
452 for _, c := range []struct {
453 size int64
454 sha string
455 at time.Time
456 }{{int64(len(body)), "nothex", signAt}, {-1, hex.EncodeToString(sum[:]), signAt}, {int64(len(body)), hex.EncodeToString(sum[:]), time.Time{}}} {
457 if _, err := m.PutReaderVersionedAt(ctx, c.at, "data-uk", "k", strings.NewReader(body), c.size, c.sha); err == nil {
458 t.Fatalf("accepted size %d sha %q at %v", c.size, c.sha, c.at)
459 }
460 }
461 if len(reqs()) != n {
462 t.Fatal("a refused put sent a request")
463 }
464 // A reader shorter than its declared size fails rather than landing a
465 // truncated object.
466 if _, err := m.PutReaderVersionedAt(ctx, signAt, "data-uk", "short", strings.NewReader("abc"), 10, hex.EncodeToString(sum[:])); err == nil {
467 t.Fatal("a short body was accepted")
468 }
469 if _, ok := bucket.Current("data-uk", "short"); ok {
470 t.Fatal("a short body landed")
471 }
472}
473