| 1 | package lease |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "crypto/sha256" |
| 7 | "encoding/hex" |
| 8 | "encoding/xml" |
| 9 | "errors" |
| 10 | "fmt" |
| 11 | "io" |
| 12 | "net/http" |
| 13 | "net/url" |
| 14 | "strings" |
| 15 | "time" |
| 16 | |
| 17 | "github.com/aws/aws-sdk-go-v2/aws" |
| 18 | v4 "github.com/aws/aws-sdk-go-v2/aws/signer/v4" |
| 19 | awsconfig "github.com/aws/aws-sdk-go-v2/config" |
| 20 | awss3 "github.com/aws/aws-sdk-go-v2/service/s3" |
| 21 | "github.com/carlosframework/platform/internal/rootaws" |
| 22 | ) |
| 23 | |
| 24 | // The staging mirror (moves spec §7.3, model 14): the writer of a move's |
| 25 | // staging area, whose every mutation is signed at a timestamp the caller |
| 26 | // passes and sent exactly once. |
| 27 | // |
| 28 | // Why it signs by hand. A SigV4 signature stays valid for 15 minutes, and |
| 29 | // the SDK signs when its middleware runs, after any retry, backoff or |
| 30 | // pause the caller cannot see. The moves guard (moves.Guarded) proves the |
| 31 | // staging lease is safe for a signature's whole life at one instant and |
| 32 | // hands that instant to the mutation as signAt. This type signs AT signAt |
| 33 | // and reads no clock of its own, so a pause of any length between the |
| 34 | // check and the send (a frozen VM, a stalled goroutine) yields a request |
| 35 | // dated at the check, which S3 refuses once that date is 15 minutes old: |
| 36 | // exactly the life the guard required of the lease. There is no retry and |
| 37 | // no re-signing; a caller that wants another attempt goes back through |
| 38 | // the guard and gets a new timestamp. The one gap left is between the |
| 39 | // signing and the socket write, bounded by the request timeout and the |
| 40 | // guard's skew allowance (§7.3). |
| 41 | // |
| 42 | // *StagingMirror deliberately satisfies none of MirrorStore's mutations, |
| 43 | // VersionedCopier or VersionedPutter: each of those lacks a signAt, and |
| 44 | // only the guard hands one out, so no code path can sign outside a |
| 45 | // check. Callers hold it as a StagingWriter behind a guard adapter. |
| 46 | |
| 47 | // StagingWriter is what the move lane needs of its staging store: the |
| 48 | // four guarded mutations and the two reads. *StagingMirror is the |
| 49 | // production implementation; the lane's tests supply a fake. |
| 50 | type StagingWriter interface { |
| 51 | // CopyVersionedAt is a server-side copy into the staging bucket, |
| 52 | // conditional on srcETag when it is set (ErrCopySourceChanged when the |
| 53 | // source no longer carries it). It returns the destination's ETag |
| 54 | // (unquoted) and version id. |
| 55 | CopyVersionedAt(ctx context.Context, signAt time.Time, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (etag, versionID string, err error) |
| 56 | // PutVersionedAt writes body over key and returns the new version id. |
| 57 | PutVersionedAt(ctx context.Context, signAt time.Time, bucket, key string, body []byte) (versionID string, err error) |
| 58 | // PutReaderVersionedAt is PutVersionedAt streaming size bytes from r, |
| 59 | // signed with the sha256 the caller computed while writing them. |
| 60 | PutReaderVersionedAt(ctx context.Context, signAt time.Time, bucket, key string, r io.Reader, size int64, sha256Hex string) (versionID string, err error) |
| 61 | // PutIfAbsentVersionedAt is PutVersionedAt with If-None-Match: *; an |
| 62 | // existing object gives ErrExists. |
| 63 | PutIfAbsentVersionedAt(ctx context.Context, signAt time.Time, bucket, key string, body []byte) (versionID string, err error) |
| 64 | // DeleteAt removes key (on a versioned bucket, adds a delete marker). |
| 65 | DeleteAt(ctx context.Context, signAt time.Time, bucket, key string) error |
| 66 | List(ctx context.Context, bucket, prefix string) (map[string]MirrorObject, error) |
| 67 | Get(ctx context.Context, bucket, key string) ([]byte, error) |
| 68 | } |
| 69 | |
| 70 | var _ StagingWriter = (*StagingMirror)(nil) |
| 71 | |
| 72 | const ( |
| 73 | stagingRequestTimeout = 120 * time.Second |
| 74 | unsignedPayload = "UNSIGNED-PAYLOAD" |
| 75 | // stagingMaxResponse caps what is read of any response: the bodies |
| 76 | // that matter (a CopyObjectResult, an error document) are tiny. |
| 77 | stagingMaxResponse = 1 << 20 |
| 78 | ) |
| 79 | |
| 80 | // StagingMirror signs and sends staging mutations itself. See the package |
| 81 | // comment above. Build it with NewStagingMirror (production) or |
| 82 | // NewStagingMirrorAt (tests, an endpoint override). |
| 83 | type StagingMirror struct { |
| 84 | reads MirrorStore // List and Get: ordinary reads, no guard needed |
| 85 | endpoint string // base URL override (path-style); "" is AWS |
| 86 | region string // the destination bucket's region: where requests go and what they are signed for |
| 87 | creds aws.CredentialsProvider |
| 88 | signer *v4.Signer |
| 89 | client *http.Client |
| 90 | } |
| 91 | |
| 92 | // newStagingClient never follows a redirect: a 307/308 would resend the |
| 93 | // mutation and a 301/302/303 would turn it into a GET, and a mutation is |
| 94 | // sent once. The 3xx answer comes back as the response and is an error. |
| 95 | func newStagingClient() *http.Client { |
| 96 | return &http.Client{ |
| 97 | Timeout: stagingRequestTimeout, |
| 98 | CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }, |
| 99 | } |
| 100 | } |
| 101 | |
| 102 | // versionIDOf is the version id a 2xx answer carries. A staging bucket |
| 103 | // must be versioned, since every artifact is sealed by version (§7.3), so |
| 104 | // an answer without one, or with the suspended bucket's "null" |
| 105 | // (UsableVersionID), fails closed: the write may have landed, and the |
| 106 | // caller learns it cannot be sealed. |
| 107 | func versionIDOf(op string, h http.Header) (string, error) { |
| 108 | v := h.Get("X-Amz-Version-Id") |
| 109 | if !UsableVersionID(v) { |
| 110 | return "", fmt.Errorf("staging mirror: %s: no usable version id in the response (%q; the staging bucket must be versioned)", op, v) |
| 111 | } |
| 112 | return v, nil |
| 113 | } |
| 114 | |
| 115 | func newStagingSigner() *v4.Signer { |
| 116 | // S3 canonicalizes the path as sent, never double-escaped. |
| 117 | return v4.NewSigner(func(o *v4.SignerOptions) { o.DisableURIPathEscaping = true }) |
| 118 | } |
| 119 | |
| 120 | // NewStagingMirror builds the production mirror: reads through the same |
| 121 | // two-region pair NewS3Mirror uses, mutations signed with the default |
| 122 | // credential chain for dstRegion (the staging bucket's region). |
| 123 | // CARLOS_S3_ENDPOINT, when set, replaces the AWS endpoint, path-style. |
| 124 | func NewStagingMirror(ctx context.Context, srcRegion, dstRegion string) (*StagingMirror, error) { |
| 125 | reads, err := NewS3Mirror(ctx, srcRegion, dstRegion) |
| 126 | if err != nil { |
| 127 | return nil, err |
| 128 | } |
| 129 | var opts []func(*awsconfig.LoadOptions) error |
| 130 | if dstRegion != "" { |
| 131 | opts = append(opts, awsconfig.WithRegion(dstRegion)) |
| 132 | } |
| 133 | cfg, err := rootaws.LoadDefaultConfig(ctx, opts...) |
| 134 | if err != nil { |
| 135 | return nil, fmt.Errorf("load aws config: %w", err) |
| 136 | } |
| 137 | if cfg.Credentials == nil { |
| 138 | return nil, errors.New("staging mirror: no AWS credentials to sign with") |
| 139 | } |
| 140 | return &StagingMirror{ |
| 141 | reads: reads, |
| 142 | endpoint: s3Endpoint(), |
| 143 | region: regionFor(dstRegion, cfg.Region), |
| 144 | creds: cfg.Credentials, |
| 145 | signer: newStagingSigner(), |
| 146 | client: newStagingClient(), |
| 147 | }, nil |
| 148 | } |
| 149 | |
| 150 | // NewStagingMirrorAt is the constructor with the endpoint and credentials |
| 151 | // given, path-style, for tests against a local S3 (leasetest.NewS3Server). |
| 152 | // Reads go to the same endpoint. |
| 153 | func NewStagingMirrorAt(endpoint, region string, creds aws.CredentialsProvider) *StagingMirror { |
| 154 | if !strings.Contains(endpoint, "://") { |
| 155 | endpoint = "https://" + endpoint |
| 156 | } |
| 157 | c := awss3.New(awss3.Options{ |
| 158 | Region: region, |
| 159 | Credentials: creds, |
| 160 | BaseEndpoint: &endpoint, |
| 161 | UsePathStyle: true, |
| 162 | RequestChecksumCalculation: aws.RequestChecksumCalculationWhenRequired, |
| 163 | ResponseChecksumValidation: aws.ResponseChecksumValidationWhenRequired, |
| 164 | }) |
| 165 | return &StagingMirror{ |
| 166 | reads: &s3Mirror{src: c, dst: c, byBucket: map[string]*awss3.Client{}}, |
| 167 | endpoint: endpoint, |
| 168 | region: region, |
| 169 | creds: creds, |
| 170 | signer: newStagingSigner(), |
| 171 | client: newStagingClient(), |
| 172 | } |
| 173 | } |
| 174 | |
| 175 | func (m *StagingMirror) List(ctx context.Context, bucket, prefix string) (map[string]MirrorObject, error) { |
| 176 | return m.reads.List(ctx, bucket, prefix) |
| 177 | } |
| 178 | |
| 179 | func (m *StagingMirror) Get(ctx context.Context, bucket, key string) ([]byte, error) { |
| 180 | return m.reads.Get(ctx, bucket, key) |
| 181 | } |
| 182 | |
| 183 | // s3Escape percent-encodes everything but the RFC 3986 unreserved set and |
| 184 | // "/", which is what S3 canonicalizes object paths (and the copy-source |
| 185 | // header) to. |
| 186 | func s3Escape(s string) string { |
| 187 | const hexdigits = "0123456789ABCDEF" |
| 188 | var b strings.Builder |
| 189 | for i := 0; i < len(s); i++ { |
| 190 | c := s[i] |
| 191 | switch { |
| 192 | case c >= 'a' && c <= 'z', c >= 'A' && c <= 'Z', c >= '0' && c <= '9', |
| 193 | c == '-', c == '_', c == '.', c == '~', c == '/': |
| 194 | b.WriteByte(c) |
| 195 | default: |
| 196 | b.WriteByte('%') |
| 197 | b.WriteByte(hexdigits[c>>4]) |
| 198 | b.WriteByte(hexdigits[c&15]) |
| 199 | } |
| 200 | } |
| 201 | return b.String() |
| 202 | } |
| 203 | |
| 204 | // objectURL addresses bucket/key: the override path-style, else AWS |
| 205 | // virtual-hosted (path-style for a dotted bucket name, whose name the |
| 206 | // wildcard certificate does not cover). |
| 207 | func (m *StagingMirror) objectURL(bucket, key string) (*url.URL, error) { |
| 208 | var u *url.URL |
| 209 | var path string |
| 210 | switch { |
| 211 | case m.endpoint != "": |
| 212 | base, err := url.Parse(m.endpoint) |
| 213 | if err != nil { |
| 214 | return nil, fmt.Errorf("staging mirror: endpoint %q: %w", m.endpoint, err) |
| 215 | } |
| 216 | u, path = base, strings.TrimSuffix(base.Path, "/")+"/"+bucket+"/"+key |
| 217 | case strings.Contains(bucket, "."): |
| 218 | u, path = &url.URL{Scheme: "https", Host: "s3." + m.region + ".amazonaws.com"}, "/"+bucket+"/"+key |
| 219 | default: |
| 220 | u, path = &url.URL{Scheme: "https", Host: bucket + ".s3." + m.region + ".amazonaws.com"}, "/"+key |
| 221 | } |
| 222 | raw := s3Escape(path) |
| 223 | dec, err := url.PathUnescape(raw) |
| 224 | if err != nil { |
| 225 | return nil, err |
| 226 | } |
| 227 | out := *u |
| 228 | out.Path, out.RawPath, out.RawQuery = dec, raw, "" |
| 229 | return &out, nil |
| 230 | } |
| 231 | |
| 232 | // send signs one request at signAt and sends it once. Nothing here reads a |
| 233 | // clock, retries, or re-signs. It returns the status, headers and (capped) |
| 234 | // body of whatever S3 answered. |
| 235 | func (m *StagingMirror) send(ctx context.Context, signAt time.Time, method, bucket, key string, hdr http.Header, body []byte) (int, http.Header, []byte, error) { |
| 236 | if body == nil { |
| 237 | return m.sendBody(ctx, signAt, method, bucket, key, hdr, nil, 0, unsignedPayload) |
| 238 | } |
| 239 | sum := sha256.Sum256(body) |
| 240 | return m.sendBody(ctx, signAt, method, bucket, key, hdr, bytes.NewReader(body), int64(len(body)), hex.EncodeToString(sum[:])) |
| 241 | } |
| 242 | |
| 243 | // sendBody is send with the body as a reader of a declared size whose |
| 244 | // sha256 (hex) the caller already knows: it is the signed payload hash, so |
| 245 | // S3 refuses a body that does not match it, and a reader that ends short of |
| 246 | // size fails the request rather than landing a truncated object. |
| 247 | func (m *StagingMirror) sendBody(ctx context.Context, signAt time.Time, method, bucket, key string, hdr http.Header, rd io.Reader, size int64, payloadHash string) (int, http.Header, []byte, error) { |
| 248 | if signAt.IsZero() { |
| 249 | return 0, nil, nil, errors.New("staging mirror: no signing time (a mutation signs at a Guarded.Before check)") |
| 250 | } |
| 251 | u, err := m.objectURL(bucket, key) |
| 252 | if err != nil { |
| 253 | return 0, nil, nil, err |
| 254 | } |
| 255 | req, err := http.NewRequestWithContext(ctx, method, u.String(), rd) |
| 256 | if err != nil { |
| 257 | return 0, nil, nil, err |
| 258 | } |
| 259 | if rd != nil { |
| 260 | req.ContentLength = size |
| 261 | } |
| 262 | for k, vs := range hdr { |
| 263 | req.Header[k] = vs |
| 264 | } |
| 265 | req.Header.Set("X-Amz-Content-Sha256", payloadHash) |
| 266 | creds, err := m.creds.Retrieve(ctx) |
| 267 | if err != nil { |
| 268 | return 0, nil, nil, fmt.Errorf("staging mirror: credentials: %w", err) |
| 269 | } |
| 270 | if err := m.signer.SignHTTP(ctx, creds, req, payloadHash, "s3", m.region, signAt); err != nil { |
| 271 | return 0, nil, nil, fmt.Errorf("staging mirror: sign: %w", err) |
| 272 | } |
| 273 | resp, err := m.client.Do(req) |
| 274 | if err != nil { |
| 275 | return 0, nil, nil, fmt.Errorf("staging mirror: %s %s/%s: %w", method, bucket, key, err) |
| 276 | } |
| 277 | defer resp.Body.Close() |
| 278 | rb, err := io.ReadAll(io.LimitReader(resp.Body, stagingMaxResponse)) |
| 279 | if err != nil { |
| 280 | return resp.StatusCode, resp.Header, nil, fmt.Errorf("staging mirror: %s %s/%s: read response: %w", method, bucket, key, err) |
| 281 | } |
| 282 | return resp.StatusCode, resp.Header, rb, nil |
| 283 | } |
| 284 | |
| 285 | // s3ErrorBody is the document S3 puts in an error response, and in the |
| 286 | // body of a 200 copy that failed after the headers went out. |
| 287 | type s3ErrorBody struct { |
| 288 | XMLName xml.Name `xml:"Error"` |
| 289 | Code string `xml:"Code"` |
| 290 | Message string `xml:"Message"` |
| 291 | } |
| 292 | |
| 293 | // s3Code is the error document's Code, or "" if the body is not one. |
| 294 | func s3Code(body []byte) string { |
| 295 | var e s3ErrorBody |
| 296 | if xml.Unmarshal(body, &e) != nil { |
| 297 | return "" |
| 298 | } |
| 299 | return e.Code |
| 300 | } |
| 301 | |
| 302 | func stagingFailure(op string, status int, body []byte) error { |
| 303 | code := s3Code(body) |
| 304 | if code == "" { |
| 305 | code = http.StatusText(status) |
| 306 | } |
| 307 | return fmt.Errorf("staging mirror: %s: %s (HTTP %d)", op, code, status) |
| 308 | } |
| 309 | |
| 310 | func (m *StagingMirror) CopyVersionedAt(ctx context.Context, signAt time.Time, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (string, string, error) { |
| 311 | hdr := http.Header{} |
| 312 | hdr.Set("X-Amz-Copy-Source", "/"+srcBucket+"/"+s3Escape(srcKey)) |
| 313 | if srcETag != "" { |
| 314 | hdr.Set("X-Amz-Copy-Source-If-Match", `"`+normalizeETag(srcETag)+`"`) |
| 315 | } |
| 316 | op := "copy " + srcBucket + "/" + srcKey |
| 317 | status, h, body, err := m.send(ctx, signAt, http.MethodPut, dstBucket, dstKey, hdr, nil) |
| 318 | if err != nil { |
| 319 | return "", "", err |
| 320 | } |
| 321 | if status == http.StatusPreconditionFailed && srcETag != "" && s3Code(body) == "PreconditionFailed" { |
| 322 | return "", "", ErrCopySourceChanged |
| 323 | } |
| 324 | if status/100 != 2 { |
| 325 | return "", "", stagingFailure(op, status, body) |
| 326 | } |
| 327 | // S3 may answer 200 and then fail the copy, putting an <Error> document |
| 328 | // where the result belongs. |
| 329 | if code := s3Code(body); code != "" { |
| 330 | return "", "", fmt.Errorf("staging mirror: %s: %s (HTTP %d, late failure)", op, code, status) |
| 331 | } |
| 332 | var res struct { |
| 333 | XMLName xml.Name `xml:"CopyObjectResult"` |
| 334 | ETag string `xml:"ETag"` |
| 335 | } |
| 336 | if err := xml.Unmarshal(body, &res); err != nil || res.ETag == "" { |
| 337 | return "", "", fmt.Errorf("staging mirror: %s: no CopyObjectResult in a 200 response", op) |
| 338 | } |
| 339 | ver, err := versionIDOf(op, h) |
| 340 | if err != nil { |
| 341 | return "", "", err |
| 342 | } |
| 343 | return normalizeETag(res.ETag), ver, nil |
| 344 | } |
| 345 | |
| 346 | func (m *StagingMirror) put(ctx context.Context, signAt time.Time, bucket, key string, body []byte, createOnly bool) (string, error) { |
| 347 | if body == nil { |
| 348 | body = []byte{} |
| 349 | } |
| 350 | hdr := http.Header{} |
| 351 | hdr.Set("Content-Type", "application/json") |
| 352 | if createOnly { |
| 353 | hdr.Set("If-None-Match", "*") |
| 354 | } |
| 355 | status, h, rb, err := m.send(ctx, signAt, http.MethodPut, bucket, key, hdr, body) |
| 356 | if err != nil { |
| 357 | return "", err |
| 358 | } |
| 359 | if createOnly && status == http.StatusPreconditionFailed { |
| 360 | return "", ErrExists |
| 361 | } |
| 362 | if status/100 != 2 { |
| 363 | return "", stagingFailure("put "+bucket+"/"+key, status, rb) |
| 364 | } |
| 365 | return versionIDOf("put "+bucket+"/"+key, h) |
| 366 | } |
| 367 | |
| 368 | func (m *StagingMirror) PutVersionedAt(ctx context.Context, signAt time.Time, bucket, key string, body []byte) (string, error) { |
| 369 | return m.put(ctx, signAt, bucket, key, body, false) |
| 370 | } |
| 371 | |
| 372 | // PutReaderVersionedAt is PutVersionedAt for a body read from r: size bytes |
| 373 | // whose sha256 is sha256Hex. The archive is staged this way from its file, |
| 374 | // so a 256 MiB archive is never held in memory. |
| 375 | func (m *StagingMirror) PutReaderVersionedAt(ctx context.Context, signAt time.Time, bucket, key string, r io.Reader, size int64, sha256Hex string) (string, error) { |
| 376 | if size < 0 { |
| 377 | return "", fmt.Errorf("staging mirror: put %s/%s: negative size", bucket, key) |
| 378 | } |
| 379 | if b, err := hex.DecodeString(sha256Hex); err != nil || len(b) != sha256.Size { |
| 380 | return "", fmt.Errorf("staging mirror: put %s/%s: not a sha256", bucket, key) |
| 381 | } |
| 382 | hdr := http.Header{} |
| 383 | hdr.Set("Content-Type", "application/octet-stream") |
| 384 | status, h, rb, err := m.sendBody(ctx, signAt, http.MethodPut, bucket, key, hdr, io.LimitReader(r, size), size, sha256Hex) |
| 385 | if err != nil { |
| 386 | return "", err |
| 387 | } |
| 388 | if status/100 != 2 { |
| 389 | return "", stagingFailure("put "+bucket+"/"+key, status, rb) |
| 390 | } |
| 391 | return versionIDOf("put "+bucket+"/"+key, h) |
| 392 | } |
| 393 | |
| 394 | func (m *StagingMirror) PutIfAbsentVersionedAt(ctx context.Context, signAt time.Time, bucket, key string, body []byte) (string, error) { |
| 395 | return m.put(ctx, signAt, bucket, key, body, true) |
| 396 | } |
| 397 | |
| 398 | func (m *StagingMirror) DeleteAt(ctx context.Context, signAt time.Time, bucket, key string) error { |
| 399 | status, _, body, err := m.send(ctx, signAt, http.MethodDelete, bucket, key, nil, nil) |
| 400 | if err != nil { |
| 401 | return err |
| 402 | } |
| 403 | if status/100 != 2 { |
| 404 | return stagingFailure("delete "+bucket+"/"+key, status, body) |
| 405 | } |
| 406 | return nil |
| 407 | } |
| 408 | |
| 409 | // UsableVersionID reports whether v names one immutable object version. "" |
| 410 | // is no version at all. "null" is what a versioning-suspended (or never |
| 411 | // versioned) bucket answers: the next write to the key overwrites that very |
| 412 | // version, so it pins nothing and a seal by it is no seal (final review |
| 413 | // finding 4; AWS, "Adding objects to versioning-suspended buckets"). |
| 414 | func UsableVersionID(v string) bool { return v != "" && v != "null" } |
| 415 | |
amadan