package objstore

import (
	"bytes"
	"context"
	"crypto/hmac"
	"crypto/md5"
	"crypto/sha256"
	"encoding/base64"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"mime/multipart"
	"net/http"
	"os"
	"strings"
	"testing"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/credentials"
	awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
	s3types "github.com/aws/aws-sdk-go-v2/service/s3/types"
	"github.com/aws/smithy-go"
)

// TestDeliveredContractMinio pins spec §1's delivered contract from outside
// the package: given exactly the four values a consuming app receives after
// grant (endpoint, key, secret, bucket) plus a second bucket the scoped user
// was never granted, build an S3 client the way that consuming app would —
// not via NewAWSProvisioner or any objstore helper — and exercise Put/Get/
// List/Delete plus cross-bucket denial against a real (Minio) S3-compatible
// server.
//
// This is deliberately not a unit test against a stub: internal/objstore's
// own tests already cover the provisioner logic with fakes. What has never
// been exercised anywhere in this tree is the actual wire contract a member
// app's client.Options() produces from delivered env, against a real
// S3-compatible implementation. Skipped by default because it needs a
// running Minio; CI's 30-minio-contract step (.amadan/ci.d/30-minio-contract)
// sets CARLOS_MINIO_TEST=1 and stands one up, as a plain process on
// loopback rather than a container. That step is now the only place this
// test runs live anywhere, so treat a change to it as a change to what
// this test is worth.
func TestDeliveredContractMinio(t *testing.T) {
	if os.Getenv("CARLOS_MINIO_TEST") != "1" {
		t.Skip("set CARLOS_MINIO_TEST=1 (with a running Minio and the MINIO_* env vars) to run the delivered-contract test; see .amadan/ci.d/30-minio-contract, which stands one up")
	}

	endpoint := requireEnv(t, "MINIO_ENDPOINT")
	key := requireEnv(t, "MINIO_SCOPED_KEY")
	secret := requireEnv(t, "MINIO_SCOPED_SECRET")
	bucket := requireEnv(t, "MINIO_BUCKET")
	deniedBucket := requireEnv(t, "MINIO_DENIED_BUCKET")

	ctx := context.Background()

	// Built exactly the way a consuming app builds it from delivered env:
	// static credentials, region us-east-1 (spec §1's delivered region),
	// BaseEndpoint + UsePathStyle for addressing, and the checksum
	// WhenRequired pair — the same four option lines aws.go's
	// NewAWSProvisioner applies for a non-empty endpoint, mirrored here
	// independently rather than by calling into objstore's own client
	// construction, since the whole point is to pin what a member app gets,
	// not to test this package's internals against themselves.
	client := awss3.New(awss3.Options{
		Region:                     "us-east-1",
		Credentials:                credentials.NewStaticCredentialsProvider(key, secret, ""),
		BaseEndpoint:               aws.String(endpoint),
		UsePathStyle:               true,
		RequestChecksumCalculation: aws.RequestChecksumCalculationWhenRequired,
		ResponseChecksumValidation: aws.ResponseChecksumValidationWhenRequired,
	})

	const objectKey = "contract-test/hello.txt"
	const body = "hello from the delivered contract test"

	t.Run("put", func(t *testing.T) {
		_, err := client.PutObject(ctx, &awss3.PutObjectInput{
			Bucket: aws.String(bucket),
			Key:    aws.String(objectKey),
			Body:   bytes.NewReader([]byte(body)),
		})
		if err != nil {
			t.Fatalf("PutObject: %v", err)
		}
	})

	t.Run("get round-trips the body", func(t *testing.T) {
		out, err := client.GetObject(ctx, &awss3.GetObjectInput{
			Bucket: aws.String(bucket),
			Key:    aws.String(objectKey),
		})
		if err != nil {
			t.Fatalf("GetObject: %v", err)
		}
		defer out.Body.Close()
		got, err := io.ReadAll(out.Body)
		if err != nil {
			t.Fatalf("read GetObject body: %v", err)
		}
		if string(got) != body {
			t.Fatalf("GetObject body = %q, want %q", got, body)
		}
	})

	t.Run("list shows the key", func(t *testing.T) {
		out, err := client.ListObjectsV2(ctx, &awss3.ListObjectsV2Input{
			Bucket: aws.String(bucket),
		})
		if err != nil {
			t.Fatalf("ListObjectsV2: %v", err)
		}
		if !containsKey(out.Contents, objectKey) {
			t.Fatalf("ListObjectsV2 contents = %v, want to contain %q", keyNames(out.Contents), objectKey)
		}
	})

	// THE VERSION A VERDICT BINDS TO HAS TO BE THE SAME STRING FROM BOTH
	// DOORS, and this is the only place that can be checked against a real
	// server rather than argued about.
	//
	// The sweep never calls HeadObject: it takes ETag, size and last-modified
	// off the listing page (sweep.go:31) and binds its tag to that
	// ScanObject.Version(). Status does the opposite — it heads the object to
	// find the current version for itself (clamav.go:199) — and then compares
	// the tag it read against ITS version, character for character
	// (`taggedVersion != obj.Version()`).
	//
	// So if a gateway reports last-modified differently on the two calls —
	// rounding one where it truncates the other — a tag the sweep just wrote
	// reads as "nothing is known" to every door, the drain rescans, re-tags
	// with the Head version, and the next sweep finds THAT stale and rescans
	// again. A scan ping-pong that costs money and never converges, from two
	// strings that differ in a second.
	//
	// RFC3339 truncates, so the two agree unless the server rounds. That is a
	// narrow window, which is exactly why it wants a test and not a memo.
	t.Run("head and list agree on the version", func(t *testing.T) {
		lst, err := client.ListObjectsV2(ctx, &awss3.ListObjectsV2Input{
			Bucket: aws.String(bucket),
			Prefix: aws.String(objectKey),
		})
		if err != nil {
			t.Fatalf("ListObjectsV2: %v", err)
		}
		var listed ScanObject
		for _, o := range lst.Contents {
			if aws.ToString(o.Key) != objectKey {
				continue
			}
			// Exactly what s3Lister.ListPage builds (sweep.go).
			listed = ScanObject{Key: objectKey, ETag: normalizeETag(aws.ToString(o.ETag))}
			if o.LastModified != nil {
				listed.Modified = *o.LastModified
			}
			if o.Size != nil {
				listed.Size = *o.Size
			}
		}
		if listed.ETag == "" {
			t.Fatalf("ListObjectsV2 did not return %q", objectKey)
		}

		head, err := client.HeadObject(ctx, &awss3.HeadObjectInput{
			Bucket: aws.String(bucket),
			Key:    aws.String(objectKey),
		})
		if err != nil {
			t.Fatalf("HeadObject: %v", err)
		}
		// Exactly what clamavScanner.Status builds (clamav.go:212-219).
		headed := ScanObject{Key: objectKey, ETag: normalizeETag(aws.ToString(head.ETag))}
		if head.LastModified != nil {
			headed.Modified = *head.LastModified
		}

		if listed.Version() != headed.Version() {
			t.Fatalf("the sweep and the status door would bind different versions to the same bytes:\n  list: %s\n  head: %s\nEvery verdict the sweep writes reads as unknown, forever.", listed.Version(), headed.Version())
		}
		// Size only comes from the listing in the sweep's path, but Status
		// reports it too and the drain uses it to decide the ceiling.
		if head.ContentLength != nil && listed.Size != *head.ContentLength {
			t.Fatalf("size disagrees: list %d, head %d", listed.Size, *head.ContentLength)
		}
	})

	// THE ANTI-FORGERY PROPERTY, against a real IAM rather than a policy
	// document. Everything about a clean verdict rests on one asymmetry: the
	// console can write the carlos-scan tag and the app cannot. The app's
	// Allow has no s3:PutObjectTagging and never will (aws.go's policy, and
	// the standing rule in AGENTS.md) — but "the permission is absent" is only
	// half a proof. The other half is that the server actually requires it,
	// by BOTH routes, and that half has never been exercised anywhere.
	//
	// The second route is the one that would be missed. x-amz-tagging on a
	// PutObject sets tags without any call named PutObjectTagging, and an
	// implementation that only gates the explicit call would hand an app the
	// ability to upload a payload pre-marked clean.
	t.Run("the app cannot tag its own object", func(t *testing.T) {
		t.Run("not by PutObjectTagging", func(t *testing.T) {
			_, err := client.PutObjectTagging(ctx, &awss3.PutObjectTaggingInput{
				Bucket: aws.String(bucket),
				Key:    aws.String(objectKey),
				Tagging: &s3types.Tagging{TagSet: []s3types.Tag{
					{Key: aws.String("carlos-scan"), Value: aws.String("clean")},
				}},
			})
			requireDenied(t, err, "PutObjectTagging")
		})

		t.Run("and not by x-amz-tagging on upload", func(t *testing.T) {
			_, err := client.PutObject(ctx, &awss3.PutObjectInput{
				Bucket:  aws.String(bucket),
				Key:     aws.String("contract-test/pre-tagged.txt"),
				Body:    bytes.NewReader([]byte("payload claiming to be clean")),
				Tagging: aws.String("carlos-scan=clean"),
			})
			// Clean up if it somehow succeeded, so the delete subtest below
			// is not reasoning about a bucket this one left dirty.
			if err == nil {
				_, _ = client.DeleteObject(ctx, &awss3.DeleteObjectInput{
					Bucket: aws.String(bucket),
					Key:    aws.String("contract-test/pre-tagged.txt"),
				})
			}
			requireDenied(t, err, "PutObject with x-amz-tagging")
		})
	})

	// THE OTHER ROUTES. "By either route" was two routes, and the property has
	// more than two.
	//
	// Each of these is authorised as plain `s3:PutObject` and carries tags
	// somewhere other than the `x-amz-tagging` header of a simple PUT, so the
	// Deny added for that header is not obviously in the path. Whether a gateway
	// populates `s3:RequestObjectTagKeys` in these handlers is a gateway-specific
	// fact of exactly the kind that already bit this design once — which is why
	// they are asserted against a real server rather than reasoned about.
	t.Run("the app cannot tag by any other route", func(t *testing.T) {
		// Multipart: any upload over the SDK's threshold takes this path, so
		// an app does not even have to choose it deliberately.
		t.Run("not by CreateMultipartUpload", func(t *testing.T) {
			const key = "contract-test/multipart-tagged.bin"
			out, err := client.CreateMultipartUpload(ctx, &awss3.CreateMultipartUploadInput{
				Bucket:  aws.String(bucket),
				Key:     aws.String(key),
				Tagging: aws.String("carlos-scan=clean"),
			})
			if err == nil {
				// Nothing was completed, but the upload id exists; clean it up so
				// the bucket is not left holding a live multipart session.
				_, _ = client.AbortMultipartUpload(ctx, &awss3.AbortMultipartUploadInput{
					Bucket: aws.String(bucket), Key: aws.String(key), UploadId: out.UploadId,
				})
			}
			requireDenied(t, err, "CreateMultipartUpload with x-amz-tagging")
		})

		// Copy with REPLACE: the app already has GetObject and PutObject, so it
		// can copy its own object over itself and name the tags on the way.
		t.Run("not by CopyObject with a REPLACE tagging directive", func(t *testing.T) {
			const src = "contract-test/copy-src.txt"
			if _, err := client.PutObject(ctx, &awss3.PutObjectInput{
				Bucket: aws.String(bucket),
				Key:    aws.String(src),
				Body:   bytes.NewReader([]byte("payload")),
			}); err != nil {
				t.Fatalf("seed the copy source: %v", err)
			}
			defer client.DeleteObject(ctx, &awss3.DeleteObjectInput{
				Bucket: aws.String(bucket), Key: aws.String(src),
			})

			const dst = "contract-test/copy-tagged.txt"
			_, err := client.CopyObject(ctx, &awss3.CopyObjectInput{
				Bucket:           aws.String(bucket),
				Key:              aws.String(dst),
				CopySource:       aws.String(bucket + "/" + src),
				TaggingDirective: s3types.TaggingDirectiveReplace,
				Tagging:          aws.String("carlos-scan=clean"),
			})
			if err == nil {
				_, _ = client.DeleteObject(ctx, &awss3.DeleteObjectInput{
					Bucket: aws.String(bucket), Key: aws.String(dst),
				})
			}
			requireDenied(t, err, "CopyObject with a REPLACE tagging directive")
		})

		// THE CONTROLS, WITHOUT WHICH THE TWO ABOVE PROVE NOTHING.
		//
		// Both assert AccessDenied. If these operations were denied anyway —
		// because the app policy never permitted multipart or copy in the
		// first place — the assertions would pass while saying nothing at all
		// about tagging, and a later policy that lost the tag-key Deny would
		// still look green. So each route is exercised again WITHOUT our tag
		// keys and must succeed: that is what makes the refusals above
		// attributable to the tag and not to the verb.
		t.Run("but multipart itself is allowed", func(t *testing.T) {
			const key = "contract-test/multipart-plain.bin"
			out, err := client.CreateMultipartUpload(ctx, &awss3.CreateMultipartUploadInput{
				Bucket: aws.String(bucket),
				Key:    aws.String(key),
			})
			if err != nil {
				t.Fatalf("CreateMultipartUpload without tags = %v — the refusal above was about the VERB, not the tag, and proves nothing", err)
			}
			_, _ = client.AbortMultipartUpload(ctx, &awss3.AbortMultipartUploadInput{
				Bucket: aws.String(bucket), Key: aws.String(key), UploadId: out.UploadId,
			})
		})

		t.Run("and copy itself is allowed", func(t *testing.T) {
			const src = "contract-test/copy-control-src.txt"
			const dst = "contract-test/copy-control-dst.txt"
			if _, err := client.PutObject(ctx, &awss3.PutObjectInput{
				Bucket: aws.String(bucket), Key: aws.String(src),
				Body: bytes.NewReader([]byte("payload")),
			}); err != nil {
				t.Fatalf("seed: %v", err)
			}
			defer client.DeleteObject(ctx, &awss3.DeleteObjectInput{
				Bucket: aws.String(bucket), Key: aws.String(src),
			})
			_, err := client.CopyObject(ctx, &awss3.CopyObjectInput{
				Bucket:     aws.String(bucket),
				Key:        aws.String(dst),
				CopySource: aws.String(bucket + "/" + src),
			})
			if err != nil {
				t.Fatalf("CopyObject without tags = %v — the refusal above was about the VERB, not the tag, and proves nothing", err)
			}
			_, _ = client.DeleteObject(ctx, &awss3.DeleteObjectInput{
				Bucket: aws.String(bucket), Key: aws.String(dst),
			})
		})

		// THE ROUTE IAM CANNOT CLOSE, AND WHAT ACTUALLY CLOSES IT.
		//
		// A browser-direct POST-policy upload carries its tags in a form
		// field. MinIO does not populate s3:RequestObjectTagKeys for that
		// handler, so the Deny that stops every other route does not fire and
		// THE UPLOAD SUCCEEDS — carlos-scan=clean and all. No IAM condition
		// key describes a form upload, and a bucket policy uses the same keys,
		// so nothing in a policy document prevents this.
		//
		// It does not need to. The tag is written and it means nothing.
		//
		// THIS TEST PERFORMS THE WHOLE ATTACK, and it has to, because the
		// obvious version of it is vacuous. A forged tag carrying a nonsense
		// carlos-scan-version is rejected on the version comparison and never
		// reaches the signature check at all — so it passes with verification
		// disabled, which is exactly what the first draft of this test did.
		//
		// The real attack is available to the app: the ETag of a single-part
		// upload is the MD5 of bytes it chose, and last-modified is the server
		// clock at upload to the second, which it can guess and retry until it
		// lands. So the loop below guesses, checks, and retries — and only
		// then asserts. A forged tag with a MATCHING version is believed by
		// every check except the one that matters.
		t.Run("a POST-policy forged tag with a matching version is still not a verdict", func(t *testing.T) {
			const forged = "contract-test/post-forged.txt"
			body := []byte("a payload the app would like called clean")
			etag := fmt.Sprintf("%x", md5.Sum(body))

			defer client.DeleteObject(ctx, &awss3.DeleteObjectInput{
				Bucket: aws.String(bucket), Key: aws.String(forged),
			})

			var landed string
			for attempt := 0; attempt < 8 && landed == ""; attempt++ {
				guess := time.Now().UTC().Truncate(time.Second)
				want := etag + "@" + guess.Format(time.RFC3339)

				status, resp := postPolicyUploadBody(t, endpoint, key, secret, bucket, forged, body,
					"<Tagging><TagSet>"+
						"<Tag><Key>carlos-scan</Key><Value>clean</Value></Tag>"+
						"<Tag><Key>carlos-scan-version</Key><Value>"+want+"</Value></Tag>"+
						"<Tag><Key>carlos-scan-at</Key><Value>"+guess.Format(time.RFC3339)+"</Value></Tag>"+
						"</TagSet></Tagging>")
				if status < 200 || status >= 300 {
					// A gateway that refuses this is a better world, and this
					// test should say so rather than quietly passing.
					t.Skipf("this gateway refused a tagged POST-policy upload (%d) — the IAM condition key reaches it here:\n%s", status, resp)
				}
				head, err := client.HeadObject(ctx, &awss3.HeadObjectInput{
					Bucket: aws.String(bucket), Key: aws.String(forged),
				})
				if err != nil {
					t.Fatal(err)
				}
				obj := ScanObject{Key: forged, ETag: normalizeETag(aws.ToString(head.ETag))}
				if head.LastModified != nil {
					obj.Modified = *head.LastModified
				}
				if obj.Version() == want {
					landed = want
				}
			}
			if landed == "" {
				t.Skip("could not land a forged tag on a matching version in eight attempts — the attack is real but this run did not reproduce it")
			}

			// The tag really is on the object, with a version that really does
			// match. Everything short of the signature now says "clean".
			tags, err := client.GetObjectTagging(ctx, &awss3.GetObjectTaggingInput{
				Bucket: aws.String(bucket), Key: aws.String(forged),
			})
			if err != nil {
				t.Fatal(err)
			}
			var gotVerdict, gotVersion string
			for _, tg := range tags.TagSet {
				switch aws.ToString(tg.Key) {
				case TagVerdict:
					gotVerdict = aws.ToString(tg.Value)
				case TagVersion:
					gotVersion = aws.ToString(tg.Value)
				}
			}
			if gotVerdict != string(VerdictClean) || gotVersion != landed {
				t.Fatalf("the forgery is not in place (verdict=%q version=%q), so the refusal below proves nothing", gotVerdict, gotVersion)
			}

			// And the part that matters. The engine is nil deliberately:
			// reaching it would mean the scanner decided to scan, which is the
			// right outcome but not what this asserts. This asserts the forged
			// verdict was not believed.
			sc := &clamavScanner{
				s3: client, engine: nil, maxBytes: 1 << 20,
				signer: testSigner(t, "a-real-deployment-key", ""),
			}
			v, err := sc.Verdict(ctx, bucket, forged, landed)
			if err != nil {
				t.Fatalf("Verdict over a forged tag: %v", err)
			}
			if v != VerdictUnknown {
				t.Fatalf("verdict = %q for a tag the APP wrote, on a real gateway, with a matching version — an app just marked its own malware clean", v)
			}
		})

		// And the same control for the plain PUT route, which has the same
		// hazard: an ordinary untagged upload must still work, or the Deny is
		// firing on every write and the store is simply broken.
		t.Run("and an ordinary tagged-with-something-else upload is allowed", func(t *testing.T) {
			const key = "contract-test/foreign-tag.txt"
			_, err := client.PutObject(ctx, &awss3.PutObjectInput{
				Bucket:  aws.String(bucket),
				Key:     aws.String(key),
				Body:    bytes.NewReader([]byte("payload")),
				Tagging: aws.String("team=payments"),
			})
			if err != nil {
				t.Fatalf("PutObject with a tag of the app's own = %v — the Deny is keyed on OUR tag names and must not touch a tenant's", err)
			}
			_, _ = client.DeleteObject(ctx, &awss3.DeleteObjectInput{
				Bucket: aws.String(bucket), Key: aws.String(key),
			})
		})
	})

	t.Run("delete removes it", func(t *testing.T) {
		_, err := client.DeleteObject(ctx, &awss3.DeleteObjectInput{
			Bucket: aws.String(bucket),
			Key:    aws.String(objectKey),
		})
		if err != nil {
			t.Fatalf("DeleteObject: %v", err)
		}
		out, err := client.ListObjectsV2(ctx, &awss3.ListObjectsV2Input{
			Bucket: aws.String(bucket),
		})
		if err != nil {
			t.Fatalf("ListObjectsV2 after delete: %v", err)
		}
		if containsKey(out.Contents, objectKey) {
			t.Fatalf("ListObjectsV2 after delete still contains %q", objectKey)
		}
	})

	t.Run("cross-bucket denial", func(t *testing.T) {
		_, err := client.PutObject(ctx, &awss3.PutObjectInput{
			Bucket: aws.String(deniedBucket),
			Key:    aws.String(objectKey),
			Body:   bytes.NewReader([]byte(body)),
		})
		if err == nil {
			t.Fatalf("PutObject to %q: want AccessDenied, got success", deniedBucket)
		}
		var apiErr smithy.APIError
		if !errors.As(err, &apiErr) {
			t.Fatalf("PutObject to %q: err = %v (%T), want a smithy.APIError", deniedBucket, err, err)
		}
		if apiErr.ErrorCode() != "AccessDenied" {
			t.Fatalf("PutObject to %q: error code = %q, want %q (message: %s)", deniedBucket, apiErr.ErrorCode(), "AccessDenied", apiErr.ErrorMessage())
		}
	})
}

