amadan

carlos / platform Public

moves: staging mirror fails closed on a missing version id and never follows a redirect; the guard rejects a negative or overflowing skew (Task 6 fix round 1)

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
claude 336e14e0de43b684dcfbe3df1d66a177b4ae8123 parent 6070cdf
4 files changed, +141 −8
  • internal/lease/staging.go +30 −4
  • internal/lease/staging_test.go +65 −0
  • internal/moves/lease.go +20 −4
  • internal/moves/lease_test.go +26 −0
diff --git a/internal/lease/staging.go b/internal/lease/staging.go
index 5e0a691cb..d31e37cb9 100644
--- a/internal/lease/staging.go
+++ b/internal/lease/staging.go
@@ -85,6 +85,28 @@ type StagingMirror struct {
client *http.Client
}
+// newStagingClient never follows a redirect: a 307/308 would resend the
+// mutation and a 301/302/303 would turn it into a GET, and a mutation is
+// sent once. The 3xx answer comes back as the response and is an error.
+func newStagingClient() *http.Client {
+ return &http.Client{
+ Timeout: stagingRequestTimeout,
+ CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
+ }
+}
+
+// versionIDOf is the version id a 2xx answer carries. A staging bucket
+// must be versioned, since every artifact is sealed by version (§7.3), so
+// an answer without one fails closed: the write may have landed, and the
+// caller learns it cannot be sealed.
+func versionIDOf(op string, h http.Header) (string, error) {
+ v := h.Get("X-Amz-Version-Id")
+ if v == "" {
+ return "", fmt.Errorf("staging mirror: %s: no version id in the response (the staging bucket must be versioned)", op)
+ }
+ return v, nil
+}
+
func newStagingSigner() *v4.Signer {
// S3 canonicalizes the path as sent, never double-escaped.
return v4.NewSigner(func(o *v4.SignerOptions) { o.DisableURIPathEscaping = true })
@@ -116,7 +138,7 @@ func NewStagingMirror(ctx context.Context, srcRegion, dstRegion string) (*Stagin
region: regionFor(dstRegion, cfg.Region),
creds: cfg.Credentials,
signer: newStagingSigner(),
- client: &http.Client{Timeout: stagingRequestTimeout},
+ client: newStagingClient(),
}, nil
}
@@ -141,7 +163,7 @@ func NewStagingMirrorAt(endpoint, region string, creds aws.CredentialsProvider)
region: region,
creds: creds,
signer: newStagingSigner(),
- client: &http.Client{Timeout: stagingRequestTimeout},
+ client: newStagingClient(),
}
}
@@ -301,7 +323,11 @@ func (m *StagingMirror) CopyVersionedAt(ctx context.Context, signAt time.Time, s
if err := xml.Unmarshal(body, &res); err != nil || res.ETag == "" {
return "", "", fmt.Errorf("staging mirror: %s: no CopyObjectResult in a 200 response", op)
}
- return normalizeETag(res.ETag), h.Get("X-Amz-Version-Id"), nil
+ ver, err := versionIDOf(op, h)
+ if err != nil {
+ return "", "", err
+ }
+ return normalizeETag(res.ETag), ver, nil
}
func (m *StagingMirror) put(ctx context.Context, signAt time.Time, bucket, key string, body []byte, createOnly bool) (string, error) {
@@ -323,7 +349,7 @@ func (m *StagingMirror) put(ctx context.Context, signAt time.Time, bucket, key s
if status/100 != 2 {
return "", stagingFailure("put "+bucket+"/"+key, status, rb)
}
- return h.Get("X-Amz-Version-Id"), nil
+ return versionIDOf("put "+bucket+"/"+key, h)
}
func (m *StagingMirror) PutVersionedAt(ctx context.Context, signAt time.Time, bucket, key string, body []byte) (string, error) {
diff --git a/internal/lease/staging_test.go b/internal/lease/staging_test.go
index 4a100b89e..46204e45e 100644
--- a/internal/lease/staging_test.go
+++ b/internal/lease/staging_test.go
@@ -309,3 +309,68 @@ func TestStagingMirrorGetDelegates(t *testing.T) {
t.Fatalf("get of an absent key: %v", err)
}
}
+
+// A staging bucket must be versioned: a 2xx with no x-amz-version-id is a
+// failure for every mutation that is sealed by version, never an empty id.
+func TestStagingMirrorRefusesAMissingVersionID(t *testing.T) {
+ m, bucket, _ := stagingFixture(t, func(w http.ResponseWriter, r *http.Request) bool {
+ if !strings.Contains(r.URL.Path, "unversioned") {
+ return false
+ }
+ if r.Header.Get("X-Amz-Copy-Source") != "" {
+ io.WriteString(w, `<CopyObjectResult><ETag>"abc"</ETag></CopyObjectResult>`)
+ }
+ w.Header().Set("ETag", `"abc"`)
+ return true // a 200 with no x-amz-version-id
+ })
+ bucket.Put("data-ie", "k", []byte("body"))
+ ctx := context.Background()
+ for name, call := range map[string]func() error{
+ "copy": func() error {
+ _, _, err := m.CopyVersionedAt(ctx, signAt, "data-ie", "k", "", "data-uk", "unversioned/c")
+ return err
+ },
+ "put": func() error {
+ _, err := m.PutVersionedAt(ctx, signAt, "data-uk", "unversioned/p", []byte("x"))
+ return err
+ },
+ "put-if-absent": func() error {
+ _, err := m.PutIfAbsentVersionedAt(ctx, signAt, "data-uk", "unversioned/pia", []byte("x"))
+ return err
+ },
+ } {
+ if err := call(); err == nil || !strings.Contains(err.Error(), "version id") {
+ t.Errorf("%s with no version id: %v", name, err)
+ }
+ }
+}
+
+// A redirect is an error, and it is never followed: no second request,
+// whatever the status (a 307/308 would resend the mutation, a 301/302/303
+// would turn it into a GET).
+func TestStagingMirrorNeverFollowsARedirect(t *testing.T) {
+ for _, code := range []int{301, 302, 303, 307, 308} {
+ m, bucket, reqs := stagingFixture(t, func(w http.ResponseWriter, r *http.Request) bool {
+ if r.URL.Path == "/elsewhere" {
+ t.Errorf("a redirect was followed: %s %s", r.Method, r.URL.Path)
+ }
+ w.Header().Set("Location", "http://"+r.Host+"/elsewhere")
+ w.WriteHeader(code)
+ return true
+ })
+ bucket.Put("data-ie", "k", []byte("body"))
+ ctx := context.Background()
+ _, cerr := m.PutVersionedAt(ctx, signAt, "data-uk", "k", []byte("x"))
+ _, _, kerr := m.CopyVersionedAt(ctx, signAt, "data-ie", "k", "", "data-uk", "d")
+ _, perr := m.PutIfAbsentVersionedAt(ctx, signAt, "data-uk", "m", []byte("x"))
+ derr := m.DeleteAt(ctx, signAt, "data-uk", "k")
+ for name, err := range map[string]error{"put": cerr, "copy": kerr, "put-if-absent": perr, "delete": derr} {
+ if err == nil || !strings.Contains(err.Error(), fmt.Sprint(code)) {
+ t.Errorf("%d on %s: %v", code, name, err)
+ }
+ }
+ if n := len(reqs()); n != 4 {
+ t.Errorf("%d: %d requests for 4 calls", code, n)
+ }
+ }
+}
diff --git a/internal/moves/lease.go b/internal/moves/lease.go
index bc3211fd4..8eb0c08e3 100644
--- a/internal/moves/lease.go
+++ b/internal/moves/lease.go
@@ -3,6 +3,7 @@ package moves
import (
"errors"
"fmt"
+ "math"
"time"
)
@@ -32,6 +33,11 @@ func StagingLeasePrefix(targetRegion, host string) string {
return "control/staging-leases/" + targetRegion + "/" + host
}
+// ErrInvalidSkew is a negative Guarded.Skew: it would shrink the window
+// the guard exists to keep (a -15 minute skew admits a signature that
+// outlives the lease), so it is refused rather than honoured.
+var ErrInvalidSkew = errors.New("moves: staging guard skew must not be negative")
+
// SafeUntilGuard is the one method of *lease.Held the guard reads.
type SafeUntilGuard interface{ SafeUntil() (time.Time, error) }
@@ -54,16 +60,26 @@ type Guarded struct {
// and handed to fn as the signing time. lease.Held.SafeUntil already
// subtracts the renewal margin; Before subtracts the signature's life.
func (g Guarded) Before(fn func(signAt time.Time) error) error {
+ skew := g.Skew
+ if skew < 0 {
+ return ErrInvalidSkew
+ }
+ if skew == 0 {
+ skew = DefaultSkew
+ }
safe, err := g.Lease.SafeUntil()
if err != nil {
return fmt.Errorf("%w: %v", ErrLeaseUnsafe, err)
}
- skew := g.Skew
- if skew == 0 {
- skew = DefaultSkew
+ // A skew so large that SignatureLife+skew would overflow can never be
+ // satisfied: refuse rather than wrap into a pass.
+ if skew > math.MaxInt64-SignatureLife {
+ return ErrLeaseUnsafe
}
now := g.Now()
- if now.Add(SignatureLife + skew).After(safe) {
+ // safe.Sub saturates, so no sum of times can overflow; the comparison
+ // is the same as now+SignatureLife+skew being after safe.
+ if safe.Sub(now) < SignatureLife+skew {
return ErrLeaseUnsafe
}
return fn(now)
diff --git a/internal/moves/lease_test.go b/internal/moves/lease_test.go
index 179eaf6cc..351a664ab 100644
--- a/internal/moves/lease_test.go
+++ b/internal/moves/lease_test.go
@@ -3,6 +3,7 @@ package moves
import (
"context"
"errors"
+ "math"
"net/http"
"net/http/httptest"
"testing"
@@ -159,3 +160,28 @@ func TestStagingMirrorSignsAtTheCheck(t *testing.T) {
t.Fatalf("second create-only put: %v", err)
}
}
+
+// A negative skew would widen the window the guard is there to close (a
+// -15 minute skew admits a signature that outlives the lease), so it is a
+// configuration error and fn never runs. A skew so large that
+// SignatureLife+Skew overflows must refuse, not wrap into a pass.
+func TestGuardedSkewIsValidated(t *testing.T) {
+ t0 := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
+ plenty := fixedLease{t0.Add(24 * time.Hour)}
+ ran := false
+ fn := func(time.Time) error { ran = true; return nil }
+ neg := Guarded{Lease: fixedLease{t0.Add(time.Second)}, Now: func() time.Time { return t0 }, Skew: -15 * time.Minute}
+ if err := neg.Before(fn); !errors.Is(err, ErrInvalidSkew) || ran {
+ t.Fatalf("negative skew: err %v ran %v", err, ran)
+ }
+ neg.Lease = plenty
+ if err := neg.Before(fn); !errors.Is(err, ErrInvalidSkew) || ran {
+ t.Fatalf("negative skew with room: err %v ran %v", err, ran)
+ }
+ for _, skew := range []time.Duration{time.Duration(math.MaxInt64), time.Duration(math.MaxInt64) - SignatureLife + 1} {
+ huge := Guarded{Lease: plenty, Now: func() time.Time { return t0 }, Skew: skew}
+ if err := huge.Before(fn); !errors.Is(err, ErrLeaseUnsafe) || ran {
+ t.Fatalf("skew %v: err %v ran %v", skew, err, ran)
+ }
+ }
+}