feat(s3): multipart uploads, Object Lock retention and legal holds, and request payment

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.

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