// requireEnv fetches name from the environment, failing the test fast with
// a clear message if it is unset — a missing var here means the CI job's
// env block and this test's expectations have drifted, not that the
// contract itself is broken.
func requireEnv(t *testing.T, name string) string {
	t.Helper()
	v := os.Getenv(name)
	if v == "" {
		t.Fatalf("%s is not set (required when CARLOS_MINIO_TEST=1)", name)
	}
	return v
}

// containsKey and keyNames are small helpers over ListObjectsV2's Contents so
// the assertions above read as intent, not index arithmetic.
func containsKey(objs []s3types.Object, key string) bool {
	for _, o := range objs {
		if aws.ToString(o.Key) == key {
			return true
		}
	}
	return false
}

func keyNames(objs []s3types.Object) []string {
	names := make([]string, len(objs))
	for i, o := range objs {
		names[i] = aws.ToString(o.Key)
	}
	return names
}

// requireDenied is the assertion both tagging routes need: not merely "it
// failed", but AccessDenied specifically. A malformed request, a 404 or a
// signature problem would also return an error, and any of them would let a
// broken policy pass this test while proving nothing about permission.
func requireDenied(t *testing.T, err error, what string) {
	t.Helper()
	if err == nil {
		t.Fatalf("%s SUCCEEDED — an app can mark its own object clean, and every clean verdict this deployment reports is forgeable", what)
	}
	var apiErr smithy.APIError
	if !errors.As(err, &apiErr) {
		t.Fatalf("%s: err = %v (%T), want a smithy.APIError carrying AccessDenied", what, err, err)
	}
	if apiErr.ErrorCode() != "AccessDenied" {
		t.Fatalf("%s: error code = %q, want AccessDenied (message: %s)", what, apiErr.ErrorCode(), apiErr.ErrorMessage())
	}
}

