| 1 | package leasetest |
| 2 | |
| 3 | import ( |
| 4 | "crypto/sha256" |
| 5 | "encoding/base64" |
| 6 | "encoding/hex" |
| 7 | "encoding/xml" |
| 8 | "errors" |
| 9 | "fmt" |
| 10 | "io" |
| 11 | "net/http" |
| 12 | "net/url" |
| 13 | "sort" |
| 14 | "strconv" |
| 15 | "strings" |
| 16 | "sync/atomic" |
| 17 | "time" |
| 18 | |
| 19 | "github.com/carlosframework/platform/internal/lease" |
| 20 | ) |
| 21 | |
| 22 | type xmlVersion struct { |
| 23 | XMLName xml.Name `xml:"Version"` |
| 24 | Key string |
| 25 | VersionId string |
| 26 | IsLatest bool |
| 27 | LastModified string |
| 28 | ETag string |
| 29 | Size int64 |
| 30 | } |
| 31 | |
| 32 | type xmlMarker struct { |
| 33 | XMLName xml.Name `xml:"DeleteMarker"` |
| 34 | Key string |
| 35 | VersionId string |
| 36 | IsLatest bool |
| 37 | LastModified string |
| 38 | } |
| 39 | |
| 40 | type listVersionsResult struct { |
| 41 | XMLName xml.Name `xml:"ListVersionsResult"` |
| 42 | Name string |
| 43 | Prefix string |
| 44 | KeyMarker string |
| 45 | VersionIdMarker string |
| 46 | NextKeyMarker string `xml:",omitempty"` |
| 47 | NextVersionIdMarker string `xml:",omitempty"` |
| 48 | MaxKeys int |
| 49 | IsTruncated bool |
| 50 | Entries []any |
| 51 | } |
| 52 | |
| 53 | func s3err(w http.ResponseWriter, status int, code string) { s3errMsg(w, status, code, code) } |
| 54 | |
| 55 | func s3errMsg(w http.ResponseWriter, status int, code, msg string) { |
| 56 | w.Header().Set("Content-Type", "application/xml") |
| 57 | w.WriteHeader(status) |
| 58 | io.WriteString(w, "<Error><Code>"+code+"</Code><Message>"+msg+"</Message></Error>") |
| 59 | } |
| 60 | |
| 61 | // readValidatedBody reads a PUT's body and checks it as real S3 does: its |
| 62 | // length must equal the declared Content-Length (when one is declared), and, |
| 63 | // when x-amz-content-sha256 is a hex digest, its sha256 must equal it. Any |
| 64 | // other value of that header (UNSIGNED-PAYLOAD, a streaming marker) or none |
| 65 | // is not checked. On a mismatch it answers 400 with S3's code and reports |
| 66 | // false, and the caller stores nothing. |
| 67 | func readValidatedBody(w http.ResponseWriter, r *http.Request) ([]byte, bool) { |
| 68 | body, err := io.ReadAll(r.Body) |
| 69 | if err != nil || (r.ContentLength >= 0 && int64(len(body)) != r.ContentLength) { |
| 70 | s3errMsg(w, http.StatusBadRequest, "IncompleteBody", "You did not provide the number of bytes specified by the Content-Length HTTP header.") |
| 71 | return nil, false |
| 72 | } |
| 73 | if want := r.Header.Get("x-amz-content-sha256"); isSHA256Hex(want) { |
| 74 | sum := sha256.Sum256(body) |
| 75 | if !strings.EqualFold(want, hex.EncodeToString(sum[:])) { |
| 76 | s3errMsg(w, http.StatusBadRequest, "XAmzContentSHA256Mismatch", "The provided 'x-amz-content-sha256' header does not match what was computed.") |
| 77 | return nil, false |
| 78 | } |
| 79 | } |
| 80 | return body, true |
| 81 | } |
| 82 | |
| 83 | func isSHA256Hex(s string) bool { |
| 84 | if len(s) != 2*sha256.Size { |
| 85 | return false |
| 86 | } |
| 87 | _, err := hex.DecodeString(s) |
| 88 | return err == nil |
| 89 | } |
| 90 | |
| 91 | // copyMarkerMsg is S3's answer to a COPY naming a delete marker's version. |
| 92 | const copyMarkerMsg = "The source of a copy request may not specifically refer to a delete marker by version id." |
| 93 | |
| 94 | // writeErr maps a model error onto the S3 response real S3 gives. |
| 95 | func writeErr(w http.ResponseWriter, err error) { |
| 96 | switch { |
| 97 | case errors.Is(err, lease.ErrMirrorNotFound): |
| 98 | s3err(w, http.StatusNotFound, "NoSuchVersion") |
| 99 | case errors.Is(err, lease.ErrIsDeleteMarker): |
| 100 | w.Header().Set("x-amz-delete-marker", "true") |
| 101 | s3err(w, http.StatusMethodNotAllowed, "MethodNotAllowed") |
| 102 | case errors.Is(err, lease.ErrDestinationExists), errors.Is(err, lease.ErrExists): |
| 103 | s3err(w, http.StatusPreconditionFailed, "PreconditionFailed") |
| 104 | default: |
| 105 | s3err(w, http.StatusInternalServerError, "InternalError") |
| 106 | } |
| 107 | } |
| 108 | |
| 109 | // NewS3Server speaks the subset of the S3 wire protocol VersionedStore uses, |
| 110 | // and ListObjectsV2 (serveListV2), path-style, over f, holding f.mu for each request. The second result |
| 111 | // counts requests served. With serves non-empty, the endpoint answers only |
| 112 | // for those request buckets (a regional endpoint); any other bucket gets |
| 113 | // 301 PermanentRedirect, as S3 answers a wrong-region request. A copy's |
| 114 | // source bucket may be any bucket in f. |
| 115 | func NewS3Server(f *VersionedBucket, serves ...string) (http.Handler, func() int) { |
| 116 | var n atomic.Int64 |
| 117 | allowed := map[string]bool{} |
| 118 | for _, b := range serves { |
| 119 | allowed[b] = true |
| 120 | } |
| 121 | h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 122 | n.Add(1) |
| 123 | bucket, rawKey, _ := strings.Cut(strings.TrimPrefix(r.URL.EscapedPath(), "/"), "/") |
| 124 | key, err := url.PathUnescape(rawKey) |
| 125 | if err != nil { |
| 126 | s3err(w, http.StatusBadRequest, "InvalidURI") |
| 127 | return |
| 128 | } |
| 129 | q := r.URL.Query() |
| 130 | if len(allowed) > 0 && !allowed[bucket] { |
| 131 | s3err(w, http.StatusMovedPermanently, "PermanentRedirect") |
| 132 | return |
| 133 | } |
| 134 | var body []byte |
| 135 | if r.Method == http.MethodPut { // every PUT, a copy included, is validated before it is applied |
| 136 | var ok bool |
| 137 | if body, ok = readValidatedBody(w, r); !ok { |
| 138 | return |
| 139 | } |
| 140 | } |
| 141 | f.mu.Lock() |
| 142 | defer f.mu.Unlock() |
| 143 | if err := f.known(bucket); err != nil { |
| 144 | s3err(w, http.StatusNotFound, "NoSuchBucket") |
| 145 | return |
| 146 | } |
| 147 | switch { |
| 148 | case r.Method == http.MethodGet && key == "" && q.Has("versions"): |
| 149 | f.serveList(w, bucket, q) |
| 150 | case r.Method == http.MethodGet && key == "" && q.Get("list-type") == "2": |
| 151 | f.serveListV2(w, bucket, q) |
| 152 | case r.Method == http.MethodGet && key != "": |
| 153 | versionID := q.Get("versionId") |
| 154 | if versionID == "" { // no version named: the current one |
| 155 | cur := f.buckets[bucket][key] |
| 156 | if len(cur) == 0 || cur[0].deleteMarker { |
| 157 | s3err(w, http.StatusNotFound, "NoSuchKey") |
| 158 | return |
| 159 | } |
| 160 | versionID = cur[0].id |
| 161 | } |
| 162 | body, err := f.read(bucket, key, versionID) |
| 163 | if err != nil { |
| 164 | writeErr(w, err) |
| 165 | return |
| 166 | } |
| 167 | if v, ok := f.find(bucket, key, versionID); ok { |
| 168 | w.Header().Set("ETag", `"`+v.etag+`"`) |
| 169 | } |
| 170 | if hdr := r.Header.Get("Range"); hdr != "" { |
| 171 | span, ok := strings.CutPrefix(hdr, "bytes=") |
| 172 | start, end, dash := strings.Cut(span, "-") |
| 173 | offset, startErr := strconv.ParseInt(start, 10, 64) |
| 174 | length := int64(-1) |
| 175 | if end != "" { |
| 176 | last, endErr := strconv.ParseInt(end, 10, 64) |
| 177 | if endErr != nil || last < offset || last-offset == 1<<63-1 { |
| 178 | s3err(w, http.StatusRequestedRangeNotSatisfiable, "InvalidRange") |
| 179 | return |
| 180 | } |
| 181 | length = last - offset + 1 |
| 182 | } |
| 183 | total := len(body) |
| 184 | body, err = versionRange(body, offset, length) |
| 185 | if !ok || !dash || startErr != nil || err != nil { |
| 186 | s3err(w, http.StatusRequestedRangeNotSatisfiable, "InvalidRange") |
| 187 | return |
| 188 | } |
| 189 | w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", offset, offset+int64(len(body))-1, total)) |
| 190 | w.Header().Set("Content-Length", strconv.Itoa(len(body))) |
| 191 | w.WriteHeader(http.StatusPartialContent) |
| 192 | } |
| 193 | w.Write(body) |
| 194 | case r.Method == http.MethodPut && r.Header.Get("x-amz-copy-source") != "": |
| 195 | f.serveCopy(w, r, bucket, key) |
| 196 | case r.Method == http.MethodPut: |
| 197 | if match := r.Header.Get("If-Match"); match != "" { |
| 198 | current := f.buckets[bucket][key] |
| 199 | if len(current) == 0 || current[0].deleteMarker || current[0].etag != strings.Trim(match, `"`) { |
| 200 | s3err(w, http.StatusPreconditionFailed, "PreconditionFailed") |
| 201 | return |
| 202 | } |
| 203 | } |
| 204 | var id string |
| 205 | var err error |
| 206 | if r.Header.Get("If-None-Match") == "*" { |
| 207 | id, err = f.putIfAbsentLocked(bucket, key, body) |
| 208 | } else { |
| 209 | id, err = f.putLocked(bucket, key, body) // a new version, as S3 with versioning on |
| 210 | } |
| 211 | if err != nil { |
| 212 | writeErr(w, err) |
| 213 | return |
| 214 | } |
| 215 | w.Header().Set("x-amz-version-id", id) |
| 216 | w.Header().Set("ETag", `"`+md5hex(body)+`"`) |
| 217 | case r.Method == http.MethodDelete && key != "" && q.Get("versionId") != "": |
| 218 | if err := f.deleteLocked(bucket, key, q.Get("versionId")); err != nil { |
| 219 | writeErr(w, err) |
| 220 | return |
| 221 | } |
| 222 | w.WriteHeader(http.StatusNoContent) |
| 223 | case r.Method == http.MethodDelete && key != "": |
| 224 | if match := r.Header.Get("If-Match"); match != "" { // as S3 enforces it on DeleteObject |
| 225 | current := f.buckets[bucket][key] |
| 226 | if len(current) == 0 || current[0].deleteMarker { |
| 227 | s3err(w, http.StatusNotFound, "NoSuchKey") |
| 228 | return |
| 229 | } |
| 230 | if current[0].etag != strings.Trim(match, `"`) { |
| 231 | s3err(w, http.StatusPreconditionFailed, "PreconditionFailed") |
| 232 | return |
| 233 | } |
| 234 | } |
| 235 | id, err := f.bareDeleteLocked(bucket, key) // versioning on: a delete marker |
| 236 | if err != nil { |
| 237 | writeErr(w, err) |
| 238 | return |
| 239 | } |
| 240 | w.Header().Set("x-amz-delete-marker", "true") |
| 241 | w.Header().Set("x-amz-version-id", id) |
| 242 | w.WriteHeader(http.StatusNoContent) |
| 243 | default: |
| 244 | s3err(w, http.StatusMethodNotAllowed, "MethodNotAllowed") |
| 245 | } |
| 246 | }) |
| 247 | return h, func() int { return int(n.Load()) } |
| 248 | } |
| 249 | |
| 250 | func (f *VersionedBucket) serveList(w http.ResponseWriter, bucket string, q url.Values) { |
| 251 | if err := f.fail("list", q.Get("prefix")); err != nil { |
| 252 | writeErr(w, err) |
| 253 | return |
| 254 | } |
| 255 | all := f.listAll(bucket, q.Get("prefix")) |
| 256 | km, vm := q.Get("key-marker"), q.Get("version-id-marker") |
| 257 | start := 0 |
| 258 | if km != "" { |
| 259 | start = len(all) // a marker not found resumes at the end |
| 260 | for i, v := range all { |
| 261 | if v.Key == km && v.VersionID == vm { |
| 262 | start = i + 1 |
| 263 | break |
| 264 | } |
| 265 | } |
| 266 | } |
| 267 | size := f.PageSize |
| 268 | if size <= 0 { |
| 269 | size = 1000 |
| 270 | } |
| 271 | page := (start + size - 1) / size // pages already served before this one |
| 272 | vs, more, err := f.pageOf(all[start:], 0) |
| 273 | if f.FailListAfterPages > 0 && page >= f.FailListAfterPages { |
| 274 | err = errors.New("leasetest: injected listing failure") |
| 275 | } |
| 276 | if err != nil { |
| 277 | writeErr(w, err) |
| 278 | return |
| 279 | } |
| 280 | res := listVersionsResult{Name: bucket, Prefix: q.Get("prefix"), KeyMarker: km, VersionIdMarker: vm, MaxKeys: size, IsTruncated: more} |
| 281 | for _, v := range vs { |
| 282 | lm := v.LastModified.UTC().Format(time.RFC3339) |
| 283 | if v.DeleteMarker { |
| 284 | res.Entries = append(res.Entries, xmlMarker{Key: v.Key, VersionId: v.VersionID, IsLatest: v.IsLatest, LastModified: lm}) |
| 285 | } else { |
| 286 | res.Entries = append(res.Entries, xmlVersion{Key: v.Key, VersionId: v.VersionID, IsLatest: v.IsLatest, LastModified: lm, ETag: `"` + v.ETag + `"`, Size: v.Size}) |
| 287 | } |
| 288 | } |
| 289 | if more && len(vs) > 0 { |
| 290 | res.NextKeyMarker, res.NextVersionIdMarker = vs[len(vs)-1].Key, vs[len(vs)-1].VersionID |
| 291 | } |
| 292 | w.Header().Set("Content-Type", "application/xml") |
| 293 | xml.NewEncoder(w).Encode(res) |
| 294 | } |
| 295 | |
| 296 | type xmlObject struct { |
| 297 | Key string |
| 298 | LastModified string |
| 299 | ETag string |
| 300 | Size int64 |
| 301 | StorageClass string |
| 302 | } |
| 303 | |
| 304 | type xmlCommonPrefix struct { |
| 305 | Prefix string |
| 306 | } |
| 307 | |
| 308 | type listBucketResult struct { |
| 309 | XMLName xml.Name `xml:"ListBucketResult"` |
| 310 | Name string |
| 311 | Prefix string |
| 312 | Delimiter string `xml:",omitempty"` |
| 313 | MaxKeys int |
| 314 | KeyCount int |
| 315 | IsTruncated bool |
| 316 | ContinuationToken string `xml:",omitempty"` |
| 317 | NextContinuationToken string `xml:",omitempty"` |
| 318 | StartAfter string `xml:",omitempty"` |
| 319 | Contents []xmlObject |
| 320 | CommonPrefixes []xmlCommonPrefix |
| 321 | } |
| 322 | |
| 323 | // serveListV2 is ListObjectsV2, the subset litestream's replica client |
| 324 | // (its restore lists with a delimiter) and lease.NewS3Mirror page through: |
| 325 | // the current objects under prefix in key order (an object whose newest |
| 326 | // version is a delete marker is not listed), a delimiter rolling every key |
| 327 | // that holds it after the prefix into one common prefix, max-keys (at most |
| 328 | // 1000, and at most PageSize when that is set lower; keys and common |
| 329 | // prefixes both count) and continuation-token pagination. The token is |
| 330 | // opaque to the client; here it is the last key or common prefix of the |
| 331 | // previous page, and a page resumes after it (after every key under it, |
| 332 | // for a common prefix). start-after is honoured on the first page. |
| 333 | func (f *VersionedBucket) serveListV2(w http.ResponseWriter, bucket string, q url.Values) { |
| 334 | prefix, delim := q.Get("prefix"), q.Get("delimiter") |
| 335 | if err := f.fail("list", prefix); err != nil { |
| 336 | writeErr(w, err) |
| 337 | return |
| 338 | } |
| 339 | size := 1000 |
| 340 | if f.PageSize > 0 && f.PageSize < size { |
| 341 | size = f.PageSize |
| 342 | } |
| 343 | if raw := q.Get("max-keys"); raw != "" { |
| 344 | n, err := strconv.Atoi(raw) |
| 345 | if err != nil || n < 0 { |
| 346 | s3err(w, http.StatusBadRequest, "InvalidArgument") |
| 347 | return |
| 348 | } |
| 349 | size = min(size, n) |
| 350 | } |
| 351 | after := q.Get("start-after") |
| 352 | token := q.Get("continuation-token") |
| 353 | if token != "" { |
| 354 | last, err := base64.StdEncoding.DecodeString(token) |
| 355 | if err != nil || len(last) == 0 { |
| 356 | s3errMsg(w, http.StatusBadRequest, "InvalidArgument", "The continuation token provided is incorrect") |
| 357 | return |
| 358 | } |
| 359 | after = string(last) |
| 360 | } |
| 361 | // entries are the page candidates in key order: a key, or a common |
| 362 | // prefix (isPrefix) standing for every key under it. |
| 363 | type entry struct { |
| 364 | name string |
| 365 | isPrefix bool |
| 366 | } |
| 367 | var keys []string |
| 368 | for k, vs := range f.buckets[bucket] { |
| 369 | if strings.HasPrefix(k, prefix) && len(vs) > 0 && !vs[0].deleteMarker { |
| 370 | keys = append(keys, k) |
| 371 | } |
| 372 | } |
| 373 | sort.Strings(keys) |
| 374 | var entries []entry |
| 375 | for _, k := range keys { |
| 376 | e := entry{name: k} |
| 377 | if delim != "" { |
| 378 | if i := strings.Index(k[len(prefix):], delim); i >= 0 { |
| 379 | e = entry{name: k[:len(prefix)+i+len(delim)], isPrefix: true} |
| 380 | } |
| 381 | } |
| 382 | if n := len(entries); n > 0 && entries[n-1] == e { |
| 383 | continue |
| 384 | } |
| 385 | // A common prefix that ended the last page is that page's token |
| 386 | // itself, and so is every key rolled into it. |
| 387 | if e.name <= after { |
| 388 | continue |
| 389 | } |
| 390 | entries = append(entries, e) |
| 391 | } |
| 392 | res := listBucketResult{Name: bucket, Prefix: prefix, Delimiter: delim, MaxKeys: size, ContinuationToken: token} |
| 393 | if token == "" { |
| 394 | res.StartAfter = q.Get("start-after") |
| 395 | } |
| 396 | page := entries |
| 397 | if len(page) > size { |
| 398 | page = page[:size] |
| 399 | res.IsTruncated = size > 0 |
| 400 | } |
| 401 | for _, e := range page { |
| 402 | if e.isPrefix { |
| 403 | res.CommonPrefixes = append(res.CommonPrefixes, xmlCommonPrefix{Prefix: e.name}) |
| 404 | continue |
| 405 | } |
| 406 | v := f.buckets[bucket][e.name][0] |
| 407 | res.Contents = append(res.Contents, xmlObject{Key: e.name, LastModified: v.modified.UTC().Format(time.RFC3339), ETag: `"` + v.etag + `"`, Size: int64(len(v.body)), StorageClass: "STANDARD"}) |
| 408 | } |
| 409 | res.KeyCount = len(page) |
| 410 | if res.IsTruncated && len(page) > 0 { |
| 411 | res.NextContinuationToken = base64.StdEncoding.EncodeToString([]byte(page[len(page)-1].name)) |
| 412 | } |
| 413 | w.Header().Set("Content-Type", "application/xml") |
| 414 | xml.NewEncoder(w).Encode(res) |
| 415 | } |
| 416 | |
| 417 | func (f *VersionedBucket) serveCopy(w http.ResponseWriter, r *http.Request, bucket, key string) { |
| 418 | src := strings.TrimPrefix(r.Header.Get("x-amz-copy-source"), "/") |
| 419 | srcPath, rawQ, _ := strings.Cut(src, "?") |
| 420 | if strings.ContainsAny(srcPath, "+? ") { |
| 421 | s3err(w, http.StatusBadRequest, "InvalidArgument") // AWS's encoder escapes these; a raw one is a client bug |
| 422 | return |
| 423 | } |
| 424 | sq, err := url.ParseQuery(rawQ) |
| 425 | if err != nil { |
| 426 | s3err(w, http.StatusBadRequest, "InvalidArgument") |
| 427 | return |
| 428 | } |
| 429 | unesc, err := url.PathUnescape(srcPath) |
| 430 | if err != nil { |
| 431 | s3err(w, http.StatusBadRequest, "InvalidArgument") |
| 432 | return |
| 433 | } |
| 434 | sb, sk, _ := strings.Cut(unesc, "/") |
| 435 | srcVersion := sq.Get("versionId") |
| 436 | if srcVersion == "" { // no version named: the source's current one |
| 437 | cur := f.buckets[sb][sk] |
| 438 | if len(cur) == 0 || cur[0].deleteMarker { |
| 439 | s3err(w, http.StatusNotFound, "NoSuchKey") |
| 440 | return |
| 441 | } |
| 442 | srcVersion = cur[0].id |
| 443 | } |
| 444 | if im := r.Header.Get("x-amz-copy-source-if-match"); im != "" { |
| 445 | if v, ok := f.find(sb, sk, srcVersion); ok && strings.Trim(im, `"`) != v.etag { |
| 446 | s3err(w, http.StatusPreconditionFailed, "PreconditionFailed") |
| 447 | return |
| 448 | } |
| 449 | } |
| 450 | id, etag, err := f.copyLocked(sb, sk, srcVersion, bucket, key, r.Header.Get("If-None-Match") == "*") |
| 451 | if errors.Is(err, lease.ErrIsDeleteMarker) { |
| 452 | s3errMsg(w, http.StatusBadRequest, "InvalidRequest", copyMarkerMsg) // a COPY is 400, not GET's 405 |
| 453 | return |
| 454 | } |
| 455 | if err != nil { |
| 456 | writeErr(w, err) |
| 457 | return |
| 458 | } |
| 459 | w.Header().Set("x-amz-version-id", id) |
| 460 | w.Header().Set("Content-Type", "application/xml") |
| 461 | io.WriteString(w, `<CopyObjectResult><ETag>"`+etag+`"</ETag><LastModified>2026-10-05T00:00:00Z</LastModified></CopyObjectResult>`) |
| 462 | } |
| 463 | |
amadan