diff --git a/README.md b/README.md
index 5121b22..3a2eb76 100644
--- a/README.md
+++ b/README.md
@@ -25,7 +25,7 @@ failure replays from its seed.
 | [`aws/cur`](https://pkg.go.dev/github.com/parallelworks/squall/aws/cur) | AWS Cost and Usage Reports: report definitions (put with the API's field rules, a per-account limit and a check that the S3 bucket's policy allows report delivery, describe with pages, delete). |
 | [`aws/ec2`](https://pkg.go.dev/github.com/parallelworks/squall/aws/ec2) | Amazon EC2: instances (run with client tokens, describe with filters and pages, start, stop, terminate, change type or which attached volumes terminating deletes), images (describe, register from snapshots, deregister), instance types, EBS volumes (create, describe, attach, detach, delete), EBS snapshots (create, copy across regions, describe with filters and pages, delete), placement groups, tags, Elastic IP addresses (allocate, describe, release, associate with an interface or instance, disassociate), network interfaces (create with client tokens, describe, delete), and VPC networking: VPCs and their DNS attributes, subnets, availability zones, internet gateways (create, attach, detach, delete), NAT gateways on Elastic IP addresses, route tables with their routes and subnet associations, DHCP options, security groups (create, describe groups and rules with filters and pages, authorize and revoke ingress and egress rules, delete) with each VPC's default group, and VPC attachments to transit gateways the test sets up, refusing deletes with dependencies as EC2 does. |
 | [`aws/route53`](https://pkg.go.dev/github.com/parallelworks/squall/aws/route53) | Amazon Route 53: private hosted zones (create with caller references, get, list by name with pages, delete), their VPC associations, tags, and record sets (change in atomic batches with CREATE, UPSERT and DELETE, list with pages), and changes (get). |
-| [`aws/s3`](https://pkg.go.dev/github.com/parallelworks/squall/aws/s3) | Amazon S3: general purpose buckets (create, head, list with pages and filters, delete), their tags, versioning, policy and CORS rules, and objects (put, delete one or many, list with pages, prefixes and delimiters, list versions and delete markers). |
+| [`aws/s3`](https://pkg.go.dev/github.com/parallelworks/squall/aws/s3) | Amazon S3: general purpose buckets (create, head, list with pages and filters, delete), their tags, versioning, policy, CORS rules and request payment, objects (put, delete one or many, list with pages, prefixes and delimiters, list versions and delete markers), multipart uploads (create, upload parts, complete with S3's part checks, abort, list uploads and parts with pages), and Object Lock: bucket configuration with default retention, per-version GOVERNANCE and COMPLIANCE retention and legal holds that refuse deletes of a locked version, with governance bypass, on the server's clock. |
 | [`azure/arm`](https://pkg.go.dev/github.com/parallelworks/squall/azure/arm) | Azure Resource Manager: resource groups (get, create or update, list with tag filters, list the resources in one, delete with everything in them that nothing outside uses), virtual machines (create or update, get, list, instance view, deallocate, start, delete) and their extensions (create or update, get, list, delete), Uniform scale sets (create or update, update, get, list, delete, delete instances) and their VMs, managed disks (VM OS disks included) and snapshots, with cross-region incremental snapshot copies whose completionPercent climbs to 100, Azure Compute Gallery galleries, image definitions and image versions with their replication status, virtual networks and their subnets, NAT gateways, public load balancers with backend pools and outbound rules, network security groups, network interfaces and public IP addresses, the IP configurations of scale set VMs, private DNS zones, virtual network links and A record sets, resource provider registration, storage accounts with their blob containers (soft-deleted ones listed, legal holds and immutability policies that block deletes), file shares with usage stats, snapshots and soft delete, queues and tables, Azure NetApp Files accounts, capacity pools, volumes, volume snapshots, caches, backup vaults with their backups, backup policies, snapshot policies and volume groups, the Blob service's data plane (containers, and block blobs put, listed with versions, snapshots, soft-deleted and uncommitted blobs, and deleted) with RBAC data actions and account network rules, Cost Management exports at a subscription's scope, Log Analytics workspaces (create or update, recover, get, list, delete with soft deletion) and the default workspace Application Insights makes for a component that names none, Cognitive Services deployments with partner model attestations, and long-running operations to poll. |
 | [`azure/graph`](https://pkg.go.dev/github.com/parallelworks/squall/azure/graph) | Microsoft Graph: the organization (list), a directory's domains (list) and users (create with Microsoft Entra's name and password rules, get by ID or user principal name, delete, permanently delete once deleted), with application permissions checked as Graph documents them. |
 | [`gcp/bigquery`](https://pkg.go.dev/github.com/parallelworks/squall/gcp/bigquery) | BigQuery: datasets (insert with the API's ID and label rules, get, list with pages, patch friendly name, description and labels with etag checks, delete refusing a dataset that holds tables unless `deleteContents`), optionally refused in a project where a `gcp/serviceusage` server has not enabled the API. |
@@ -351,10 +351,28 @@ For Amazon S3:
   deleted. `DeleteBucket` fails with `BucketNotEmpty` while any version or
   delete marker remains.
 - Objects keep their size and ETag, not their data. Checksums are not
-  verified, and multipart uploads, copies, object tags and lifecycle are not
-  supported. Policies and CORS rules are recorded, not enforced. S3 documents
-  in-progress multipart uploads as blocking `DeleteBucket` only for directory
-  buckets.
+  verified, and copies, object tags and lifecycle are not supported. Policies
+  and CORS rules are recorded, not enforced, and so is request payment: with
+  one account, every caller owns the bucket and pays nothing extra.
+- Multipart uploads check their parts on completion as S3 does
+  (`InvalidPartOrder`, `InvalidPart`, `EntityTooSmall` for a part under 5 MiB
+  before the last), and completing one ends it, so a retry after a lost
+  response fails with `NoSuchUpload`. `MultipartUploads` reports the parts an
+  upload holds, which S3 bills as storage, and `CreateMultipartUpload` starts
+  one directly. S3 documents in-progress uploads as blocking `DeleteBucket`
+  only for directory buckets, so deleting a general purpose bucket discards
+  them.
+- Object Lock refuses to delete a version under an unexpired retention or a
+  legal hold with 403 `AccessDenied`, by the server's clock. `COMPLIANCE`
+  never yields; `GOVERNANCE` yields to `x-amz-bypass-governance-retention`
+  from a caller granted `s3:BypassGovernanceRetention`. A delete without a
+  version ID adds a delete marker, and `DeleteBucket` fails while any version
+  remains. Changing a `GOVERNANCE` retention to `COMPLIANCE` also needs the
+  bypass, which is inferred: S3 documents it only for shortening and removal,
+  but lets no one alter governance lock settings without special permissions.
+  Object Lock writes need Content-MD5 or a checksum, multipart parts
+  included. `SetRetention` and `SetLegalHold` lock a version directly. Event
+  holds are not supported.
 - A fault rule's `Resource` is the bucket name, or `bucket/key` for an object
   call. A lost `HeadBucket` breaks off inside its status line, since a sent
   header would read as an answer.
diff --git a/aws/s3/buckets.go b/aws/s3/buckets.go
index 7f95f8f..f7cfb64 100644
--- a/aws/s3/buckets.go
+++ b/aws/s3/buckets.go
@@ -38,6 +38,12 @@ type Bucket struct {
 	Policy string
 	// CORS is the CORSConfiguration XML as sent, empty for none.
 	CORS string
+	// ObjectLock is whether Object Lock is enabled, which needs Versioning "Enabled".
+	ObjectLock bool
+	// DefaultRetention is the retention an Object Lock bucket gives new versions, nil for none.
+	DefaultRetention *DefaultRetention
+	// RequesterPays is whether the bucket's request payment configuration names the requester as payer.
+	RequesterPays bool
 }
 
 type bucket struct {
@@ -46,11 +52,16 @@ type bucket struct {
 	objects map[string][]*version
 	// nullSeqs holds each key's last "null" version's sequence number, kept after a delete so a marker naming it keeps its place.
 	nullSeqs map[string]uint64
+	// uploads holds the in-progress multipart uploads by upload ID.
+	uploads map[string]*upload
 }
 
 func (b *bucket) clone() Bucket {
 	c := b.Bucket
 	c.Tags = maps.Clone(b.Tags)
+	if b.DefaultRetention != nil {
+		c.DefaultRetention = new(*b.DefaultRetention)
+	}
 	return c
 }
 
@@ -129,9 +140,7 @@ func (s *Server) createBucket(c *call) (*reply, *apiError) {
 	if o := c.r.Header.Get("x-amz-object-ownership"); o != "" && o != "BucketOwnerEnforced" {
 		return nil, invalidArgument("squall: object ownership %s is not supported", o)
 	}
-	if strings.EqualFold(c.r.Header.Get("x-amz-bucket-object-lock-enabled"), "true") {
-		return nil, invalidArgument("squall: object lock is not supported")
-	}
+	locked := strings.EqualFold(c.r.Header.Get("x-amz-bucket-object-lock-enabled"), "true")
 	var cfg createBucketConfiguration
 	if len(c.body) > 0 {
 		if err := decodeXML(c.body, &cfg); err != nil {
@@ -173,7 +182,11 @@ func (s *Server) createBucket(c *call) (*reply, *apiError) {
 			Message: fmt.Sprintf("The %s location constraint is incompatible for the region specific endpoint this request was sent to.", constraint),
 		}
 	}
-	s.buckets[c.bucket] = &bucket{Name: c.bucket, Region: c.region, CreationDate: s.now().UTC(), objects: map[string][]*version{}}
+	b := &bucket{Name: c.bucket, Region: c.region, CreationDate: s.now().UTC(), objects: map[string][]*version{}}
+	if locked {
+		b.ObjectLock, b.Versioning = true, versioningEnabled
+	}
+	s.buckets[c.bucket] = b
 	location := "/" + c.bucket
 	if c.region != "us-east-1" {
 		location = "http://" + c.bucket + ".s3.amazonaws.com/"
@@ -274,7 +287,7 @@ func (s *Server) listBuckets(c *call) (*reply, *apiError) {
 }
 
 type taggingXML struct {
-	XMLName xml.Name `xml:"Tagging"`
+	XMLName xml.Name
 	TagSet  []tagXML `xml:"TagSet>Tag"`
 }
 
@@ -311,7 +324,7 @@ func validTagText(s string) bool {
 
 func parseTags(body []byte) (map[string]string, *apiError) {
 	var in taggingXML
-	if err := decodeXML(body, &in); err != nil {
+	if err := decodeRootXML(body, "Tagging", &in); err != nil {
 		return nil, err
 	}
 	if len(in.TagSet) > maxTags {
@@ -365,9 +378,9 @@ func (s *Server) deleteBucketTagging(c *call) (*reply, *apiError) {
 }
 
 type versioningXML struct {
-	XMLName   xml.Name `xml:"VersioningConfiguration"`
-	Status    string   `xml:"Status,omitempty"`
-	MfaDelete string   `xml:"MfaDelete,omitempty"`
+	XMLName   xml.Name
+	Status    string `xml:"Status,omitempty"`
+	MfaDelete string `xml:"MfaDelete,omitempty"`
 }
 
 func (s *Server) getBucketVersioning(c *call) (*reply, *apiError) {
@@ -382,7 +395,7 @@ func (s *Server) getBucketVersioning(c *call) (*reply, *apiError) {
 
 func (s *Server) putBucketVersioning(c *call) (*reply, *apiError) {
 	var in versioningXML
-	if err := decodeXML(c.body, &in); err != nil {
+	if err := decodeRootXML(c.body, "VersioningConfiguration", &in); err != nil {
 		return nil, err
 	}
 	if in.MfaDelete != "" {
@@ -397,6 +410,12 @@ func (s *Server) putBucketVersioning(c *call) (*reply, *apiError) {
 	if err != nil {
 		return nil, err
 	}
+	if b.ObjectLock && in.Status != versioningEnabled {
+		return nil, &apiError{
+			Status: http.StatusConflict, Code: "InvalidBucketState",
+			Message: "An Object Lock configuration is present on this bucket, so the versioning state cannot be changed.", Bucket: c.bucket,
+		}
+	}
 	b.Versioning = in.Status
 	return &reply{}, nil
 }
@@ -454,6 +473,43 @@ func (s *Server) putBucketCors(c *call) (*reply, *apiError) {
 	return &reply{}, nil
 }
 
+type requestPaymentXML struct {
+	XMLName xml.Name
+	Payer   string `xml:"Payer"`
+}
+
+func (s *Server) getBucketRequestPayment(c *call) (*reply, *apiError) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	b, err := s.lookup(c)
+	if err != nil {
+		return nil, err
+	}
+	out := requestPaymentXML{XMLName: xml.Name{Space: namespace, Local: "RequestPaymentConfiguration"}, Payer: "BucketOwner"}
+	if b.RequesterPays {
+		out.Payer = "Requester"
+	}
+	return &reply{body: out}, nil
+}
+
+func (s *Server) putBucketRequestPayment(c *call) (*reply, *apiError) {
+	var in requestPaymentXML
+	if err := decodeRoot(c, "RequestPaymentConfiguration", &in); err != nil {
+		return nil, err
+	}
+	if in.Payer != "Requester" && in.Payer != "BucketOwner" {
+		return nil, malformedXML()
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	b, err := s.lookup(c)
+	if err != nil {
+		return nil, err
+	}
+	b.RequesterPays = in.Payer == "Requester"
+	return &reply{}, nil
+}
+
 // Buckets returns every bucket the account owns, ordered by name.
 func (s *Server) Buckets() []Bucket {
 	s.mu.Lock()
@@ -487,6 +543,18 @@ func (s *Server) PutBucket(b Bucket) (Bucket, error) {
 	if b.Versioning != "" && b.Versioning != versioningEnabled && b.Versioning != versioningSuspended {
 		return Bucket{}, fmt.Errorf("squall: versioning %q is not one of Enabled and Suspended", b.Versioning)
 	}
+	if b.ObjectLock && b.Versioning != versioningEnabled {
+		return Bucket{}, errors.New("squall: Object Lock needs versioning Enabled")
+	}
+	if b.DefaultRetention != nil {
+		if !b.ObjectLock {
+			return Bucket{}, errors.New("squall: a default retention needs Object Lock")
+		}
+		if err := b.DefaultRetention.valid(); err != nil {
+			return Bucket{}, err
+		}
+		b.DefaultRetention = new(*b.DefaultRetention)
+	}
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	if _, ok := s.foreign[b.Name]; ok {
@@ -499,7 +567,7 @@ func (s *Server) PutBucket(b Bucket) (Bucket, error) {
 	b.Tags = maps.Clone(b.Tags)
 	stored := &bucket{Bucket: b, objects: map[string][]*version{}}
 	if old, ok := s.buckets[b.Name]; ok {
-		stored.objects, stored.nullSeqs = old.objects, old.nullSeqs
+		stored.objects, stored.nullSeqs, stored.uploads = old.objects, old.nullSeqs, old.uploads
 	}
 	s.buckets[b.Name] = stored
 	return stored.clone(), nil
diff --git a/aws/s3/http_test.go b/aws/s3/http_test.go
index bcf93f7..f981d57 100644
--- a/aws/s3/http_test.go
+++ b/aws/s3/http_test.go
@@ -2,6 +2,7 @@ package s3_test
 
 import (
 	"bytes"
+	"cmp"
 	"encoding/xml"
 	"errors"
 	"io"
@@ -9,6 +10,7 @@ import (
 	"net/http/httptest"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/parallelworks/squall/aws/s3"
 )
@@ -223,6 +225,202 @@ func TestHostileBodies(t *testing.T) {
 	}
 }
 
+func TestMultipartAndLockShapes(t *testing.T) {
+	s := s3.New()
+	if _, err := s.PutBucket(s3.Bucket{Name: "vault", Region: "us-east-1", Versioning: "Enabled", ObjectLock: true}); err != nil {
+		t.Fatal(err)
+	}
+	code, _, body := do(t, s, http.MethodPost, "/vault/a%20b?uploads", authorization, "")
+	if code != http.StatusOK || !strings.Contains(body,
+		`<InitiateMultipartUploadResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Bucket>vault</Bucket><Key>a b</Key><UploadId>`) {
+		t.Fatalf("create upload = %d %s", code, body)
+	}
+	if code, _, body := do(t, s, http.MethodGet, "/vault?uploads&max-uploads=5000", authorization, ""); code != http.StatusOK || !strings.Contains(body, "<MaxUploads>1000</MaxUploads>") {
+		t.Fatalf("max-uploads 5000 = %d %s, want a page of 1000", code, body)
+	}
+	code, _, body = do(t, s, http.MethodGet, "/vault?uploads&encoding-type=url", authorization, "")
+	if code != http.StatusOK || !strings.Contains(body, `<KeyMarker></KeyMarker><UploadIdMarker></UploadIdMarker><NextKeyMarker>a+b</NextKeyMarker><NextUploadIdMarker>`) || !strings.Contains(body, `</NextUploadIdMarker><Prefix></Prefix><MaxUploads>1000</MaxUploads>`+
+		`<EncodingType>url</EncodingType><IsTruncated>false</IsTruncated><Upload><Key>a+b</Key>`) ||
+		!strings.Contains(body, `<StorageClass>STANDARD</StorageClass><Initiated>`) {
+		t.Fatalf("list uploads = %d %s", code, body)
+	}
+	lockConfig := "<ObjectLockConfiguration><ObjectLockEnabled>Enabled</ObjectLockEnabled></ObjectLockConfiguration>"
+	for _, tc := range []struct {
+		name, method, target string
+		header               map[string]string
+		body                 string
+		status               int
+		code                 string
+	}{
+		{"a retention without a checksum", http.MethodPut, "/vault/k", map[string]string{
+			"x-amz-object-lock-mode": "GOVERNANCE", "x-amz-object-lock-retain-until-date": "2099-01-01T00:00:00Z",
+		}, "", http.StatusBadRequest, "InvalidRequest"},
+		{"a mode without a date", http.MethodPut, "/vault/k", map[string]string{"x-amz-object-lock-mode": "GOVERNANCE"}, "", http.StatusBadRequest, "InvalidArgument"},
+		{"an unknown mode", http.MethodPut, "/vault/k", map[string]string{
+			"x-amz-object-lock-mode": "STRICT", "x-amz-object-lock-retain-until-date": "2099-01-01T00:00:00Z",
+		}, "", http.StatusBadRequest, "InvalidArgument"},
+		{"a date in the past", http.MethodPut, "/vault/k", map[string]string{
+			"x-amz-object-lock-mode": "GOVERNANCE", "x-amz-object-lock-retain-until-date": "2000-01-01T00:00:00Z", "Content-MD5": "x",
+		}, "", http.StatusBadRequest, "InvalidArgument"},
+		{"a bad legal hold", http.MethodPut, "/vault/k", map[string]string{"x-amz-object-lock-legal-hold": "on"}, "", http.StatusBadRequest, "InvalidArgument"},
+		{"an event hold", http.MethodPut, "/vault/k", map[string]string{"x-amz-object-lock-event-hold": "ON"}, "", http.StatusBadRequest, "InvalidArgument"},
+		{"a completion under another root", http.MethodPost, "/vault/k?uploadId=x", nil, "<Complete><Part><PartNumber>1</PartNumber><ETag>x</ETag></Part></Complete>", http.StatusBadRequest, "MalformedXML"},
+		{"a part without an upload", http.MethodPut, "/vault/k?partNumber=1", nil, "", http.StatusBadRequest, "InvalidArgument"},
+		{"a part of an unknown upload", http.MethodPut, "/vault/k?partNumber=1&uploadId=nope", nil, "", http.StatusNotFound, "NoSuchUpload"},
+		{"part zero", http.MethodPut, "/vault/k?partNumber=0&uploadId=nope", nil, "", http.StatusBadRequest, "InvalidArgument"},
+		{"a negative max-uploads", http.MethodGet, "/vault?uploads&max-uploads=-1", nil, "", http.StatusBadRequest, "InvalidArgument"},
+		{"max-uploads 0", http.MethodGet, "/vault?uploads&max-uploads=0", nil, "", http.StatusBadRequest, "InvalidArgument"},
+		{"max-parts 0", http.MethodGet, "/vault/k?uploadId=x&max-parts=0", nil, "", http.StatusBadRequest, "InvalidArgument"},
+		{"a retention with only a checksum type", http.MethodPut, "/vault/k", map[string]string{
+			"x-amz-object-lock-mode": "GOVERNANCE", "x-amz-object-lock-retain-until-date": "2099-01-01T00:00:00Z",
+			"x-amz-checksum-type": "FULL_OBJECT", "x-amz-checksum-algorithm": "CRC32", "x-amz-sdk-checksum-algorithm": "CRC32",
+		}, "", http.StatusBadRequest, "InvalidRequest"},
+		{
+			"a versioning configuration under another root", http.MethodPut, "/vault?versioning", nil,
+			"<Status><Status>Enabled</Status></Status>", http.StatusBadRequest, "MalformedXML",
+		},
+		{
+			"a request payment configuration under another root", http.MethodPut, "/vault?requestPayment",
+			map[string]string{"Content-MD5": "x"},
+			"<Payment><Payer>Requester</Payer></Payment>", http.StatusBadRequest, "MalformedXML",
+		},
+		{
+			"a request payment configuration without a checksum", http.MethodPut, "/vault?requestPayment", nil,
+			"<RequestPaymentConfiguration><Payer>Requester</Payer></RequestPaymentConfiguration>", http.StatusBadRequest, "InvalidRequest",
+		},
+		{"retention of a missing key", http.MethodGet, "/vault/missing?retention", nil, "", http.StatusNotFound, "NoSuchKey"},
+		{"a lock configuration without a checksum", http.MethodPut, "/vault?object-lock", nil, lockConfig, http.StatusBadRequest, "InvalidRequest"},
+		{
+			"a lock configuration under another root", http.MethodPut, "/vault?object-lock",
+			map[string]string{"Content-MD5": "x"},
+			"<Nope><ObjectLockEnabled>Enabled</ObjectLockEnabled></Nope>", http.StatusBadRequest, "MalformedXML",
+		},
+		{
+			"a retention without a checksum", http.MethodPut, "/vault/missing?retention", nil,
+			"<Retention><Mode>GOVERNANCE</Mode><RetainUntilDate>2099-01-01T00:00:00Z</RetainUntilDate></Retention>", http.StatusBadRequest, "InvalidRequest",
+		},
+		{
+			"a retention under another root", http.MethodPut, "/vault/missing?retention",
+			map[string]string{"Content-MD5": "x"},
+			"<LegalHold><Status>ON</Status></LegalHold>", http.StatusBadRequest, "MalformedXML",
+		},
+		{"a legal hold without a checksum", http.MethodPut, "/vault/missing?legal-hold", nil, "<LegalHold><Status>ON</Status></LegalHold>", http.StatusBadRequest, "InvalidRequest"},
+		{
+			"a legal hold under another root", http.MethodPut, "/vault/missing?legal-hold",
+			map[string]string{"Content-MD5": "x"},
+			"<Retention><Status>ON</Status></Retention>", http.StatusBadRequest, "MalformedXML",
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			req := httptest.NewRequestWithContext(t.Context(), tc.method, tc.target, strings.NewReader(cmp.Or(tc.body, "x")))
+			req.Header.Set("Authorization", authorization)
+			for k, v := range tc.header {
+				req.Header.Set(k, v)
+			}
+			rec := httptest.NewRecorder()
+			s.ServeHTTP(rec, req)
+			if rec.Code != tc.status || !strings.Contains(rec.Body.String(), "<Code>"+tc.code+"</Code>") {
+				t.Fatalf("= %d %s, want %d %s", rec.Code, rec.Body, tc.status, tc.code)
+			}
+		})
+	}
+	req := httptest.NewRequestWithContext(t.Context(), http.MethodPut, "/vault?object-lock", strings.NewReader(
+		"<ObjectLockConfiguration><ObjectLockEnabled>Enabled</ObjectLockEnabled><Rule><DefaultRetention><Mode>GOVERNANCE</Mode><Years>1</Years></DefaultRetention></Rule></ObjectLockConfiguration>"))
+	req.Header.Set("Authorization", authorization)
+	req.Header.Set("Content-MD5", "x")
+	rec := httptest.NewRecorder()
+	s.ServeHTTP(rec, req)
+	if rec.Code != http.StatusOK {
+		t.Fatalf("put lock configuration = %d %s", rec.Code, rec.Body)
+	}
+	code, _, body = do(t, s, http.MethodGet, "/vault?object-lock", authorization, "")
+	if code != http.StatusOK || !strings.Contains(body,
+		`<ObjectLockConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><ObjectLockEnabled>Enabled</ObjectLockEnabled><Rule><DefaultRetention><Mode>GOVERNANCE</Mode><Years>1</Years></DefaultRetention></Rule></ObjectLockConfiguration>`) {
+		t.Fatalf("get lock configuration = %d %s", code, body)
+	}
+	if code, _, body = do(t, s, http.MethodPut, "/vault/k", authorization, "x"); code != http.StatusBadRequest || !strings.Contains(body, "<Code>InvalidRequest</Code>") {
+		t.Fatalf("a put under default retention without a checksum = %d %s", code, body)
+	}
+}
+
+func TestListMultipartUploadsNextMarkers(t *testing.T) {
+	s := s3.New()
+	if _, err := s.PutBucket(s3.Bucket{Name: "markers", Region: "us-east-1"}); err != nil {
+		t.Fatal(err)
+	}
+	var sample string
+	for _, key := range []string{"a/1", "sample.jpg", "videos/1"} {
+		up, err := s.CreateMultipartUpload("markers", key)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if key == "sample.jpg" {
+			sample = up.UploadID
+		}
+	}
+	code, _, body := do(t, s, http.MethodGet, "/markers?uploads&delimiter=/", authorization, "")
+	if want := "<NextKeyMarker>sample.jpg</NextKeyMarker><NextUploadIdMarker>" + sample + "</NextUploadIdMarker>"; code != http.StatusOK ||
+		!strings.Contains(body, want) || !strings.Contains(body, "<IsTruncated>false</IsTruncated>") {
+		t.Fatalf("an untruncated page = %d %s, want %s", code, body, want)
+	}
+	code, _, body = do(t, s, http.MethodGet, "/markers?uploads&delimiter=/&max-uploads=1", authorization, "")
+	if want := "<NextKeyMarker>a/</NextKeyMarker><NextUploadIdMarker></NextUploadIdMarker>"; code != http.StatusOK ||
+		!strings.Contains(body, want) || !strings.Contains(body, "<IsTruncated>true</IsTruncated>") {
+		t.Fatalf("a page ending on a prefix = %d %s, want %s", code, body, want)
+	}
+	code, _, body = do(t, s, http.MethodGet, "/markers?uploads&delimiter=/&key-marker=a/1", authorization, "")
+	if code != http.StatusOK || strings.Contains(body, "<Prefix>a/</Prefix>") || !strings.Contains(body, "<Prefix>videos/</Prefix>") {
+		t.Fatalf("after a key inside a/ = %d %s, want videos/ and not a/", code, body)
+	}
+}
+
+func TestBucketConfigurationsUseTheS3Namespace(t *testing.T) {
+	s := s3.New()
+	if _, err := s.PutBucket(s3.Bucket{Name: "namespaced", Region: "us-east-1", Versioning: "Enabled", Tags: map[string]string{"k": "v"}}); err != nil {
+		t.Fatal(err)
+	}
+	for _, tc := range []struct{ query, root string }{
+		{"versioning", "VersioningConfiguration"},
+		{"requestPayment", "RequestPaymentConfiguration"},
+		{"tagging", "Tagging"},
+	} {
+		code, _, body := do(t, s, http.MethodGet, "/namespaced?"+tc.query, authorization, "")
+		if want := "<" + tc.root + ` xmlns="http://s3.amazonaws.com/doc/2006-03-01/">`; code != http.StatusOK || !strings.Contains(body, want) {
+			t.Errorf("%s = %d %s, want %s", tc.query, code, body, want)
+		}
+	}
+}
+
+func TestRetainUntilUnderAMillisecondAheadIsInThePast(t *testing.T) {
+	s := s3.New(s3.WithClock(func() time.Time { return time.Date(2026, 1, 1, 0, 0, 0, 500_000, time.UTC) }))
+	if _, err := s.PutBucket(s3.Bucket{Name: "vault", Region: "us-east-1", Versioning: "Enabled", ObjectLock: true}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.PutObject("vault", "k", []byte("k")); err != nil {
+		t.Fatal(err)
+	}
+	const soon = "2026-01-01T00:00:00.0009Z"
+	req := httptest.NewRequestWithContext(t.Context(), http.MethodPut, "/vault/k", strings.NewReader("k"))
+	req.Header.Set("Authorization", authorization)
+	req.Header.Set("Content-MD5", "x")
+	req.Header.Set("x-amz-object-lock-mode", "COMPLIANCE")
+	req.Header.Set("x-amz-object-lock-retain-until-date", soon)
+	rec := httptest.NewRecorder()
+	s.ServeHTTP(rec, req)
+	if rec.Code != http.StatusBadRequest || !strings.Contains(rec.Body.String(), "<Code>InvalidArgument</Code>") {
+		t.Fatalf("put = %d %s, want 400 InvalidArgument", rec.Code, rec.Body)
+	}
+	req = httptest.NewRequestWithContext(t.Context(), http.MethodPut, "/vault/k?retention", strings.NewReader(
+		"<Retention><Mode>COMPLIANCE</Mode><RetainUntilDate>"+soon+"</RetainUntilDate></Retention>"))
+	req.Header.Set("Authorization", authorization)
+	req.Header.Set("Content-MD5", "x")
+	rec = httptest.NewRecorder()
+	s.ServeHTTP(rec, req)
+	if rec.Code != http.StatusBadRequest || !strings.Contains(rec.Body.String(), "<Code>InvalidArgument</Code>") {
+		t.Fatalf("put retention = %d %s, want 400 InvalidArgument", rec.Code, rec.Body)
+	}
+}
+
 func FuzzServeHTTP(f *testing.F) {
 	for _, seed := range []struct{ method, target, header, body string }{
 		{http.MethodPut, "/fuzz/k", "Content-Encoding: aws-chunked", "5;chunk-signature=0\r\nhello\r\n0\r\n\r\n"},
@@ -232,6 +430,10 @@ func FuzzServeHTTP(f *testing.F) {
 		{http.MethodPut, "/fuzz?tagging", "X-Amz-Content-Sha256: STREAMING-UNSIGNED-PAYLOAD-TRAILER", "0\r\n\r\n"},
 		{http.MethodGet, "/?max-buckets=1&continuation-token=YQ&bucket-region=us-east-1", "", ""},
 		{http.MethodPut, "/other", "", "<CreateBucketConfiguration><LocationConstraint>eu-west-1</LocationConstraint></CreateBucketConfiguration>"},
+		{http.MethodGet, "/fuzz?uploads&delimiter=/&key-marker=a/&upload-id-marker=0&max-uploads=1&encoding-type=url", "", ""},
+		{http.MethodPost, "/fuzz/k?uploadId=0", "", "<CompleteMultipartUpload><Part><PartNumber>1</PartNumber><ETag>x</ETag></Part></CompleteMultipartUpload>"},
+		{http.MethodPut, "/fuzz/k?retention", "", "<Retention><Mode>COMPLIANCE</Mode><RetainUntilDate>2099-01-01T00:00:00Z</RetainUntilDate></Retention>"},
+		{http.MethodPut, "/fuzz?object-lock", "", "<ObjectLockConfiguration><ObjectLockEnabled>Enabled</ObjectLockEnabled></ObjectLockConfiguration>"},
 	} {
 		f.Add(seed.method, seed.target, seed.header, []byte(seed.body))
 	}
diff --git a/aws/s3/multipart.go b/aws/s3/multipart.go
new file mode 100644
index 0000000..e49ba8b
--- /dev/null
+++ b/aws/s3/multipart.go
@@ -0,0 +1,445 @@
+package s3
+
+import (
+	"cmp"
+	"crypto/md5" //nolint:gosec // multipart ETags are MD5 digests; nothing here relies on them for security
+	"encoding/hex"
+	"encoding/xml"
+	"errors"
+	"fmt"
+	"maps"
+	"net/http"
+	"net/url"
+	"slices"
+	"strconv"
+	"strings"
+	"time"
+)
+
+const (
+	maxParts    = 10_000
+	minPartSize = 5 << 20
+	maxListed   = 1000
+)
+
+// MultipartUpload is an upload that was created and not yet completed or aborted.
+type MultipartUpload struct {
+	Bucket    string
+	Key       string
+	UploadID  string
+	Initiated time.Time
+	// Parts are the parts uploaded so far, by part number.
+	Parts []Part
+}
+
+// Part is one uploaded part of a multipart upload.
+type Part struct {
+	Number int
+	Size   int64
+	// ETag is the quoted hex MD5 of the part's data.
+	ETag         string
+	LastModified time.Time
+}
+
+type upload struct {
+	key       string
+	id        string
+	initiated time.Time
+	parts     map[int]Part
+	lock      lockRequest
+}
+
+func (u *upload) export(bucketName string) MultipartUpload {
+	out := MultipartUpload{Bucket: bucketName, Key: u.key, UploadID: u.id, Initiated: u.initiated}
+	for _, n := range slices.Sorted(maps.Keys(u.parts)) {
+		out.Parts = append(out.Parts, u.parts[n])
+	}
+	return out
+}
+
+// sortedUploads returns b's uploads by key and then upload ID, which as a fixed-width counter orders them by initiation.
+func sortedUploads(b *bucket) []*upload {
+	return slices.SortedFunc(maps.Values(b.uploads), func(x, y *upload) int {
+		return cmp.Or(strings.Compare(x.key, y.key), strings.Compare(x.id, y.id))
+	})
+}
+
+func noSuchUpload() *apiError {
+	return &apiError{
+		Status: http.StatusNotFound, Code: "NoSuchUpload",
+		Message: "The specified upload does not exist. The upload ID may be invalid, or the upload may have been aborted or completed.",
+	}
+}
+
+// startUpload creates an upload of key. The caller holds s.mu.
+func (s *Server) startUpload(b *bucket, key string, lock lockRequest) *upload {
+	s.nextUpload++
+	u := &upload{key: key, id: fmt.Sprintf("%032x", s.nextUpload), initiated: s.now().UTC(), parts: map[int]Part{}, lock: lock}
+	if b.uploads == nil {
+		b.uploads = map[string]*upload{}
+	}
+	b.uploads[u.id] = u
+	return u
+}
+
+// findUpload returns c's upload, which must be of c's key. The caller holds s.mu.
+func (s *Server) findUpload(c *call) (*bucket, *upload, *apiError) {
+	b, err := s.lookup(c)
+	if err != nil {
+		return nil, nil, err
+	}
+	u, ok := b.uploads[c.query.Get("uploadId")]
+	if !ok || u.key != c.key {
+		return nil, nil, noSuchUpload()
+	}
+	return b, u, nil
+}
+
+type initiateXML struct {
+	XMLName  xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ InitiateMultipartUploadResult"`
+	Bucket   string   `xml:"Bucket"`
+	Key      string   `xml:"Key"`
+	UploadID string   `xml:"UploadId"`
+}
+
+func (s *Server) createMultipartUpload(c *call) (*reply, *apiError) {
+	if len(c.key) > maxKeyBytes {
+		return nil, keyTooLong()
+	}
+	if acl := c.r.Header.Get("x-amz-acl"); acl != "" && acl != "private" && acl != "bucket-owner-full-control" {
+		return nil, &apiError{Status: http.StatusBadRequest, Code: "AccessControlListNotSupported", Message: "The bucket does not allow ACLs"}
+	}
+	if c.r.Header.Get("X-Amz-Tagging") != "" {
+		return nil, invalidArgument("squall: the x-amz-tagging header is not supported")
+	}
+	lock, err := s.lockHeaders(c.r.Header)
+	if err != nil {
+		return nil, err
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	b, err := s.lookup(c)
+	if err != nil {
+		return nil, err
+	}
+	if err := checkLockable(b, lock); err != nil {
+		return nil, err
+	}
+	u := s.startUpload(b, c.key, lock)
+	return &reply{body: initiateXML{Bucket: c.bucket, Key: c.key, UploadID: u.id}}, nil
+}
+
+// partNumber reads a part number, from 1 to 10,000.
+func partNumber(v string) (int, bool) {
+	n, err := strconv.Atoi(v)
+	return n, err == nil && n >= 1 && n <= maxParts
+}
+
+func (s *Server) uploadPart(c *call) (*reply, *apiError) {
+	if c.r.Header.Get("x-amz-copy-source") != "" {
+		return nil, &apiError{Status: http.StatusNotFound, Code: "NotImplemented", Message: "squall: UploadPartCopy is not implemented"}
+	}
+	if !c.query.Has("uploadId") {
+		return nil, invalidArgument("squall: UploadPart needs an uploadId")
+	}
+	n, ok := partNumber(c.query.Get("partNumber"))
+	if !ok {
+		return nil, invalidArgument("Part number must be an integer between 1 and 10000, inclusive")
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	b, u, err := s.findUpload(c)
+	if err != nil {
+		return nil, err
+	}
+	if (u.lock.mode != "" || b.DefaultRetention != nil) && !hasChecksum(c.r.Header) {
+		return nil, lockChecksumMissing("Put Part")
+	}
+	p := Part{Number: n, Size: int64(len(c.body)), ETag: etagOf(c.body), LastModified: s.now().UTC()}
+	u.parts[n] = p
+	return &reply{header: map[string]string{"ETag": p.ETag}}, nil
+}
+
+type completeXML struct {
+	Parts []struct {
+		PartNumber int    `xml:"PartNumber"`
+		ETag       string `xml:"ETag"`
+	} `xml:"Part"`
+}
+
+type completeResultXML struct {
+	XMLName  xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ CompleteMultipartUploadResult"`
+	Location string   `xml:"Location"`
+	Bucket   string   `xml:"Bucket"`
+	Key      string   `xml:"Key"`
+	ETag     string   `xml:"ETag"`
+}
+
+func (s *Server) completeMultipartUpload(c *call) (*reply, *apiError) {
+	for _, h := range []string{"If-Match", "If-None-Match"} {
+		if c.r.Header.Get(h) != "" {
+			return nil, invalidArgument("squall: the %s header is not supported", h)
+		}
+	}
+	var in completeXML
+	if err := decodeRootXML(c.body, "CompleteMultipartUpload", &in); err != nil {
+		return nil, err
+	}
+	if len(in.Parts) == 0 || len(in.Parts) > maxParts {
+		return nil, malformedXML()
+	}
+	for i, p := range in.Parts {
+		if i > 0 && p.PartNumber <= in.Parts[i-1].PartNumber {
+			return nil, &apiError{
+				Status: http.StatusBadRequest, Code: "InvalidPartOrder",
+				Message: "The list of parts was not in ascending order. The parts list must be specified in order by part number.",
+			}
+		}
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	b, u, err := s.findUpload(c)
+	if err != nil {
+		return nil, err
+	}
+	parts := make([]Part, len(in.Parts))
+	for i, want := range in.Parts {
+		p, ok := u.parts[want.PartNumber]
+		if !ok || strings.Trim(want.ETag, `"`) != strings.Trim(p.ETag, `"`) {
+			return nil, &apiError{
+				Status: http.StatusBadRequest, Code: "InvalidPart",
+				Message: "One or more of the specified parts could not be found.  The part may not have been uploaded, or the specified entity tag may not match the part's entity tag.",
+			}
+		}
+		parts[i] = p
+	}
+	digests := make([]byte, 0, md5.Size*len(parts))
+	var size int64
+	for i, p := range parts {
+		if i < len(parts)-1 && p.Size < minPartSize {
+			return nil, &apiError{Status: http.StatusBadRequest, Code: "EntityTooSmall", Message: "Your proposed upload is smaller than the minimum allowed object size."}
+		}
+		sum, _ := hex.DecodeString(strings.Trim(p.ETag, `"`))
+		digests = append(digests, sum...)
+		size += p.Size
+	}
+	sum := md5.Sum(digests) //nolint:gosec // an ETag, not a security check
+	etag := fmt.Sprintf(`"%s-%d"`, hex.EncodeToString(sum[:]), len(in.Parts))
+	v := s.storeVersion(b, c.key, size, etag)
+	s.applyLock(b, v, u.lock)
+	delete(b.uploads, u.id)
+	header := map[string]string{}
+	if b.Versioning != "" {
+		header["x-amz-version-id"] = v.id
+	}
+	location := fmt.Sprintf("https://%s.s3.%s.amazonaws.com/%s", c.bucket, b.Region, (&url.URL{Path: c.key}).EscapedPath())
+	return &reply{header: header, body: completeResultXML{Location: location, Bucket: c.bucket, Key: c.key, ETag: etag}}, nil
+}
+
+func (s *Server) abortMultipartUpload(c *call) (*reply, *apiError) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	b, u, err := s.findUpload(c)
+	if err != nil {
+		return nil, err
+	}
+	delete(b.uploads, u.id)
+	return &reply{status: http.StatusNoContent}, nil
+}
+
+// listLimit reads a listing's page size parameter, at least 1, which defaults to and is capped at 1000.
+func listLimit(c *call, name string) (int, *apiError) {
+	if !c.query.Has(name) {
+		return maxListed, nil
+	}
+	n, err := strconv.Atoi(c.query.Get(name))
+	if err != nil || n < 1 {
+		return 0, invalidArgument("Provided %s not an integer or within integer range", name)
+	}
+	return min(n, maxListed), nil
+}
+
+var listUploadsParams = []string{"prefix", "delimiter", "key-marker", "upload-id-marker", "max-uploads", "encoding-type"}
+
+type listUploadsXML struct {
+	XMLName            xml.Name          `xml:"http://s3.amazonaws.com/doc/2006-03-01/ ListMultipartUploadsResult"`
+	Bucket             string            `xml:"Bucket"`
+	KeyMarker          string            `xml:"KeyMarker"`
+	UploadIDMarker     string            `xml:"UploadIdMarker"`
+	NextKeyMarker      string            `xml:"NextKeyMarker"`
+	NextUploadIDMarker string            `xml:"NextUploadIdMarker"`
+	Prefix             string            `xml:"Prefix"`
+	Delimiter          string            `xml:"Delimiter,omitempty"`
+	MaxUploads         int               `xml:"MaxUploads"`
+	EncodingType       string            `xml:"EncodingType,omitempty"`
+	IsTruncated        bool              `xml:"IsTruncated"`
+	Uploads            []uploadXML       `xml:"Upload"`
+	CommonPrefixes     []commonPrefixXML `xml:"CommonPrefixes"`
+}
+
+type uploadXML struct {
+	Key          string   `xml:"Key"`
+	UploadID     string   `xml:"UploadId"`
+	Initiator    ownerXML `xml:"Initiator"`
+	Owner        ownerXML `xml:"Owner"`
+	StorageClass string   `xml:"StorageClass"`
+	Initiated    string   `xml:"Initiated"`
+}
+
+func (s *Server) listMultipartUploads(c *call) (*reply, *apiError) {
+	limit, err := listLimit(c, "max-uploads")
+	if err != nil {
+		return nil, err
+	}
+	encode, err := encodingOf(c)
+	if err != nil {
+		return nil, err
+	}
+	prefix, delimiter := c.query.Get("prefix"), c.query.Get("delimiter")
+	keyMarker, idMarker := c.query.Get("key-marker"), c.query.Get("upload-id-marker")
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	b, err := s.lookup(c)
+	if err != nil {
+		return nil, err
+	}
+	out := listUploadsXML{
+		Bucket: c.bucket, KeyMarker: encode(keyMarker), UploadIDMarker: idMarker, Prefix: encode(prefix),
+		Delimiter: encode(delimiter), MaxUploads: limit,
+	}
+	if c.query.Get("encoding-type") != "" {
+		out.EncodingType = "url"
+	}
+	uploads := sortedUploads(b)
+	var lastEntry, lastID string
+	count := 0
+	for _, u := range uploads {
+		if !strings.HasPrefix(u.key, prefix) ||
+			keyMarker != "" && (u.key < keyMarker || u.key == keyMarker && (idMarker == "" || u.id <= idMarker)) {
+			continue
+		}
+		entry, rolled := u.key, false
+		if delimiter != "" {
+			if i := strings.Index(u.key[len(prefix):], delimiter); i >= 0 {
+				entry, rolled = u.key[:len(prefix)+i+len(delimiter)], true
+			}
+		}
+		if rolled && (entry <= keyMarker || count > 0 && lastID == "" && entry == lastEntry) {
+			continue
+		}
+		if count == limit {
+			out.IsTruncated = true
+			if lastID == "" {
+				out.NextKeyMarker, out.NextUploadIDMarker = encode(lastEntry), ""
+			}
+			break
+		}
+		if rolled {
+			out.CommonPrefixes = append(out.CommonPrefixes, commonPrefixXML{Prefix: encode(entry)})
+			lastEntry, lastID = entry, ""
+		} else {
+			out.Uploads = append(out.Uploads, uploadXML{
+				Key: encode(u.key), UploadID: u.id, Initiator: ownerXML{ID: ownerID}, Owner: ownerXML{ID: ownerID},
+				StorageClass: "STANDARD", Initiated: u.initiated.Format(timeFormat),
+			})
+			lastEntry, lastID = u.key, u.id
+			out.NextKeyMarker, out.NextUploadIDMarker = encode(u.key), u.id
+		}
+		count++
+	}
+	return &reply{body: out}, nil
+}
+
+type listPartsXML struct {
+	XMLName              xml.Name  `xml:"http://s3.amazonaws.com/doc/2006-03-01/ ListPartsResult"`
+	Bucket               string    `xml:"Bucket"`
+	Key                  string    `xml:"Key"`
+	UploadID             string    `xml:"UploadId"`
+	Initiator            ownerXML  `xml:"Initiator"`
+	Owner                ownerXML  `xml:"Owner"`
+	StorageClass         string    `xml:"StorageClass"`
+	PartNumberMarker     int       `xml:"PartNumberMarker"`
+	NextPartNumberMarker int       `xml:"NextPartNumberMarker,omitempty"`
+	MaxParts             int       `xml:"MaxParts"`
+	IsTruncated          bool      `xml:"IsTruncated"`
+	Parts                []partXML `xml:"Part"`
+}
+
+type partXML struct {
+	PartNumber   int    `xml:"PartNumber"`
+	LastModified string `xml:"LastModified"`
+	ETag         string `xml:"ETag"`
+	Size         int64  `xml:"Size"`
+}
+
+func (s *Server) listParts(c *call) (*reply, *apiError) {
+	limit, err := listLimit(c, "max-parts")
+	if err != nil {
+		return nil, err
+	}
+	marker := 0
+	if c.query.Has("part-number-marker") {
+		n, err := strconv.Atoi(c.query.Get("part-number-marker"))
+		if err != nil || n < 0 {
+			return nil, invalidArgument("Provided part-number-marker not an integer or within integer range")
+		}
+		marker = n
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	_, u, err := s.findUpload(c)
+	if err != nil {
+		return nil, err
+	}
+	out := listPartsXML{
+		Bucket: c.bucket, Key: c.key, UploadID: u.id, Initiator: ownerXML{ID: ownerID}, Owner: ownerXML{ID: ownerID},
+		StorageClass: "STANDARD", PartNumberMarker: marker, MaxParts: limit,
+	}
+	for _, n := range slices.Sorted(maps.Keys(u.parts)) {
+		if n <= marker {
+			continue
+		}
+		if len(out.Parts) == limit {
+			out.IsTruncated = true
+			out.NextPartNumberMarker = out.Parts[limit-1].PartNumber
+			break
+		}
+		p := u.parts[n]
+		out.Parts = append(out.Parts, partXML{PartNumber: n, LastModified: p.LastModified.Format(timeFormat), ETag: p.ETag, Size: p.Size})
+	}
+	return &reply{body: out}, nil
+}
+
+// CreateMultipartUpload starts an upload of key in the named bucket, as CreateMultipartUpload does without
+// Object Lock headers, and returns it.
+func (s *Server) CreateMultipartUpload(bucketName, key string) (MultipartUpload, error) {
+	if key == "" {
+		return MultipartUpload{}, errors.New("squall: an upload needs a key")
+	}
+	if len(key) > maxKeyBytes {
+		return MultipartUpload{}, keyTooLong()
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	b, ok := s.buckets[bucketName]
+	if !ok {
+		return MultipartUpload{}, noSuchBucket(bucketName)
+	}
+	return s.startUpload(b, key, lockRequest{}).export(bucketName), nil
+}
+
+// MultipartUploads returns the named bucket's in-progress uploads in ListMultipartUploads order.
+func (s *Server) MultipartUploads(bucketName string) []MultipartUpload {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	b, ok := s.buckets[bucketName]
+	if !ok {
+		return nil
+	}
+	uploads := sortedUploads(b)
+	out := make([]MultipartUpload, 0, len(uploads))
+	for _, u := range uploads {
+		out = append(out, u.export(bucketName))
+	}
+	return out
+}
diff --git a/aws/s3/objectlock.go b/aws/s3/objectlock.go
new file mode 100644
index 0000000..e54ded9
--- /dev/null
+++ b/aws/s3/objectlock.go
@@ -0,0 +1,485 @@
+package s3
+
+import (
+	"encoding/xml"
+	"errors"
+	"fmt"
+	"net/http"
+	"slices"
+	"strings"
+	"time"
+
+	"github.com/parallelworks/squall/access"
+)
+
+const (
+	lockGovernance = "GOVERNANCE"
+	lockCompliance = "COMPLIANCE"
+
+	legalHoldOn  = "ON"
+	legalHoldOff = "OFF"
+
+	governanceOverride = "s3:BypassGovernanceRetention"
+
+	maxRetentionDays  = 36500
+	maxRetentionYears = 100
+)
+
+// DefaultRetention is the retention an Object Lock bucket gives each new version: a mode and Days or Years.
+type DefaultRetention struct {
+	// Mode is GOVERNANCE or COMPLIANCE.
+	Mode  string
+	Days  int
+	Years int
+}
+
+func (d *DefaultRetention) valid() error {
+	if d.Mode != lockGovernance && d.Mode != lockCompliance {
+		return fmt.Errorf("squall: retention mode %q is not one of GOVERNANCE and COMPLIANCE", d.Mode)
+	}
+	if (d.Days == 0) == (d.Years == 0) {
+		return errors.New("squall: a default retention needs one of Days and Years")
+	}
+	if d.Days < 0 || d.Years < 0 || d.Days > maxRetentionDays || d.Years > maxRetentionYears {
+		return errors.New("squall: a default retention runs from 1 to 36500 days or 1 to 100 years")
+	}
+	return nil
+}
+
+// objectLock is a version's retention and legal hold.
+type objectLock struct {
+	mode  string
+	until time.Time
+	// hold is "" for a version never given a legal hold, "ON" or "OFF".
+	hold string
+}
+
+// retains reports whether the lock's retention period is still running at now.
+func (l objectLock) retains(now time.Time) bool {
+	return l.mode != "" && l.until.After(now)
+}
+
+func objectLocked() *apiError {
+	return &apiError{Status: http.StatusForbidden, Code: "AccessDenied", Message: "Access Denied because object protected by object lock."}
+}
+
+func missingLockConfiguration() *apiError {
+	return &apiError{Status: http.StatusBadRequest, Code: "InvalidRequest", Message: "Bucket is missing Object Lock Configuration"}
+}
+
+func noSuchObjectLock() *apiError {
+	return &apiError{Status: http.StatusNotFound, Code: "NoSuchObjectLockConfiguration", Message: "The specified object does not have a ObjectLock configuration"}
+}
+
+// retainUntil keeps a retain-until date to the millisecond, the precision S3's timestamps carry.
+func retainUntil(t time.Time) time.Time { return t.UTC().Truncate(time.Millisecond) }
+
+// lockChecksumMissing refuses a write with Object Lock parameters and no checksum; what is "Put Object" or "Put Part".
+func lockChecksumMissing(what string) *apiError {
+	return &apiError{
+		Status: http.StatusBadRequest, Code: "InvalidRequest",
+		Message: "Content-MD5 OR x-amz-checksum- HTTP header is required for " + what + " requests with Object Lock parameters",
+	}
+}
+
+func checksumMissing() *apiError {
+	return &apiError{Status: http.StatusBadRequest, Code: "InvalidRequest", Message: "Missing required header for this request: Content-MD5"}
+}
+
+// decodeRoot reads an XML request body into v, which must be the root element named root.
+func decodeRoot(c *call, root string, v any) *apiError {
+	if !hasChecksum(c.r.Header) {
+		return checksumMissing()
+	}
+	return decodeRootXML(c.body, root, v)
+}
+
+func retainInPast() *apiError {
+	return invalidArgument("The retain until date must be in the future!")
+}
+
+// protects returns why v may not be deleted now, or nil. The caller holds s.mu.
+func (s *Server) protects(v *version, bypass bool, method, resource string) *apiError {
+	if !v.deleteMarker && v.lock.hold == legalHoldOn {
+		return objectLocked()
+	}
+	return s.retentionProtects(v, bypass, method, resource)
+}
+
+// retentionProtects is protects without the legal hold. The caller holds s.mu.
+func (s *Server) retentionProtects(v *version, bypass bool, method, resource string) *apiError {
+	if v.deleteMarker || !v.lock.retains(s.now()) {
+		return nil
+	}
+	if v.lock.mode == lockCompliance || !bypass {
+		return objectLocked()
+	}
+	if s.policy.Check(access.Call{Method: method, Resource: resource, Needs: []string{governanceOverride}}) != nil {
+		return notAuthorized(governanceOverride, resource)
+	}
+	return nil
+}
+
+// lockRequest is the Object Lock a PutObject or CreateMultipartUpload asks for in its headers.
+type lockRequest struct {
+	mode  string
+	until time.Time
+	hold  string
+}
+
+func (l lockRequest) any() bool { return l.mode != "" || l.hold != "" }
+
+// lockHeaders reads the x-amz-object-lock headers of a request that writes an object.
+func (s *Server) lockHeaders(h http.Header) (lockRequest, *apiError) {
+	for name := range h {
+		known := func(n string) bool { return strings.EqualFold(n, name) }
+		if hasPrefixFold(name, "X-Amz-Object-Lock-") &&
+			!slices.ContainsFunc([]string{"X-Amz-Object-Lock-Mode", "X-Amz-Object-Lock-Retain-Until-Date", "X-Amz-Object-Lock-Legal-Hold"}, known) {
+			return lockRequest{}, invalidArgument("squall: the %s header is not supported", name)
+		}
+	}
+	var l lockRequest
+	mode, until := h.Get("X-Amz-Object-Lock-Mode"), h.Get("X-Amz-Object-Lock-Retain-Until-Date")
+	if (mode == "") != (until == "") {
+		return lockRequest{}, invalidArgument("x-amz-object-lock-retain-until-date and x-amz-object-lock-mode must both be supplied")
+	}
+	if mode != "" {
+		if mode != lockGovernance && mode != lockCompliance {
+			return lockRequest{}, invalidArgument("Unknown wormMode directive.")
+		}
+		t, err := time.Parse(time.RFC3339, until)
+		if err != nil {
+			return lockRequest{}, invalidArgument("The retain until date is not a valid date.")
+		}
+		t = retainUntil(t)
+		if !t.After(s.now()) {
+			return lockRequest{}, retainInPast()
+		}
+		l.mode, l.until = mode, t
+	}
+	switch hold := h.Get("X-Amz-Object-Lock-Legal-Hold"); hold {
+	case "", legalHoldOn, legalHoldOff:
+		l.hold = hold
+	default:
+		return lockRequest{}, invalidArgument("Legal Hold must be either of 'ON' or 'OFF'")
+	}
+	return l, nil
+}
+
+// checkLockable refuses a lock on a bucket without Object Lock.
+func checkLockable(b *bucket, l lockRequest) *apiError {
+	if l.any() && !b.ObjectLock {
+		return missingLockConfiguration()
+	}
+	return nil
+}
+
+// applyLock gives a new version l, or the bucket's default retention when l names none. The caller holds s.mu.
+func (s *Server) applyLock(b *bucket, v *version, l lockRequest) {
+	switch {
+	case l.mode != "":
+		v.lock.mode, v.lock.until = l.mode, l.until
+	case b.DefaultRetention != nil:
+		d := b.DefaultRetention
+		v.lock.mode, v.lock.until = d.Mode, retainUntil(v.modified.AddDate(d.Years, 0, d.Days))
+	}
+	v.lock.hold = l.hold
+}
+
+// hasChecksum reports whether a request carries Content-MD5 or a checksum, in a header or a trailer.
+func hasChecksum(h http.Header) bool {
+	if h.Get("Content-MD5") != "" {
+		return true
+	}
+	trailer := strings.ToLower(h.Get("X-Amz-Trailer"))
+	for _, name := range checksumHeaders {
+		if h.Get(name) != "" || strings.Contains(trailer, strings.ToLower(name)) {
+			return true
+		}
+	}
+	return false
+}
+
+// checksumHeaders are the headers that carry a checksum's value.
+var checksumHeaders = []string{
+	"X-Amz-Checksum-Crc32", "X-Amz-Checksum-Crc32c", "X-Amz-Checksum-Crc64nvme", "X-Amz-Checksum-Sha1", "X-Amz-Checksum-Sha256",
+	"X-Amz-Checksum-Sha512", "X-Amz-Checksum-Md5", "X-Amz-Checksum-Xxhash64", "X-Amz-Checksum-Xxhash3", "X-Amz-Checksum-Xxhash128",
+}
+
+func hasPrefixFold(s, prefix string) bool {
+	return len(s) >= len(prefix) && strings.EqualFold(s[:len(prefix)], prefix)
+}
+
+type objectLockConfigurationXML struct {
+	XMLName           xml.Name
+	ObjectLockEnabled string             `xml:"ObjectLockEnabled,omitempty"`
+	Rule              *objectLockRuleXML `xml:"Rule,omitempty"`
+}
+
+type objectLockRuleXML struct {
+	DefaultRetention *defaultRetentionXML `xml:"DefaultRetention"`
+}
+
+type defaultRetentionXML struct {
+	Mode             string    `xml:"Mode,omitempty"`
+	Days             *int      `xml:"Days,omitempty"`
+	Years            *int      `xml:"Years,omitempty"`
+	DefaultEventHold *struct{} `xml:"DefaultEventHold,omitempty"`
+}
+
+func (s *Server) getObjectLockConfiguration(c *call) (*reply, *apiError) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	b, err := s.lookup(c)
+	if err != nil {
+		return nil, err
+	}
+	if !b.ObjectLock {
+		return nil, &apiError{
+			Status: http.StatusNotFound, Code: "ObjectLockConfigurationNotFoundError",
+			Message: "Object Lock configuration does not exist for this bucket", Bucket: c.bucket,
+		}
+	}
+	out := objectLockConfigurationXML{XMLName: xml.Name{Space: namespace, Local: "ObjectLockConfiguration"}, ObjectLockEnabled: "Enabled"}
+	if d := b.DefaultRetention; d != nil {
+		r := &defaultRetentionXML{Mode: d.Mode}
+		if d.Days > 0 {
+			r.Days = &d.Days
+		} else {
+			r.Years = &d.Years
+		}
+		out.Rule = &objectLockRuleXML{DefaultRetention: r}
+	}
+	return &reply{body: out}, nil
+}
+
+func (s *Server) putObjectLockConfiguration(c *call) (*reply, *apiError) {
+	var in objectLockConfigurationXML
+	if err := decodeRoot(c, "ObjectLockConfiguration", &in); err != nil {
+		return nil, err
+	}
+	if in.ObjectLockEnabled != "Enabled" {
+		return nil, malformedXML()
+	}
+	var retention *DefaultRetention
+	if in.Rule != nil {
+		r := in.Rule.DefaultRetention
+		if r == nil || r.Mode != lockGovernance && r.Mode != lockCompliance || (r.Days == nil) == (r.Years == nil) {
+			return nil, malformedXML()
+		}
+		if r.DefaultEventHold != nil {
+			return nil, invalidArgument("squall: default event holds are not supported")
+		}
+		retention = &DefaultRetention{Mode: r.Mode}
+		if r.Days != nil {
+			retention.Days = *r.Days
+		} else {
+			retention.Years = *r.Years
+		}
+		if retention.Days < 0 || retention.Years < 0 || retention.Days+retention.Years == 0 {
+			return nil, invalidArgument("Default retention period must be a positive integer value.")
+		}
+		if retention.Days > maxRetentionDays || retention.Years > maxRetentionYears {
+			return nil, invalidArgument("Default retention period is too large.")
+		}
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	b, err := s.lookup(c)
+	if err != nil {
+		return nil, err
+	}
+	if !b.ObjectLock && b.Versioning != versioningEnabled {
+		return nil, &apiError{
+			Status: http.StatusConflict, Code: "InvalidBucketState",
+			Message: "Versioning must be 'Enabled' on the bucket to apply a Object Lock configuration", Bucket: c.bucket,
+		}
+	}
+	b.ObjectLock, b.DefaultRetention = true, retention
+	return &reply{}, nil
+}
+
+// lockedVersion returns the version a retention or legal hold call names, in a bucket with Object Lock. The caller holds s.mu.
+func (s *Server) lockedVersion(c *call) (*version, *apiError) {
+	id := c.query.Get("versionId")
+	if c.query.Has("versionId") {
+		if _, ok := validVersionID(id); !ok {
+			return nil, invalidVersion()
+		}
+	}
+	b, err := s.lookup(c)
+	if err != nil {
+		return nil, err
+	}
+	if !b.ObjectLock {
+		return nil, missingLockConfiguration()
+	}
+	return findVersion(b, c.key, id)
+}
+
+// findVersion returns version id of key, or its newest with no id. The caller holds s.mu.
+func findVersion(b *bucket, key, id string) (*version, *apiError) {
+	versions := b.objects[key]
+	if id == "" {
+		if len(versions) == 0 || versions[0].deleteMarker {
+			return nil, &apiError{Status: http.StatusNotFound, Code: "NoSuchKey", Message: "The specified key does not exist."}
+		}
+		return versions[0], nil
+	}
+	i := slices.IndexFunc(versions, func(v *version) bool { return v.id == id })
+	if i < 0 {
+		return nil, &apiError{Status: http.StatusNotFound, Code: "NoSuchVersion", Message: "The specified version does not exist."}
+	}
+	if versions[i].deleteMarker {
+		return nil, &apiError{Status: http.StatusMethodNotAllowed, Code: "MethodNotAllowed", Message: "The specified method is not allowed against this resource."}
+	}
+	return versions[i], nil
+}
+
+type retentionXML struct {
+	XMLName           xml.Name
+	Mode              string    `xml:"Mode,omitempty"`
+	RetainUntilDate   string    `xml:"RetainUntilDate,omitempty"`
+	EventHold         *string   `xml:"EventHold,omitempty"`
+	EventHoldDuration *struct{} `xml:"EventHoldDuration,omitempty"`
+}
+
+func (s *Server) getObjectRetention(c *call) (*reply, *apiError) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	v, err := s.lockedVersion(c)
+	if err != nil {
+		return nil, err
+	}
+	if v.lock.mode == "" {
+		return nil, noSuchObjectLock()
+	}
+	return &reply{body: retentionXML{
+		XMLName: xml.Name{Space: namespace, Local: "Retention"}, Mode: v.lock.mode, RetainUntilDate: v.lock.until.Format(timeFormat),
+	}}, nil
+}
+
+func (s *Server) putObjectRetention(c *call) (*reply, *apiError) {
+	var in retentionXML
+	if err := decodeRoot(c, "Retention", &in); err != nil {
+		return nil, err
+	}
+	if in.EventHold != nil || in.EventHoldDuration != nil {
+		return nil, invalidArgument("squall: event holds are not supported")
+	}
+	if (in.Mode == "") != (in.RetainUntilDate == "") || in.Mode != "" && in.Mode != lockGovernance && in.Mode != lockCompliance {
+		return nil, malformedXML()
+	}
+	var until time.Time
+	if in.RetainUntilDate != "" {
+		t, err := time.Parse(time.RFC3339, in.RetainUntilDate)
+		if err != nil {
+			return nil, malformedXML()
+		}
+		until = retainUntil(t)
+		if !until.After(s.now()) {
+			return nil, retainInPast()
+		}
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	v, err := s.lockedVersion(c)
+	if err != nil {
+		return nil, err
+	}
+	if in.Mode != v.lock.mode || until.Before(v.lock.until) {
+		if err := s.retentionProtects(v, bypassing(c), "s3.PutObjectRetention", c.resource()); err != nil {
+			return nil, err
+		}
+	}
+	v.lock.mode, v.lock.until = in.Mode, until
+	return &reply{}, nil
+}
+
+type legalHoldXML struct {
+	XMLName xml.Name
+	Status  string `xml:"Status"`
+}
+
+func (s *Server) getObjectLegalHold(c *call) (*reply, *apiError) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	v, err := s.lockedVersion(c)
+	if err != nil {
+		return nil, err
+	}
+	if v.lock.hold == "" {
+		return nil, noSuchObjectLock()
+	}
+	return &reply{body: legalHoldXML{XMLName: xml.Name{Space: namespace, Local: "LegalHold"}, Status: v.lock.hold}}, nil
+}
+
+func (s *Server) putObjectLegalHold(c *call) (*reply, *apiError) {
+	var in legalHoldXML
+	if err := decodeRoot(c, "LegalHold", &in); err != nil {
+		return nil, err
+	}
+	if in.Status != legalHoldOn && in.Status != legalHoldOff {
+		return nil, malformedXML()
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	v, err := s.lockedVersion(c)
+	if err != nil {
+		return nil, err
+	}
+	v.lock.hold = in.Status
+	return &reply{}, nil
+}
+
+// testVersion returns the version a test helper names, in a bucket with Object Lock. The caller holds s.mu.
+func (s *Server) testVersion(bucketName, key, versionID string) (*version, error) {
+	b, ok := s.buckets[bucketName]
+	if !ok {
+		return nil, noSuchBucket(bucketName)
+	}
+	if !b.ObjectLock {
+		return nil, missingLockConfiguration()
+	}
+	v, err := findVersion(b, key, versionID)
+	if err != nil {
+		return nil, err
+	}
+	return v, nil
+}
+
+// SetRetention gives a version of key in the named bucket a retention period in mode, GOVERNANCE or
+// COMPLIANCE, until until, or with mode "" removes it, whatever retention the version has. An empty
+// versionID names the newest version. The bucket needs Object Lock.
+func (s *Server) SetRetention(bucketName, key, versionID, mode string, until time.Time) error {
+	if mode != "" && mode != lockGovernance && mode != lockCompliance {
+		return fmt.Errorf("squall: retention mode %q is not one of GOVERNANCE and COMPLIANCE", mode)
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	v, err := s.testVersion(bucketName, key, versionID)
+	if err != nil {
+		return err
+	}
+	v.lock.mode, v.lock.until = mode, time.Time{}
+	if mode != "" {
+		v.lock.until = retainUntil(until)
+	}
+	return nil
+}
+
+// SetLegalHold places or removes a legal hold on a version of key in the named bucket. An empty versionID
+// names the newest version. The bucket needs Object Lock.
+func (s *Server) SetLegalHold(bucketName, key, versionID string, on bool) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	v, err := s.testVersion(bucketName, key, versionID)
+	if err != nil {
+		return err
+	}
+	v.lock.hold = legalHoldOff
+	if on {
+		v.lock.hold = legalHoldOn
+	}
+	return nil
+}
diff --git a/aws/s3/objects.go b/aws/s3/objects.go
index 3ac1425..3d947f8 100644
--- a/aws/s3/objects.go
+++ b/aws/s3/objects.go
@@ -40,6 +40,10 @@ type ObjectVersion struct {
 	// ETag is the quoted hex MD5 of the object's data; empty for a delete marker.
 	ETag         string
 	LastModified time.Time
+	// RetentionMode is GOVERNANCE or COMPLIANCE for a version given a retention period, which may have expired.
+	RetentionMode string
+	RetainUntil   time.Time
+	LegalHold     bool
 }
 
 type version struct {
@@ -50,10 +54,14 @@ type version struct {
 	size         int64
 	etag         string
 	modified     time.Time
+	lock         objectLock
 }
 
 func (v *version) export(key string, latest bool) ObjectVersion {
-	return ObjectVersion{Key: key, VersionID: v.id, IsLatest: latest, DeleteMarker: v.deleteMarker, Size: v.size, ETag: v.etag, LastModified: v.modified}
+	return ObjectVersion{
+		Key: key, VersionID: v.id, IsLatest: latest, DeleteMarker: v.deleteMarker, Size: v.size, ETag: v.etag, LastModified: v.modified,
+		RetentionMode: v.lock.mode, RetainUntil: v.lock.until, LegalHold: v.lock.hold == legalHoldOn,
+	}
 }
 
 // newVersion returns a version written now, "null" unless versioning is on. The caller holds s.mu.
@@ -108,13 +116,16 @@ type deleted struct {
 }
 
 // remove deletes version id of key, or with no id deletes key as the bucket's versioning says. The caller holds s.mu.
-func (s *Server) remove(b *bucket, key, id string) deleted {
+func (s *Server) remove(b *bucket, key, id string, bypass bool, method string) (deleted, *apiError) {
 	if id != "" {
 		out := deleted{Key: key, VersionID: id}
 		versions := b.objects[key]
 		i := slices.IndexFunc(versions, func(v *version) bool { return v.id == id })
 		if i < 0 {
-			return out
+			return out, nil
+		}
+		if err := s.protects(versions[i], bypass, method, b.Name+"/"+key); err != nil {
+			return deleted{}, err
 		}
 		if versions[i].deleteMarker {
 			out.DeleteMarker, out.DeleteMarkerVersionID = true, id
@@ -125,16 +136,16 @@ func (s *Server) remove(b *bucket, key, id string) deleted {
 		} else {
 			b.objects[key] = versions
 		}
-		return out
+		return out, nil
 	}
 	if b.Versioning == "" {
 		b.forget(key)
-		return deleted{Key: key}
+		return deleted{Key: key}, nil
 	}
 	marker := s.newVersion(b)
 	marker.deleteMarker = true
 	b.write(key, marker)
-	return deleted{Key: key, DeleteMarker: true, DeleteMarkerVersionID: marker.id}
+	return deleted{Key: key, DeleteMarker: true, DeleteMarkerVersionID: marker.id}, nil
 }
 
 func (s *Server) putObject(c *call) (*reply, *apiError) {
@@ -147,10 +158,12 @@ func (s *Server) putObject(c *call) (*reply, *apiError) {
 	if acl := c.r.Header.Get("x-amz-acl"); acl != "" && acl != "private" && acl != "bucket-owner-full-control" {
 		return nil, &apiError{Status: http.StatusBadRequest, Code: "AccessControlListNotSupported", Message: "The bucket does not allow ACLs"}
 	}
-	for h := range c.r.Header {
-		if strings.HasPrefix(strings.ToLower(h), "x-amz-object-lock-") || strings.EqualFold(h, "x-amz-tagging") {
-			return nil, invalidArgument("squall: the %s header is not supported", h)
-		}
+	if c.r.Header.Get("X-Amz-Tagging") != "" {
+		return nil, invalidArgument("squall: the x-amz-tagging header is not supported")
+	}
+	lock, err := s.lockHeaders(c.r.Header)
+	if err != nil {
+		return nil, err
 	}
 	s.mu.Lock()
 	defer s.mu.Unlock()
@@ -158,7 +171,14 @@ func (s *Server) putObject(c *call) (*reply, *apiError) {
 	if err != nil {
 		return nil, err
 	}
+	if err := checkLockable(b, lock); err != nil {
+		return nil, err
+	}
+	if (lock.mode != "" || b.DefaultRetention != nil) && !hasChecksum(c.r.Header) {
+		return nil, lockChecksumMissing("Put Object")
+	}
 	v := s.storeObject(b, c.key, c.body)
+	s.applyLock(b, v, lock)
 	header := map[string]string{"ETag": v.etag}
 	if b.Versioning != "" {
 		header["x-amz-version-id"] = v.id
@@ -168,13 +188,27 @@ func (s *Server) putObject(c *call) (*reply, *apiError) {
 
 // storeObject writes data as key's newest version. The caller holds s.mu.
 func (s *Server) storeObject(b *bucket, key string, data []byte) *version {
-	sum := md5.Sum(data) //nolint:gosec // an ETag, not a security check
+	return s.storeVersion(b, key, int64(len(data)), etagOf(data))
+}
+
+// storeVersion writes a version of size bytes with etag as key's newest. The caller holds s.mu.
+func (s *Server) storeVersion(b *bucket, key string, size int64, etag string) *version {
 	v := s.newVersion(b)
-	v.size, v.etag = int64(len(data)), `"`+hex.EncodeToString(sum[:])+`"`
+	v.size, v.etag = size, etag
 	b.write(key, v)
 	return v
 }
 
+func etagOf(data []byte) string {
+	sum := md5.Sum(data) //nolint:gosec // an ETag, not a security check
+	return `"` + hex.EncodeToString(sum[:]) + `"`
+}
+
+// bypassing reports whether c asks to bypass governance retention.
+func bypassing(c *call) bool {
+	return strings.EqualFold(c.r.Header.Get("X-Amz-Bypass-Governance-Retention"), "true")
+}
+
 func (s *Server) deleteObject(c *call) (*reply, *apiError) {
 	id := c.query.Get("versionId")
 	if c.query.Has("versionId") {
@@ -188,7 +222,10 @@ func (s *Server) deleteObject(c *call) (*reply, *apiError) {
 	if err != nil {
 		return nil, err
 	}
-	d := s.remove(b, c.key, id)
+	d, err := s.remove(b, c.key, id, bypassing(c), "s3.DeleteObject")
+	if err != nil {
+		return nil, err
+	}
 	header := map[string]string{}
 	if d.DeleteMarker {
 		header["x-amz-delete-marker"] = "true"
@@ -253,7 +290,11 @@ func (s *Server) deleteObjects(c *call) (*reply, *apiError) {
 			out.Errors = append(out.Errors, deleteErrorXML{o.Key, o.VersionID, "AccessDenied", "Access Denied"})
 			continue
 		}
-		d := s.remove(b, o.Key, o.VersionID)
+		d, err := s.remove(b, o.Key, o.VersionID, bypassing(c), "s3.DeleteObjects")
+		if err != nil {
+			out.Errors = append(out.Errors, deleteErrorXML{o.Key, o.VersionID, err.Code, err.Message})
+			continue
+		}
 		if !in.Quiet {
 			out.Deleted = append(out.Deleted, d)
 		}
@@ -400,13 +441,9 @@ func (s *Server) listObjectsV2(c *call) (*reply, *apiError) {
 	if err != nil {
 		return nil, err
 	}
-	encode := func(v string) string { return v }
-	switch c.query.Get("encoding-type") {
-	case "":
-	case "url":
-		encode = func(v string) string { return strings.ReplaceAll(url.QueryEscape(v), "%2F", "/") }
-	default:
-		return nil, invalidArgument("Invalid Encoding Method specified in Request")
+	encode, err := encodingOf(c)
+	if err != nil {
+		return nil, err
 	}
 	var fetchOwner bool
 	switch c.query.Get("fetch-owner") {
@@ -479,6 +516,17 @@ func (s *Server) listObjectsV2(c *call) (*reply, *apiError) {
 	return &reply{body: out}, nil
 }
 
+// encodingOf reads a listing's encoding-type into how it encodes keys and prefixes.
+func encodingOf(c *call) (func(string) string, *apiError) {
+	switch c.query.Get("encoding-type") {
+	case "":
+		return func(v string) string { return v }, nil
+	case "url":
+		return func(v string) string { return strings.ReplaceAll(url.QueryEscape(v), "%2F", "/") }, nil
+	}
+	return nil, invalidArgument("Invalid Encoding Method specified in Request")
+}
+
 func entryXML(v ObjectVersion) versionXML {
 	e := versionXML{
 		XMLName: xml.Name{Local: "Version"}, Key: v.Key, VersionID: v.VersionID, IsLatest: v.IsLatest,
diff --git a/aws/s3/permissions.go b/aws/s3/permissions.go
index b4dd498..1d229ef 100644
--- a/aws/s3/permissions.go
+++ b/aws/s3/permissions.go
@@ -3,6 +3,7 @@ package s3
 import (
 	"fmt"
 	"net/http"
+	"strings"
 
 	"github.com/parallelworks/squall/access"
 )
@@ -23,6 +24,14 @@ var actions = map[string]string{
 	"PutBucketCors":       "s3:PutBucketCORS",
 	"ListObjectVersions":  "s3:ListBucketVersions",
 	"ListObjectsV2":       "s3:ListBucket",
+
+	"GetObjectLockConfiguration": "s3:GetBucketObjectLockConfiguration",
+	"PutObjectLockConfiguration": "s3:PutBucketObjectLockConfiguration",
+	"CreateMultipartUpload":      "s3:PutObject",
+	"UploadPart":                 "s3:PutObject",
+	"CompleteMultipartUpload":    "s3:PutObject",
+	"ListMultipartUploads":       "s3:ListBucketMultipartUploads",
+	"ListParts":                  "s3:ListMultipartUploadParts",
 }
 
 // needs returns the IAM actions a call to op needs, or nil for DeleteObjects,
@@ -43,10 +52,19 @@ func needs(op string, c *call) []string {
 		if c.r.Header.Get("x-amz-object-ownership") != "" {
 			perms = append(perms, "s3:PutBucketOwnershipControls")
 		}
-	case "PutObject":
+		if strings.EqualFold(c.r.Header.Get("x-amz-bucket-object-lock-enabled"), "true") {
+			perms = append(perms, "s3:PutBucketObjectLockConfiguration", "s3:PutBucketVersioning")
+		}
+	case "PutObject", "CreateMultipartUpload":
 		if c.r.Header.Get("x-amz-acl") != "" {
 			perms = append(perms, "s3:PutObjectAcl")
 		}
+		if c.r.Header.Get("X-Amz-Object-Lock-Mode") != "" || c.r.Header.Get("X-Amz-Object-Lock-Retain-Until-Date") != "" {
+			perms = append(perms, "s3:PutObjectRetention")
+		}
+		if c.r.Header.Get("X-Amz-Object-Lock-Legal-Hold") != "" {
+			perms = append(perms, "s3:PutObjectLegalHold")
+		}
 	}
 	return perms
 }
diff --git a/aws/s3/server.go b/aws/s3/server.go
index fdfaf69..a210ec7 100644
--- a/aws/s3/server.go
+++ b/aws/s3/server.go
@@ -14,10 +14,17 @@
 // It covers buckets (CreateBucket, HeadBucket, ListBuckets, DeleteBucket),
 // their tags (PutBucketTagging, GetBucketTagging, DeleteBucketTagging),
 // versioning (PutBucketVersioning, GetBucketVersioning), policy
-// (PutBucketPolicy) and CORS rules (PutBucketCors), and the objects in them
-// (PutObject, DeleteObject, DeleteObjects, ListObjectsV2, ListObjectVersions), with the real
-// API's error codes. Other operations answer 404 with the code
-// NotImplemented. Where it differs from S3:
+// (PutBucketPolicy), CORS rules (PutBucketCors), Object Lock configuration
+// (PutObjectLockConfiguration, GetObjectLockConfiguration) and request
+// payment (PutBucketRequestPayment, GetBucketRequestPayment), the objects in
+// them (PutObject, DeleteObject, DeleteObjects, ListObjectsV2,
+// ListObjectVersions) with their retention (PutObjectRetention,
+// GetObjectRetention) and legal holds (PutObjectLegalHold,
+// GetObjectLegalHold), and multipart uploads (CreateMultipartUpload,
+// UploadPart, CompleteMultipartUpload, AbortMultipartUpload,
+// ListMultipartUploads, ListParts), with the real API's error codes. Other
+// operations answer 404 with the code NotImplemented. Where it differs from
+// S3:
 //
 //   - Signatures are not checked. The region comes from the credential scope
 //     of the SigV4 Authorization header (or the X-Amz-Credential query
@@ -38,8 +45,8 @@
 //     IllegalLocationConstraintException. Creating a bucket the account
 //     already owns succeeds in us-east-1 and fails with
 //     BucketAlreadyOwnedByYou elsewhere. Object ownership is always bucket
-//     owner enforced: an ACL other than private fails, and object lock and
-//     directory buckets are not supported.
+//     owner enforced: an ACL other than private fails, and directory
+//     buckets are not supported.
 //   - ListBuckets returns the account's buckets ordered by name, in every
 //     region, and filters by prefix and bucket-region, which must be the
 //     request's region. As in S3, a request with any of its parameters pages,
@@ -51,10 +58,79 @@
 //     verified, and bodies sent with aws-chunked encoding are decoded without
 //     checking their chunk signatures. Version IDs are 32 hex digits, and
 //     "null" for an object written while versioning is off or suspended.
-//     Multipart uploads, copies, storage classes, object tags, lifecycle and
-//     replication are not supported. S3 documents in-progress multipart
-//     uploads as blocking DeleteBucket only for directory buckets, so for a
-//     general purpose bucket only versions and delete markers do.
+//     Copies, storage classes, object tags, lifecycle and replication are
+//     not supported.
+//   - A multipart upload's parts record their size and ETag like objects,
+//     and are up to 16 MiB each. CompleteMultipartUpload checks the parts
+//     are in ascending order, were uploaded with the ETags given, and are at
+//     least 5 MiB but for the last, and makes a version whose ETag is the MD5
+//     of the parts' MD5s and their count, the form S3's take, which it does
+//     not document. Completing an upload ends it, so a retry after a lost
+//     response fails with NoSuchUpload. Uploads of one key list in the order
+//     they began, and upload IDs are 32 hex digits. ListMultipartUploads
+//     takes S3's prefix, delimiter, markers and max-uploads, and counts
+//     common prefixes toward a page as ListObjectsV2 does, which is
+//     inferred. As S3 documents, it leaves out a common prefix not after
+//     key-marker, even one holding later keys, where ListObjectsV2's
+//     start-after keeps it. NextKeyMarker and NextUploadIdMarker name the
+//     last upload listed, as in S3's examples; a truncated page that ends
+//     on a common prefix names that prefix instead, which is inferred.
+//     Like max-keys, max-uploads and max-parts above 1000 return
+//     1000; one below 1 fails with 400 InvalidArgument, which is inferred.
+//     A part's checksums are not kept, and ListParts and
+//     ListMultipartUploads report no checksum algorithm; UploadPartCopy and
+//     the If-Match and If-None-Match headers are not supported. Parts are
+//     stored data, which S3 bills until the upload is completed or
+//     aborted, so [Server.MultipartUploads] reports them with their sizes.
+//     S3 documents in-progress uploads as blocking DeleteBucket only for
+//     directory buckets, so deleting a general purpose bucket discards its
+//     uploads, and only versions and delete markers block it.
+//   - Object Lock follows S3's rules. Creating a bucket with
+//     x-amz-bucket-object-lock-enabled turns on versioning, which can then
+//     not be suspended (409 InvalidBucketState, whose message is inferred);
+//     PutObjectLockConfiguration enables Object Lock on a bucket whose
+//     versioning is Enabled, and on one whose versioning is not fails with
+//     409 InvalidBucketState, inferred. A default retention needs a mode and
+//     one of Days and Years (MalformedXML otherwise), and gives each new
+//     version, from PutObject or CompleteMultipartUpload, a retention until
+//     that long after the version is made. PutObject and
+//     CreateMultipartUpload take a mode and retain-until date together and
+//     a legal hold, on a bucket with Object Lock only (400 InvalidRequest,
+//     inferred). PutObject that sets a retention, explicitly or by default,
+//     needs Content-MD5 or a checksum, as S3 documents, and so does each
+//     UploadPart of an upload that will; the error message is inferred.
+//     PutObjectLockConfiguration, PutObjectRetention and PutObjectLegalHold
+//     need one too, as the SDKs always send, and a body under another root
+//     element fails with MalformedXML. Retention runs by the server's clock
+//     ([WithClock]), and retain-until dates are kept to the millisecond, the
+//     precision S3's timestamps carry.
+//     Deleting a version under an unexpired retention or a legal hold fails
+//     with 403 AccessDenied, key by key in DeleteObjects; its message is
+//     inferred from S3's answers. A COMPLIANCE retention never yields, and a
+//     GOVERNANCE one yields to x-amz-bypass-governance-retention: true from a
+//     caller [WithPolicy] grants s3:BypassGovernanceRetention. A delete
+//     without a version ID adds a delete marker, as S3 does, and DeleteBucket
+//     fails while any version remains. PutObjectRetention may extend a
+//     retention in the same mode; shortening it, changing its mode or
+//     removing it needs the same bypass, which a COMPLIANCE retention
+//     refuses. S3 documents the bypass for shortening and removal; that
+//     changing GOVERNANCE to COMPLIANCE needs it too is inferred from
+//     governance mode letting no one alter lock settings without special
+//     permissions. A version with no retention or no legal hold answers
+//     GetObjectRetention or GetObjectLegalHold with 404
+//     NoSuchObjectLockConfiguration, and a call naming a delete marker's
+//     version fails with 405 MethodNotAllowed; both are inferred, as are the
+//     InvalidArgument messages for a retain-until date in the past, an
+//     unknown mode or legal hold status, and a mode without a date. Event
+//     holds, variable retention and default event holds are rejected with
+//     400 InvalidArgument; the x-amz-bucket-object-lock-token header is
+//     ignored, as S3 no longer needs it. [Server.SetRetention] and
+//     [Server.SetLegalHold] lock a version directly, whatever its lock.
+//   - Request payment is recorded. There is one account, which owns every
+//     bucket it can reach, and S3 does not ask a bucket owner for
+//     x-amz-request-payer, so a Requester Pays bucket answers as any other
+//     and responses carry no x-amz-request-charged. Like the Object Lock
+//     PUTs, PutBucketRequestPayment needs Content-MD5 or a checksum.
 //   - DeleteBucket fails with BucketNotEmpty while any object version or
 //     delete marker remains. ListObjectVersions orders versions by key and
 //     then newest first, and its markers are positions, so a page may follow
@@ -69,9 +145,9 @@
 //     Owners are the account's canonical ID, without a display name.
 //   - Tags follow S3's rules: up to 50, keys of 1 to 128 characters, values
 //     of up to 256, Unicode letters, numbers, separators and "+-=._:/@", no
-//     duplicates and no "aws:" prefix. PutBucketTagging with an empty tag set removes
-//     the tags, so GetBucketTagging fails with NoSuchTagSet, as it does for
-//     a bucket never tagged. Bucket policies must be JSON and are recorded,
+//     duplicates and no "aws:" prefix. PutBucketTagging with an empty tag
+//     set removes the tags, so GetBucketTagging fails with NoSuchTagSet, as
+//     it does for a bucket never tagged. Bucket policies must be JSON and are recorded,
 //     not enforced; CORS rules are checked for an allowed method and origin
 //     and recorded.
 //   - Query parameters the server does not support are rejected with 400
@@ -81,9 +157,15 @@
 // [WithPolicy] grants the caller a set of IAM actions and refuses a call that
 // needs one it lacks with 403 AccessDenied, before any fault. Each operation
 // needs the action the S3 user guide lists for it, such as s3:ListBucket for
-// HeadBucket and ListObjectsV2; deleting a version needs s3:DeleteObjectVersion instead of
-// s3:DeleteObject, an ACL header on PutObject s3:PutObjectAcl, and an
-// object ownership header on CreateBucket s3:PutBucketOwnershipControls.
+// HeadBucket and ListObjectsV2; deleting a version needs
+// s3:DeleteObjectVersion instead of s3:DeleteObject, an ACL header on PutObject s3:PutObjectAcl, an object
+// ownership header on CreateBucket s3:PutBucketOwnershipControls, and
+// enabling Object Lock there s3:PutBucketObjectLockConfiguration and
+// s3:PutBucketVersioning. A retention header on PutObject or
+// CreateMultipartUpload needs s3:PutObjectRetention and a legal hold header
+// s3:PutObjectLegalHold. Bypassing a governance retention needs
+// s3:BypassGovernanceRetention, checked only when the bypass decides the
+// call, and refused with 403 AccessDenied naming it.
 // DeleteObjects is checked key by key, as S3 checks it: a key the policy may
 // not delete is an AccessDenied error in the result, and the rest are
 // deleted. Actions are not checked against the resources they act on or
@@ -148,6 +230,7 @@ type Server struct {
 	buckets     map[string]*bucket
 	foreign     map[string]string
 	nextVersion uint64
+	nextUpload  uint64
 }
 
 // Option configures a [Server].
@@ -216,33 +299,43 @@ var subresources = []string{
 // operations maps a method, scope and subresource to its operation; an empty one is not implemented.
 func (s *Server) operations() map[string]operation {
 	return map[string]operation{
-		"GET service":            {"ListBuckets", listBucketsParams, s.listBuckets},
-		"PUT bucket":             {"CreateBucket", nil, s.createBucket},
-		"HEAD bucket":            {"HeadBucket", nil, s.headBucket},
-		"DELETE bucket":          {"DeleteBucket", nil, s.deleteBucket},
-		"GET bucket?tagging":     {"GetBucketTagging", nil, s.getBucketTagging},
-		"PUT bucket?tagging":     {"PutBucketTagging", nil, s.putBucketTagging},
-		"DELETE bucket?tagging":  {"DeleteBucketTagging", nil, s.deleteBucketTagging},
-		"GET bucket?versioning":  {"GetBucketVersioning", nil, s.getBucketVersioning},
-		"PUT bucket?versioning":  {"PutBucketVersioning", nil, s.putBucketVersioning},
-		"PUT bucket?policy":      {"PutBucketPolicy", nil, s.putBucketPolicy},
-		"PUT bucket?cors":        {"PutBucketCors", nil, s.putBucketCors},
-		"GET bucket?versions":    {"ListObjectVersions", []string{"prefix", "key-marker", "version-id-marker", "max-keys"}, s.listObjectVersions},
-		"GET bucket?list-type":   {"ListObjectsV2", listObjectsV2Params, s.listObjectsV2},
-		"POST bucket?delete":     {"DeleteObjects", nil, s.deleteObjects},
-		"PUT object":             {"PutObject", nil, s.putObject},
-		"DELETE object":          {"DeleteObject", []string{"versionId"}, s.deleteObject},
-		"HEAD service":           {},
-		"PUT service":            {},
-		"DELETE service":         {},
-		"POST service":           {},
-		"GET bucket?location":    {},
-		"GET object":             {},
-		"HEAD object":            {},
-		"POST object?uploads":    {},
-		"GET bucket?uploads":     {},
-		"PUT object?partNumber":  {},
-		"DELETE object?uploadId": {},
+		"GET service":               {"ListBuckets", listBucketsParams, s.listBuckets},
+		"PUT bucket":                {"CreateBucket", nil, s.createBucket},
+		"HEAD bucket":               {"HeadBucket", nil, s.headBucket},
+		"DELETE bucket":             {"DeleteBucket", nil, s.deleteBucket},
+		"GET bucket?tagging":        {"GetBucketTagging", nil, s.getBucketTagging},
+		"PUT bucket?tagging":        {"PutBucketTagging", nil, s.putBucketTagging},
+		"DELETE bucket?tagging":     {"DeleteBucketTagging", nil, s.deleteBucketTagging},
+		"GET bucket?versioning":     {"GetBucketVersioning", nil, s.getBucketVersioning},
+		"PUT bucket?versioning":     {"PutBucketVersioning", nil, s.putBucketVersioning},
+		"PUT bucket?policy":         {"PutBucketPolicy", nil, s.putBucketPolicy},
+		"PUT bucket?cors":           {"PutBucketCors", nil, s.putBucketCors},
+		"GET bucket?versions":       {"ListObjectVersions", []string{"prefix", "key-marker", "version-id-marker", "max-keys"}, s.listObjectVersions},
+		"GET bucket?list-type":      {"ListObjectsV2", listObjectsV2Params, s.listObjectsV2},
+		"POST bucket?delete":        {"DeleteObjects", nil, s.deleteObjects},
+		"GET bucket?object-lock":    {"GetObjectLockConfiguration", nil, s.getObjectLockConfiguration},
+		"PUT bucket?object-lock":    {"PutObjectLockConfiguration", nil, s.putObjectLockConfiguration},
+		"GET bucket?requestPayment": {"GetBucketRequestPayment", nil, s.getBucketRequestPayment},
+		"PUT bucket?requestPayment": {"PutBucketRequestPayment", nil, s.putBucketRequestPayment},
+		"GET bucket?uploads":        {"ListMultipartUploads", listUploadsParams, s.listMultipartUploads},
+		"PUT object":                {"PutObject", nil, s.putObject},
+		"DELETE object":             {"DeleteObject", []string{"versionId"}, s.deleteObject},
+		"POST object?uploads":       {"CreateMultipartUpload", nil, s.createMultipartUpload},
+		"PUT object?partNumber":     {"UploadPart", []string{"uploadId"}, s.uploadPart},
+		"POST object?uploadId":      {"CompleteMultipartUpload", nil, s.completeMultipartUpload},
+		"DELETE object?uploadId":    {"AbortMultipartUpload", nil, s.abortMultipartUpload},
+		"GET object?uploadId":       {"ListParts", []string{"max-parts", "part-number-marker"}, s.listParts},
+		"GET object?retention":      {"GetObjectRetention", []string{"versionId"}, s.getObjectRetention},
+		"PUT object?retention":      {"PutObjectRetention", []string{"versionId"}, s.putObjectRetention},
+		"GET object?legal-hold":     {"GetObjectLegalHold", []string{"versionId"}, s.getObjectLegalHold},
+		"PUT object?legal-hold":     {"PutObjectLegalHold", []string{"versionId"}, s.putObjectLegalHold},
+		"HEAD service":              {},
+		"PUT service":               {},
+		"DELETE service":            {},
+		"POST service":              {},
+		"GET bucket?location":       {},
+		"GET object":                {},
+		"HEAD object":               {},
 	}
 }
 
@@ -330,7 +423,7 @@ func (s *Server) parse(w http.ResponseWriter, r *http.Request) (operation, *call
 		}
 	}
 	limit, tooLarge := int64(maxBody), "MaxMessageLengthExceeded"
-	if op.name == "PutObject" {
+	if op.name == "PutObject" || op.name == "UploadPart" {
 		limit, tooLarge = maxObject, "EntityTooLarge"
 	}
 	c.body, err = readBody(w, r, limit, tooLarge)
@@ -442,6 +535,18 @@ func (s *Server) requestID() string {
 	return fmt.Sprintf("%016X", s.requests.Add(1))
 }
 
+// decodeRootXML reads an XML request body into v, failing with MalformedXML unless its root element is root.
+func decodeRootXML(body []byte, root string, v any) *apiError {
+	var head struct{ XMLName xml.Name }
+	if err := decodeXML(body, &head); err != nil {
+		return err
+	}
+	if head.XMLName.Local != root {
+		return malformedXML()
+	}
+	return decodeXML(body, v)
+}
+
 // decodeXML reads an XML request body into v.
 func decodeXML(body []byte, v any) *apiError {
 	//nolint:gosec // v is a fixed struct and the body is bounded by maxBody
diff --git a/sdktest/aws/s3/multipart_test.go b/sdktest/aws/s3/multipart_test.go
new file mode 100644
index 0000000..9cc3f9b
--- /dev/null
+++ b/sdktest/aws/s3/multipart_test.go
@@ -0,0 +1,271 @@
+package s3_test
+
+import (
+	"bytes"
+	"crypto/md5"
+	"encoding/hex"
+	"fmt"
+	"net/http"
+	"slices"
+	"strings"
+	"testing"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/aws/aws-sdk-go-v2/service/s3/types"
+
+	"github.com/parallelworks/squall/access"
+	"github.com/parallelworks/squall/aws/s3"
+	"github.com/parallelworks/squall/fault"
+	"github.com/parallelworks/squall/sdktest/internal/sdkerr"
+)
+
+func startUpload(t *testing.T, c *awss3.Client, bucket, key string) string {
+	t.Helper()
+	out, err := c.CreateMultipartUpload(t.Context(), &awss3.CreateMultipartUploadInput{Bucket: aws.String(bucket), Key: aws.String(key)})
+	if err != nil {
+		t.Fatalf("create upload of %s: %v", key, err)
+	}
+	if aws.ToString(out.Bucket) != bucket || aws.ToString(out.Key) != key || aws.ToString(out.UploadId) == "" {
+		t.Fatalf("create upload = %+v", out)
+	}
+	return aws.ToString(out.UploadId)
+}
+
+func uploadPart(t *testing.T, c *awss3.Client, bucket, key, id string, n int32, data []byte) types.CompletedPart {
+	t.Helper()
+	out, err := c.UploadPart(t.Context(), &awss3.UploadPartInput{
+		Bucket: aws.String(bucket), Key: aws.String(key), UploadId: aws.String(id), PartNumber: aws.Int32(n), Body: bytes.NewReader(data),
+	})
+	if err != nil {
+		t.Fatalf("upload part %d: %v", n, err)
+	}
+	return types.CompletedPart{PartNumber: aws.Int32(n), ETag: out.ETag}
+}
+
+func complete(t *testing.T, c *awss3.Client, bucket, key, id string, parts ...types.CompletedPart) (*awss3.CompleteMultipartUploadOutput, error) {
+	t.Helper()
+	return c.CompleteMultipartUpload(t.Context(), &awss3.CompleteMultipartUploadInput{
+		Bucket: aws.String(bucket), Key: aws.String(key), UploadId: aws.String(id),
+		MultipartUpload: &types.CompletedMultipartUpload{Parts: parts},
+	})
+}
+
+func TestMultipartUploadCompletes(t *testing.T) {
+	s := s3.New()
+	c := newClient(t, s, "us-west-2")
+	create(t, c, "data")
+	if _, err := c.PutBucketVersioning(t.Context(), &awss3.PutBucketVersioningInput{
+		Bucket:                  aws.String("data"),
+		VersioningConfiguration: &types.VersioningConfiguration{Status: types.BucketVersioningStatusEnabled},
+	}); err != nil {
+		t.Fatal(err)
+	}
+	id := startUpload(t, c, "data", "big")
+	first, last := bytes.Repeat([]byte("a"), 5<<20), []byte("tail")
+	p1 := uploadPart(t, c, "data", "big", id, 1, first)
+	uploadPart(t, c, "data", "big", id, 2, []byte("replaced"))
+	p2 := uploadPart(t, c, "data", "big", id, 2, last)
+
+	ups := s.MultipartUploads("data")
+	if len(ups) != 1 || len(ups[0].Parts) != 2 || ups[0].Parts[0].Size != 5<<20 || ups[0].Parts[1].Size != 4 {
+		t.Fatalf("MultipartUploads() = %+v", ups)
+	}
+	parts, err := c.ListParts(t.Context(), &awss3.ListPartsInput{Bucket: aws.String("data"), Key: aws.String("big"), UploadId: aws.String(id), MaxParts: aws.Int32(1)})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(parts.Parts) != 1 || aws.ToInt32(parts.Parts[0].PartNumber) != 1 || !aws.ToBool(parts.IsTruncated) || aws.ToString(parts.NextPartNumberMarker) != "1" {
+		t.Fatalf("ListParts = %+v", parts)
+	}
+
+	out, err := complete(t, c, "data", "big", id, p1, p2)
+	if err != nil {
+		t.Fatal(err)
+	}
+	sum1, sum2 := md5.Sum(first), md5.Sum(last)
+	whole := md5.Sum(append(sum1[:], sum2[:]...))
+	if want := fmt.Sprintf(`"%s-2"`, hex.EncodeToString(whole[:])); aws.ToString(out.ETag) != want || aws.ToString(out.VersionId) == "" {
+		t.Fatalf("complete = %s %s, want ETag %s and a version", aws.ToString(out.ETag), aws.ToString(out.VersionId), want)
+	}
+	versions := s.ObjectVersions("data")
+	if len(versions) != 1 || versions[0].Size != 5<<20+4 || versions[0].VersionID != aws.ToString(out.VersionId) {
+		t.Fatalf("versions = %+v", versions)
+	}
+	if ups := s.MultipartUploads("data"); len(ups) != 0 {
+		t.Fatalf("uploads after completing = %+v", ups)
+	}
+	if _, err := complete(t, c, "data", "big", id, p1, p2); sdkerr.AWSCode(err) != "NoSuchUpload" || sdkerr.AWSStatus(err) != http.StatusNotFound {
+		t.Fatalf("completing again = %v, want 404 NoSuchUpload", err)
+	}
+}
+
+func TestMultipartUploadErrors(t *testing.T) {
+	s := s3.New()
+	c := newClient(t, s, "us-east-1")
+	create(t, c, "data")
+	id := startUpload(t, c, "data", "k")
+	small := uploadPart(t, c, "data", "k", id, 1, []byte("small"))
+	other := uploadPart(t, c, "data", "k", id, 2, []byte("other"))
+
+	for _, tc := range []struct {
+		name  string
+		parts []types.CompletedPart
+		code  string
+	}{
+		{"parts out of order", []types.CompletedPart{other, small}, "InvalidPartOrder"},
+		{"a part repeated", []types.CompletedPart{small, small}, "InvalidPartOrder"},
+		{"a part never uploaded", []types.CompletedPart{small, {PartNumber: aws.Int32(3), ETag: other.ETag}}, "InvalidPart"},
+		{"a part's ETag wrong", []types.CompletedPart{{PartNumber: aws.Int32(1), ETag: other.ETag}}, "InvalidPart"},
+		{"a part under 5 MiB before the last", []types.CompletedPart{small, other}, "EntityTooSmall"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			if _, err := complete(t, c, "data", "k", id, tc.parts...); sdkerr.AWSCode(err) != tc.code || sdkerr.AWSStatus(err) != http.StatusBadRequest {
+				t.Fatalf("complete = %v, want 400 %s", err, tc.code)
+			}
+		})
+	}
+	if _, err := c.UploadPart(t.Context(), &awss3.UploadPartInput{
+		Bucket: aws.String("data"), Key: aws.String("k"), UploadId: aws.String(id), PartNumber: aws.Int32(10_001), Body: strings.NewReader("x"),
+	}); sdkerr.AWSCode(err) != "InvalidArgument" {
+		t.Fatalf("part 10001 = %v, want InvalidArgument", err)
+	}
+	if _, err := c.UploadPart(t.Context(), &awss3.UploadPartInput{
+		Bucket: aws.String("data"), Key: aws.String("other-key"), UploadId: aws.String(id), PartNumber: aws.Int32(1), Body: strings.NewReader("x"),
+	}); sdkerr.AWSCode(err) != "NoSuchUpload" {
+		t.Fatalf("a part for another key = %v, want NoSuchUpload", err)
+	}
+	if _, err := complete(t, c, "data", "k", id, small); err != nil {
+		t.Fatalf("one small part = %v", err)
+	}
+
+	id = startUpload(t, c, "data", "k")
+	if _, err := c.AbortMultipartUpload(t.Context(), &awss3.AbortMultipartUploadInput{Bucket: aws.String("data"), Key: aws.String("k"), UploadId: aws.String(id)}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := c.AbortMultipartUpload(t.Context(), &awss3.AbortMultipartUploadInput{Bucket: aws.String("data"), Key: aws.String("k"), UploadId: aws.String(id)}); sdkerr.AWSCode(err) != "NoSuchUpload" {
+		t.Fatalf("aborting again = %v, want NoSuchUpload", err)
+	}
+	if _, err := c.ListParts(t.Context(), &awss3.ListPartsInput{Bucket: aws.String("data"), Key: aws.String("k"), UploadId: aws.String(id)}); sdkerr.AWSCode(err) != "NoSuchUpload" {
+		t.Fatalf("parts of an aborted upload = %v, want NoSuchUpload", err)
+	}
+}
+
+func TestListMultipartUploadsPagesByKeyAndInitiation(t *testing.T) {
+	s := s3.New()
+	c := newClient(t, s, "us-east-1")
+	create(t, c, "data")
+	var want []string
+	for _, key := range []string{"b", "a", "b", "photos/1", "photos/2", "videos/1", "c"} {
+		up, err := s.CreateMultipartUpload("data", key)
+		if err != nil {
+			t.Fatal(err)
+		}
+		want = append(want, key+" "+up.UploadID)
+	}
+	slices.SortStableFunc(want, func(x, y string) int { return strings.Compare(strings.Fields(x)[0], strings.Fields(y)[0]) })
+
+	var got []string
+	pages := 0
+	p := awss3.NewListMultipartUploadsPaginator(c, &awss3.ListMultipartUploadsInput{Bucket: aws.String("data")}, func(o *awss3.ListMultipartUploadsPaginatorOptions) { o.Limit = 2 })
+	for p.HasMorePages() {
+		page, err := p.NextPage(t.Context())
+		if err != nil {
+			t.Fatal(err)
+		}
+		pages++
+		for _, u := range page.Uploads {
+			got = append(got, aws.ToString(u.Key)+" "+aws.ToString(u.UploadId))
+			if u.Initiated == nil || aws.ToString(u.Owner.ID) == "" {
+				t.Errorf("upload = %+v", u)
+			}
+		}
+	}
+	if !slices.Equal(got, want) || pages != 4 {
+		t.Fatalf("uploads in %d pages = %v, want %v", pages, got, want)
+	}
+
+	var keys, prefixes []string
+	p = awss3.NewListMultipartUploadsPaginator(c, &awss3.ListMultipartUploadsInput{Bucket: aws.String("data"), Delimiter: aws.String("/")}, func(o *awss3.ListMultipartUploadsPaginatorOptions) { o.Limit = 1 })
+	for p.HasMorePages() {
+		page, err := p.NextPage(t.Context())
+		if err != nil {
+			t.Fatal(err)
+		}
+		for _, u := range page.Uploads {
+			keys = append(keys, aws.ToString(u.Key))
+		}
+		for _, cp := range page.CommonPrefixes {
+			prefixes = append(prefixes, aws.ToString(cp.Prefix))
+		}
+	}
+	if !slices.Equal(keys, []string{"a", "b", "b", "c"}) || !slices.Equal(prefixes, []string{"photos/", "videos/"}) {
+		t.Fatalf("by delimiter = %v and %v", keys, prefixes)
+	}
+
+	inside, err := c.ListMultipartUploads(t.Context(), &awss3.ListMultipartUploadsInput{
+		Bucket: aws.String("data"), Delimiter: aws.String("/"), KeyMarker: aws.String("photos/1"),
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(inside.CommonPrefixes) != 1 || aws.ToString(inside.CommonPrefixes[0].Prefix) != "videos/" || len(inside.Uploads) != 0 {
+		t.Fatalf("after a key inside photos/ = uploads %+v, prefixes %+v, want only videos/", inside.Uploads, inside.CommonPrefixes)
+	}
+
+	out, err := c.ListMultipartUploads(t.Context(), &awss3.ListMultipartUploadsInput{Bucket: aws.String("data"), Prefix: aws.String("photos/")})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(out.Uploads) != 2 || aws.ToString(out.Prefix) != "photos/" || aws.ToBool(out.IsTruncated) {
+		t.Fatalf("by prefix = %+v", out)
+	}
+}
+
+func TestBucketWithOnlyUploadsDeletes(t *testing.T) {
+	s := s3.New()
+	c := newClient(t, s, "us-east-1")
+	create(t, c, "data")
+	startUpload(t, c, "data", "k")
+	if _, err := c.DeleteBucket(t.Context(), &awss3.DeleteBucketInput{Bucket: aws.String("data")}); err != nil {
+		t.Fatalf("delete a bucket with an upload = %v", err)
+	}
+	if ups := s.MultipartUploads("data"); ups != nil {
+		t.Fatalf("uploads = %+v", ups)
+	}
+}
+
+func TestLostCompletionLeavesTheObject(t *testing.T) {
+	faults := fault.New().Add(fault.Rule{Method: "s3.CompleteMultipartUpload", Nth: 1, Fault: fault.LoseResponse()})
+	s := s3.New(s3.WithFaults(faults))
+	c := newClient(t, s, "us-east-1")
+	create(t, c, "data")
+	id := startUpload(t, c, "data", "k")
+	part := uploadPart(t, c, "data", "k", id, 1, []byte("x"))
+	if _, err := complete(t, c, "data", "k", id, part); err == nil {
+		t.Fatal("complete with a lost response succeeded")
+	}
+	if _, err := complete(t, c, "data", "k", id, part); sdkerr.AWSCode(err) != "NoSuchUpload" {
+		t.Fatalf("retry = %v, want NoSuchUpload", err)
+	}
+	if v := s.ObjectVersions("data"); len(v) != 1 || v[0].Key != "k" {
+		t.Fatalf("versions = %+v", v)
+	}
+}
+
+func TestMultipartPermissions(t *testing.T) {
+	policy := access.Allow("s3:CreateBucket", "s3:PutObject", "s3:ListBucketMultipartUploads")
+	s := s3.New(s3.WithPolicy(policy))
+	c := newClient(t, s, "us-east-1")
+	create(t, c, "data")
+	id := startUpload(t, c, "data", "k")
+	if _, err := c.ListMultipartUploads(t.Context(), &awss3.ListMultipartUploadsInput{Bucket: aws.String("data")}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := c.AbortMultipartUpload(t.Context(), &awss3.AbortMultipartUploadInput{Bucket: aws.String("data"), Key: aws.String("k"), UploadId: aws.String(id)}); sdkerr.AWSCode(err) != "AccessDenied" {
+		t.Fatalf("abort = %v, want AccessDenied", err)
+	}
+	if _, err := c.ListParts(t.Context(), &awss3.ListPartsInput{Bucket: aws.String("data"), Key: aws.String("k"), UploadId: aws.String(id)}); !strings.Contains(fmt.Sprint(err), "s3:ListMultipartUploadParts") {
+		t.Fatalf("list parts = %v, want it to name s3:ListMultipartUploadParts", err)
+	}
+}
diff --git a/sdktest/aws/s3/objectlock_test.go b/sdktest/aws/s3/objectlock_test.go
new file mode 100644
index 0000000..312b948
--- /dev/null
+++ b/sdktest/aws/s3/objectlock_test.go
@@ -0,0 +1,403 @@
+package s3_test
+
+import (
+	"net/http"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/aws/aws-sdk-go-v2/service/s3/types"
+
+	"github.com/parallelworks/squall/access"
+	"github.com/parallelworks/squall/aws/s3"
+	"github.com/parallelworks/squall/sdktest/internal/sdkerr"
+)
+
+// clock is a settable clock for a server, safe to move while it serves.
+type clock struct{ nanos atomic.Int64 }
+
+func newClock() *clock {
+	c := &clock{}
+	c.nanos.Store(time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC).UnixNano())
+	return c
+}
+
+func (c *clock) now() time.Time                  { return time.Unix(0, c.nanos.Load()).UTC() }
+func (c *clock) advance(d time.Duration)         { c.nanos.Add(int64(d)) }
+func (c *clock) after(d time.Duration) time.Time { return c.now().Add(d) }
+
+func createLocked(t *testing.T, c *awss3.Client, name string) {
+	t.Helper()
+	if _, err := c.CreateBucket(t.Context(), &awss3.CreateBucketInput{Bucket: aws.String(name), ObjectLockEnabledForBucket: aws.Bool(true)}); err != nil {
+		t.Fatalf("create %s: %v", name, err)
+	}
+}
+
+func deleteVersion(t *testing.T, c *awss3.Client, bucket, key, id string, bypass bool) error {
+	t.Helper()
+	in := &awss3.DeleteObjectInput{Bucket: aws.String(bucket), Key: aws.String(key), VersionId: aws.String(id)}
+	if bypass {
+		in.BypassGovernanceRetention = aws.Bool(true)
+	}
+	_, err := c.DeleteObject(t.Context(), in)
+	return err
+}
+
+func isLocked(err error) bool {
+	return sdkerr.AWSCode(err) == "AccessDenied" && sdkerr.AWSStatus(err) == http.StatusForbidden && strings.Contains(err.Error(), "object lock")
+}
+
+func TestObjectLockBucket(t *testing.T) {
+	s := s3.New()
+	c := newClient(t, s, "us-east-1")
+	createLocked(t, c, "vault")
+	v, err := c.GetBucketVersioning(t.Context(), &awss3.GetBucketVersioningInput{Bucket: aws.String("vault")})
+	if err != nil || v.Status != types.BucketVersioningStatusEnabled {
+		t.Fatalf("versioning = %+v, %v", v, err)
+	}
+	if _, err := c.PutBucketVersioning(t.Context(), &awss3.PutBucketVersioningInput{
+		Bucket:                  aws.String("vault"),
+		VersioningConfiguration: &types.VersioningConfiguration{Status: types.BucketVersioningStatusSuspended},
+	}); sdkerr.AWSCode(err) != "InvalidBucketState" || sdkerr.AWSStatus(err) != http.StatusConflict {
+		t.Fatalf("suspend = %v, want 409 InvalidBucketState", err)
+	}
+	got, err := c.GetObjectLockConfiguration(t.Context(), &awss3.GetObjectLockConfigurationInput{Bucket: aws.String("vault")})
+	if err != nil || got.ObjectLockConfiguration.ObjectLockEnabled != types.ObjectLockEnabledEnabled || got.ObjectLockConfiguration.Rule != nil {
+		t.Fatalf("lock configuration = %+v, %v", got, err)
+	}
+	if _, err := c.PutObjectLockConfiguration(t.Context(), &awss3.PutObjectLockConfigurationInput{
+		Bucket: aws.String("vault"),
+		ObjectLockConfiguration: &types.ObjectLockConfiguration{
+			ObjectLockEnabled: types.ObjectLockEnabledEnabled,
+			Rule:              &types.ObjectLockRule{DefaultRetention: &types.DefaultRetention{Mode: types.ObjectLockRetentionModeGovernance, Days: aws.Int32(3)}},
+		},
+	}); err != nil {
+		t.Fatal(err)
+	}
+	got, err = c.GetObjectLockConfiguration(t.Context(), &awss3.GetObjectLockConfigurationInput{Bucket: aws.String("vault")})
+	if err != nil || got.ObjectLockConfiguration.Rule.DefaultRetention.Mode != types.ObjectLockRetentionModeGovernance ||
+		aws.ToInt32(got.ObjectLockConfiguration.Rule.DefaultRetention.Days) != 3 {
+		t.Fatalf("lock configuration = %+v, %v", got, err)
+	}
+	for _, d := range []*types.DefaultRetention{
+		{Mode: types.ObjectLockRetentionModeGovernance, Days: aws.Int32(1), Years: aws.Int32(1)},
+		{Mode: types.ObjectLockRetentionModeGovernance},
+		{Days: aws.Int32(1)},
+	} {
+		if _, err := c.PutObjectLockConfiguration(t.Context(), &awss3.PutObjectLockConfigurationInput{
+			Bucket:                  aws.String("vault"),
+			ObjectLockConfiguration: &types.ObjectLockConfiguration{ObjectLockEnabled: types.ObjectLockEnabledEnabled, Rule: &types.ObjectLockRule{DefaultRetention: d}},
+		}); sdkerr.AWSCode(err) != "MalformedXML" {
+			t.Errorf("default retention %+v = %v, want MalformedXML", d, err)
+		}
+	}
+	if _, err := c.PutObjectLockConfiguration(t.Context(), &awss3.PutObjectLockConfigurationInput{
+		Bucket: aws.String("vault"),
+		ObjectLockConfiguration: &types.ObjectLockConfiguration{
+			ObjectLockEnabled: types.ObjectLockEnabledEnabled,
+			Rule:              &types.ObjectLockRule{DefaultRetention: &types.DefaultRetention{Mode: types.ObjectLockRetentionModeGovernance, Days: aws.Int32(0)}},
+		},
+	}); sdkerr.AWSCode(err) != "InvalidArgument" {
+		t.Errorf("zero days = %v, want InvalidArgument", err)
+	}
+
+	create(t, c, "plain")
+	if _, err := c.GetObjectLockConfiguration(t.Context(), &awss3.GetObjectLockConfigurationInput{Bucket: aws.String("plain")}); sdkerr.AWSCode(err) != "ObjectLockConfigurationNotFoundError" || sdkerr.AWSStatus(err) != http.StatusNotFound {
+		t.Fatalf("plain lock configuration = %v, want 404 ObjectLockConfigurationNotFoundError", err)
+	}
+	enable := &awss3.PutObjectLockConfigurationInput{
+		Bucket:                  aws.String("plain"),
+		ObjectLockConfiguration: &types.ObjectLockConfiguration{ObjectLockEnabled: types.ObjectLockEnabledEnabled},
+	}
+	if _, err := c.PutObjectLockConfiguration(t.Context(), enable); sdkerr.AWSCode(err) != "InvalidBucketState" {
+		t.Fatalf("enabling without versioning = %v, want InvalidBucketState", err)
+	}
+	if _, err := c.PutObject(t.Context(), &awss3.PutObjectInput{
+		Bucket: aws.String("plain"), Key: aws.String("k"), Body: strings.NewReader("x"), ObjectLockLegalHoldStatus: types.ObjectLockLegalHoldStatusOn,
+	}); sdkerr.AWSCode(err) != "InvalidRequest" {
+		t.Fatalf("a legal hold outside Object Lock = %v, want InvalidRequest", err)
+	}
+	if _, err := c.PutBucketVersioning(t.Context(), &awss3.PutBucketVersioningInput{
+		Bucket:                  aws.String("plain"),
+		VersioningConfiguration: &types.VersioningConfiguration{Status: types.BucketVersioningStatusEnabled},
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := c.PutObjectLockConfiguration(t.Context(), enable); err != nil {
+		t.Fatalf("enabling on a versioned bucket = %v", err)
+	}
+	if b, _ := s.LookupBucket("plain"); !b.ObjectLock {
+		t.Fatalf("bucket = %+v", b)
+	}
+}
+
+func TestLockedVersionsRefuseDeletes(t *testing.T) {
+	clk := newClock()
+	s := s3.New(s3.WithClock(clk.now))
+	c := newClient(t, s, "us-east-1")
+	createLocked(t, c, "vault")
+
+	compliance, err := c.PutObject(t.Context(), &awss3.PutObjectInput{
+		Bucket: aws.String("vault"), Key: aws.String("c"), Body: strings.NewReader("c"),
+		ObjectLockMode: types.ObjectLockModeCompliance, ObjectLockRetainUntilDate: aws.Time(clk.after(48 * time.Hour)),
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	governance, err := c.PutObject(t.Context(), &awss3.PutObjectInput{
+		Bucket: aws.String("vault"), Key: aws.String("g"), Body: strings.NewReader("g"),
+		ObjectLockMode: types.ObjectLockModeGovernance, ObjectLockRetainUntilDate: aws.Time(clk.after(48 * time.Hour)),
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	held, err := c.PutObject(t.Context(), &awss3.PutObjectInput{
+		Bucket: aws.String("vault"), Key: aws.String("h"), Body: strings.NewReader("h"), ObjectLockLegalHoldStatus: types.ObjectLockLegalHoldStatusOn,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	for _, key := range []string{"c", "g", "h"} {
+		out, err := c.DeleteObject(t.Context(), &awss3.DeleteObjectInput{Bucket: aws.String("vault"), Key: aws.String(key)})
+		if err != nil || !aws.ToBool(out.DeleteMarker) {
+			t.Fatalf("a simple delete of %s = %+v, %v, want a delete marker", key, out, err)
+		}
+	}
+	if err := deleteVersion(t, c, "vault", "c", aws.ToString(compliance.VersionId), true); !isLocked(err) {
+		t.Fatalf("delete a compliance version with bypass = %v, want AccessDenied for the lock", err)
+	}
+	if err := deleteVersion(t, c, "vault", "g", aws.ToString(governance.VersionId), false); !isLocked(err) {
+		t.Fatalf("delete a governance version = %v, want AccessDenied for the lock", err)
+	}
+	if err := deleteVersion(t, c, "vault", "h", aws.ToString(held.VersionId), true); !isLocked(err) {
+		t.Fatalf("delete a held version with bypass = %v, want AccessDenied for the lock", err)
+	}
+	if err := deleteVersion(t, c, "vault", "g", aws.ToString(governance.VersionId), true); err != nil {
+		t.Fatalf("delete a governance version with bypass = %v", err)
+	}
+
+	out, err := c.DeleteObjects(t.Context(), &awss3.DeleteObjectsInput{
+		Bucket: aws.String("vault"),
+		Delete: &types.Delete{Objects: []types.ObjectIdentifier{
+			{Key: aws.String("c"), VersionId: compliance.VersionId},
+			{Key: aws.String("h"), VersionId: held.VersionId},
+			{Key: aws.String("g")},
+		}},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(out.Errors) != 2 || aws.ToString(out.Errors[0].Code) != "AccessDenied" || aws.ToString(out.Errors[1].Key) != "h" || len(out.Deleted) != 1 {
+		t.Fatalf("DeleteObjects = errors %+v, deleted %+v", out.Errors, out.Deleted)
+	}
+	if _, err := c.DeleteBucket(t.Context(), &awss3.DeleteBucketInput{Bucket: aws.String("vault")}); sdkerr.AWSCode(err) != "BucketNotEmpty" {
+		t.Fatalf("delete the bucket = %v, want BucketNotEmpty", err)
+	}
+
+	clk.advance(49 * time.Hour)
+	if err := deleteVersion(t, c, "vault", "c", aws.ToString(compliance.VersionId), false); err != nil {
+		t.Fatalf("delete an expired compliance version = %v", err)
+	}
+	if err := deleteVersion(t, c, "vault", "h", aws.ToString(held.VersionId), false); !isLocked(err) {
+		t.Fatalf("delete a held version = %v, want AccessDenied for the lock", err)
+	}
+	if _, err := c.PutObjectLegalHold(t.Context(), &awss3.PutObjectLegalHoldInput{
+		Bucket: aws.String("vault"), Key: aws.String("h"), VersionId: held.VersionId, LegalHold: &types.ObjectLockLegalHold{Status: types.ObjectLockLegalHoldStatusOff},
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if err := deleteVersion(t, c, "vault", "h", aws.ToString(held.VersionId), false); err != nil {
+		t.Fatalf("delete a released version = %v", err)
+	}
+}
+
+func TestRetentionAndLegalHold(t *testing.T) {
+	clk := newClock()
+	s := s3.New(s3.WithClock(clk.now))
+	c := newClient(t, s, "us-east-1")
+	createLocked(t, c, "vault")
+	put(t, c, "vault", "k", strings.NewReader("k"))
+
+	if _, err := c.GetObjectRetention(t.Context(), &awss3.GetObjectRetentionInput{Bucket: aws.String("vault"), Key: aws.String("k")}); sdkerr.AWSCode(err) != "NoSuchObjectLockConfiguration" {
+		t.Fatalf("retention of an unlocked version = %v, want NoSuchObjectLockConfiguration", err)
+	}
+	retain := func(mode types.ObjectLockRetentionMode, until time.Time, bypass bool) error {
+		in := &awss3.PutObjectRetentionInput{Bucket: aws.String("vault"), Key: aws.String("k"), Retention: &types.ObjectLockRetention{Mode: mode, RetainUntilDate: aws.Time(until)}}
+		if bypass {
+			in.BypassGovernanceRetention = aws.Bool(true)
+		}
+		_, err := c.PutObjectRetention(t.Context(), in)
+		return err
+	}
+	if err := retain(types.ObjectLockRetentionModeGovernance, clk.after(-time.Hour), false); sdkerr.AWSCode(err) != "InvalidArgument" {
+		t.Fatalf("retention in the past = %v, want InvalidArgument", err)
+	}
+	if err := retain(types.ObjectLockRetentionModeGovernance, clk.after(24*time.Hour), false); err != nil {
+		t.Fatal(err)
+	}
+	if err := retain(types.ObjectLockRetentionModeGovernance, clk.after(48*time.Hour), false); err != nil {
+		t.Fatalf("extend = %v", err)
+	}
+	if err := retain(types.ObjectLockRetentionModeGovernance, clk.after(time.Hour), false); !isLocked(err) {
+		t.Fatalf("shorten = %v, want AccessDenied for the lock", err)
+	}
+	if err := retain(types.ObjectLockRetentionModeCompliance, clk.after(time.Hour), true); err != nil {
+		t.Fatalf("shorten with bypass = %v", err)
+	}
+	got, err := c.GetObjectRetention(t.Context(), &awss3.GetObjectRetentionInput{Bucket: aws.String("vault"), Key: aws.String("k")})
+	if err != nil || got.Retention.Mode != types.ObjectLockRetentionModeCompliance || !got.Retention.RetainUntilDate.Equal(clk.after(time.Hour)) {
+		t.Fatalf("retention = %+v, %v", got, err)
+	}
+	if err := retain(types.ObjectLockRetentionModeGovernance, clk.after(2*time.Hour), true); !isLocked(err) {
+		t.Fatalf("weaken compliance with bypass = %v, want AccessDenied for the lock", err)
+	}
+
+	if _, err := c.GetObjectLegalHold(t.Context(), &awss3.GetObjectLegalHoldInput{Bucket: aws.String("vault"), Key: aws.String("k")}); sdkerr.AWSCode(err) != "NoSuchObjectLockConfiguration" {
+		t.Fatalf("legal hold never set = %v, want NoSuchObjectLockConfiguration", err)
+	}
+	if err := s.SetLegalHold("vault", "k", "", true); err != nil {
+		t.Fatal(err)
+	}
+	hold, err := c.GetObjectLegalHold(t.Context(), &awss3.GetObjectLegalHoldInput{Bucket: aws.String("vault"), Key: aws.String("k")})
+	if err != nil || hold.LegalHold.Status != types.ObjectLockLegalHoldStatusOn {
+		t.Fatalf("legal hold = %+v, %v", hold, err)
+	}
+	versions := s.ObjectVersions("vault")
+	if len(versions) != 1 || versions[0].RetentionMode != "COMPLIANCE" || !versions[0].LegalHold {
+		t.Fatalf("versions = %+v", versions)
+	}
+	if _, err := c.GetObjectRetention(t.Context(), &awss3.GetObjectRetentionInput{Bucket: aws.String("vault"), Key: aws.String("k"), VersionId: aws.String(strings.Repeat("0", 30) + "ff")}); sdkerr.AWSCode(err) != "NoSuchVersion" {
+		t.Fatalf("retention of a missing version = %v, want NoSuchVersion", err)
+	}
+}
+
+func TestDefaultRetentionLocksNewVersions(t *testing.T) {
+	clk := newClock()
+	s := s3.New(s3.WithClock(clk.now))
+	if _, err := s.PutBucket(s3.Bucket{
+		Name: "vault", Region: "us-east-1", Versioning: "Enabled", ObjectLock: true,
+		DefaultRetention: &s3.DefaultRetention{Mode: "COMPLIANCE", Days: 1},
+	}); err != nil {
+		t.Fatal(err)
+	}
+	c := newClient(t, s, "us-east-1")
+	put(t, c, "vault", "k", strings.NewReader("k"))
+	id := startUpload(t, c, "vault", "big")
+	part := uploadPart(t, c, "vault", "big", id, 1, []byte("x"))
+	if _, err := complete(t, c, "vault", "big", id, part); err != nil {
+		t.Fatal(err)
+	}
+	for _, v := range s.ObjectVersions("vault") {
+		if v.RetentionMode != "COMPLIANCE" || !v.RetainUntil.Equal(clk.after(24*time.Hour)) {
+			t.Errorf("version = %+v, want a day's compliance retention", v)
+		}
+		if err := deleteVersion(t, c, "vault", v.Key, v.VersionID, true); !isLocked(err) {
+			t.Errorf("delete %s = %v, want AccessDenied for the lock", v.Key, err)
+		}
+	}
+	clk.advance(24 * time.Hour)
+	for _, v := range s.ObjectVersions("vault") {
+		if err := deleteVersion(t, c, "vault", v.Key, v.VersionID, false); err != nil {
+			t.Errorf("delete %s after a day = %v", v.Key, err)
+		}
+	}
+	if _, err := c.DeleteBucket(t.Context(), &awss3.DeleteBucketInput{Bucket: aws.String("vault")}); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestBypassNeedsItsPermission(t *testing.T) {
+	clk := newClock()
+	policy := access.Allow("s3:CreateBucket", "s3:PutBucketObjectLockConfiguration", "s3:PutBucketVersioning", "s3:PutObject", "s3:DeleteObjectVersion")
+	s := s3.New(s3.WithClock(clk.now), s3.WithPolicy(policy))
+	c := newClient(t, s, "us-east-1")
+	createLocked(t, c, "vault")
+	out := put(t, c, "vault", "k", strings.NewReader("k"))
+	if err := s.SetRetention("vault", "k", aws.ToString(out.VersionId), "GOVERNANCE", clk.after(time.Hour)); err != nil {
+		t.Fatal(err)
+	}
+	err := deleteVersion(t, c, "vault", "k", aws.ToString(out.VersionId), true)
+	if sdkerr.AWSCode(err) != "AccessDenied" || !strings.Contains(err.Error(), "s3:BypassGovernanceRetention") {
+		t.Fatalf("bypass without the permission = %v, want AccessDenied naming s3:BypassGovernanceRetention", err)
+	}
+	if _, err := c.PutObject(t.Context(), &awss3.PutObjectInput{
+		Bucket: aws.String("vault"), Key: aws.String("k"), Body: strings.NewReader("k"), ObjectLockLegalHoldStatus: types.ObjectLockLegalHoldStatusOn,
+	}); sdkerr.AWSCode(err) != "AccessDenied" || !strings.Contains(err.Error(), "s3:PutObjectLegalHold") {
+		t.Fatalf("a legal hold without its permission = %v", err)
+	}
+}
+
+func TestRequestPayment(t *testing.T) {
+	c := newClient(t, s3.New(), "us-east-1")
+	create(t, c, "data")
+	got, err := c.GetBucketRequestPayment(t.Context(), &awss3.GetBucketRequestPaymentInput{Bucket: aws.String("data")})
+	if err != nil || got.Payer != types.PayerBucketOwner {
+		t.Fatalf("payer = %+v, %v", got, err)
+	}
+	if _, err := c.PutBucketRequestPayment(t.Context(), &awss3.PutBucketRequestPaymentInput{
+		Bucket: aws.String("data"), RequestPaymentConfiguration: &types.RequestPaymentConfiguration{Payer: types.PayerRequester},
+	}); err != nil {
+		t.Fatal(err)
+	}
+	got, err = c.GetBucketRequestPayment(t.Context(), &awss3.GetBucketRequestPaymentInput{Bucket: aws.String("data")})
+	if err != nil || got.Payer != types.PayerRequester {
+		t.Fatalf("payer = %+v, %v", got, err)
+	}
+	if _, err := c.PutBucketRequestPayment(t.Context(), &awss3.PutBucketRequestPaymentInput{
+		Bucket: aws.String("data"), RequestPaymentConfiguration: &types.RequestPaymentConfiguration{Payer: "Nobody"},
+	}); sdkerr.AWSCode(err) != "MalformedXML" {
+		t.Fatalf("payer Nobody = %v, want MalformedXML", err)
+	}
+	if _, err := c.PutObject(t.Context(), &awss3.PutObjectInput{Bucket: aws.String("data"), Key: aws.String("k"), Body: strings.NewReader("k")}); err != nil {
+		t.Fatalf("the owner writing to a Requester Pays bucket = %v", err)
+	}
+}
+
+func TestRetentionReadBackIsNotAShortening(t *testing.T) {
+	s := s3.New(s3.WithClock(func() time.Time { return time.Date(2026, 1, 1, 0, 0, 0, 123456789, time.UTC) }))
+	if _, err := s.PutBucket(s3.Bucket{
+		Name: "vault", Region: "us-east-1", Versioning: "Enabled", ObjectLock: true,
+		DefaultRetention: &s3.DefaultRetention{Mode: "COMPLIANCE", Days: 1},
+	}); err != nil {
+		t.Fatal(err)
+	}
+	c := newClient(t, s, "us-east-1")
+	put(t, c, "vault", "k", strings.NewReader("k"))
+	got, err := c.GetObjectRetention(t.Context(), &awss3.GetObjectRetentionInput{Bucket: aws.String("vault"), Key: aws.String("k")})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := c.PutObjectRetention(t.Context(), &awss3.PutObjectRetentionInput{
+		Bucket: aws.String("vault"), Key: aws.String("k"), Retention: got.Retention,
+	}); err != nil {
+		t.Fatalf("putting back the retention read = %v", err)
+	}
+	if v := s.ObjectVersions("vault"); !v[0].RetainUntil.Equal(time.Date(2026, 1, 2, 0, 0, 0, 123000000, time.UTC)) {
+		t.Fatalf("RetainUntil = %v, want it to the millisecond", v[0].RetainUntil)
+	}
+}
+
+func TestLockedUploadPartsNeedAChecksum(t *testing.T) {
+	s := s3.New()
+	url := serve(t, s)
+	c := clientAt(url, "us-east-1")
+	createLocked(t, c, "vault")
+	out, err := c.CreateMultipartUpload(t.Context(), &awss3.CreateMultipartUploadInput{
+		Bucket: aws.String("vault"), Key: aws.String("k"), ObjectLockLegalHoldStatus: types.ObjectLockLegalHoldStatusOn,
+		ObjectLockMode: types.ObjectLockModeGovernance, ObjectLockRetainUntilDate: aws.Time(time.Now().Add(time.Hour)),
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	bare := clientAt(url, "us-east-1", func(o *awss3.Options) { o.RequestChecksumCalculation = aws.RequestChecksumCalculationWhenRequired })
+	if _, err := bare.UploadPart(t.Context(), &awss3.UploadPartInput{
+		Bucket: aws.String("vault"), Key: aws.String("k"), UploadId: out.UploadId, PartNumber: aws.Int32(1), Body: strings.NewReader("x"),
+	}); sdkerr.AWSCode(err) != "InvalidRequest" || !strings.Contains(err.Error(), "Put Part requests") {
+		t.Fatalf("a part without a checksum = %v, want InvalidRequest for Put Part requests", err)
+	}
+	uploadPart(t, c, "vault", "k", aws.ToString(out.UploadId), 1, []byte("x"))
+}