// postPolicyUpload performs a browser-style direct upload: a base64 policy
// document signed with SigV4, POSTed as a multipart form. There is no helper
// for this in the AWS SDK for Go, which is a good part of why this route goes
// untested — and it is the route an app actually uses to take a file from a
// user without proxying the bytes.
//
// Hand-rolled rather than pulled in: this repo adds no third-party
// dependencies, and the whole of SigV4 for a POST policy is a four-step HMAC
// chain over a date, a region, a service and a terminator.
func postPolicyUpload(t *testing.T, endpoint, akid, secret, bucket, key, taggingXML string) (int, string) {
	t.Helper()
	return postPolicyUploadBody(t, endpoint, akid, secret, bucket, key, []byte("payload"), taggingXML)
}

// postPolicyUploadBody is the same with the bytes chosen by the caller, which
// the forgery test needs: the ETag of a single-part upload is the MD5 of those
// bytes, and predicting it is half of what makes the attack available.
func postPolicyUploadBody(t *testing.T, endpoint, akid, secret, bucket, key string, body []byte, taggingXML string) (int, string) {
	t.Helper()

	const region = "us-east-1"
	now := time.Now().UTC()
	amzDate := now.Format("20060102T150405Z")
	scopeDate := now.Format("20060102")
	credential := akid + "/" + scopeDate + "/" + region + "/s3/aws4_request"

	// The policy the browser would be handed. `starts-with` on $tagging is
	// what an app would write if it wanted the browser to choose tags — which
	// is precisely the permissive shape worth testing.
	policy := fmt.Sprintf(`{"expiration":%q,"conditions":[
		{"bucket":%q},
		{"key":%q},
		{"x-amz-algorithm":"AWS4-HMAC-SHA256"},
		{"x-amz-credential":%q},
		{"x-amz-date":%q},
		["starts-with","$tagging",""]
	]}`, now.Add(15*time.Minute).Format(time.RFC3339), bucket, key, credential, amzDate)
	encoded := base64.StdEncoding.EncodeToString([]byte(policy))

	mac := func(k []byte, s string) []byte {
		h := hmac.New(sha256.New, k)
		h.Write([]byte(s))
		return h.Sum(nil)
	}
	signing := mac(mac(mac(mac([]byte("AWS4"+secret), scopeDate), region), "s3"), "aws4_request")
	sig := hex.EncodeToString(mac(signing, encoded))

	var buf bytes.Buffer
	w := multipart.NewWriter(&buf)
	for _, f := range [][2]string{
		{"key", key},
		{"x-amz-algorithm", "AWS4-HMAC-SHA256"},
		{"x-amz-credential", credential},
		{"x-amz-date", amzDate},
		{"policy", encoded},
		{"x-amz-signature", sig},
		{"tagging", taggingXML},
	} {
		if err := w.WriteField(f[0], f[1]); err != nil {
			t.Fatal(err)
		}
	}
	// The file part must come last: S3 ignores anything after it.
	part, err := w.CreateFormFile("file", "upload.txt")
	if err != nil {
		t.Fatal(err)
	}
	if _, err := part.Write(body); err != nil {
		t.Fatal(err)
	}
	if err := w.Close(); err != nil {
		t.Fatal(err)
	}

	// Path-style, matching the client the rest of this test builds.
	req, err := http.NewRequest(http.MethodPost, strings.TrimSuffix(endpoint, "/")+"/"+bucket, &buf)
	if err != nil {
		t.Fatal(err)
	}
	req.Header.Set("Content-Type", w.FormDataContentType())

	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer resp.Body.Close()
	out, err := io.ReadAll(io.LimitReader(resp.Body, 1<<16))
	if err != nil {
		t.Fatal(err)
	}
	return resp.StatusCode, string(out)
}