package objstore

import (
	"bytes"
	"context"
	"errors"
	"io"
	"strings"
	"testing"
	"time"

	"github.com/aws/smithy-go"

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

	"github.com/carlosframework/platform/internal/clamd"
)

// stubScanS3 implements the narrow scanS3API seam with per-method hooks, in
// the same style as stubS3. An un-set hook fails the test loudly rather than
// panicking on a nil func — several tests here exist precisely to assert that
// a call did NOT happen, and that only works if an unexpected call says so.
type stubScanS3 struct {
	t *testing.T

	getObject  func(*awss3.GetObjectInput) (*awss3.GetObjectOutput, error)
	headObject func(*awss3.HeadObjectInput) (*awss3.HeadObjectOutput, error)
	getTagging func(*awss3.GetObjectTaggingInput) (*awss3.GetObjectTaggingOutput, error)
	putTagging func(*awss3.PutObjectTaggingInput) (*awss3.PutObjectTaggingOutput, error)
}

func (s *stubScanS3) GetObject(_ context.Context, in *awss3.GetObjectInput, _ ...func(*awss3.Options)) (*awss3.GetObjectOutput, error) {
	if s.getObject == nil {
		s.t.Fatal("unexpected GetObject call — bytes were fetched that should not have been")
	}
	return s.getObject(in)
}

func (s *stubScanS3) HeadObject(_ context.Context, in *awss3.HeadObjectInput, _ ...func(*awss3.Options)) (*awss3.HeadObjectOutput, error) {
	if s.headObject == nil {
		s.t.Fatal("unexpected HeadObject call")
	}
	return s.headObject(in)
}

// headOK makes HeadObject describe an object of the given shape, which is what
// Status needs in order to know the CURRENT version.
func (s *stubScanS3) headOK(etag string, modified time.Time, size int64) {
	s.headObject = func(*awss3.HeadObjectInput) (*awss3.HeadObjectOutput, error) {
		return &awss3.HeadObjectOutput{ETag: &etag, LastModified: &modified, ContentLength: &size}, nil
	}
}

func (s *stubScanS3) GetObjectTagging(_ context.Context, in *awss3.GetObjectTaggingInput, _ ...func(*awss3.Options)) (*awss3.GetObjectTaggingOutput, error) {
	if s.getTagging == nil {
		s.t.Fatal("unexpected GetObjectTagging call")
	}
	return s.getTagging(in)
}

func (s *stubScanS3) PutObjectTagging(_ context.Context, in *awss3.PutObjectTaggingInput, _ ...func(*awss3.Options)) (*awss3.PutObjectTaggingOutput, error) {
	if s.putTagging == nil {
		s.t.Fatal("unexpected PutObjectTagging call — a verdict was written that should not have been")
	}
	return s.putTagging(in)
}

// fakeEngine stands in for clamd. The wire protocol has its own tests in
// internal/clamd; what matters here is how the scanner reacts to each answer.
type fakeEngine struct {
	res     clamd.Result
	err     error
	version string

	calls int
	read  []byte
}

func (f *fakeEngine) ScanStream(_ context.Context, r io.Reader) (clamd.Result, error) {
	f.calls++
	f.read, _ = io.ReadAll(r)
	if f.err != nil {
		return clamd.Result{}, f.err
	}
	return f.res, nil
}

func (f *fakeEngine) Version(_ context.Context) (string, error) {
	if f.err != nil {
		return "", f.err
	}
	return f.version, nil
}

// allowTagRead lets tag()'s read-merge-write see an empty tag set. Scan now
// reads the existing tags before replacing them, so that a bucket CARLOS
// adopted does not lose tags someone else's tooling put there.
func (s *stubScanS3) allowTagRead(tags ...string) {
	s.getTagging = func(*awss3.GetObjectTaggingInput) (*awss3.GetObjectTaggingOutput, error) {
		return taggingOutput(tags...), nil
	}
}

func objectBody(s string) *awss3.GetObjectOutput {
	return &awss3.GetObjectOutput{Body: io.NopCloser(strings.NewReader(s))}
}

func tagsOf(in *awss3.PutObjectTaggingInput) map[string]string {
	m := map[string]string{}
	if in.Tagging == nil {
		return m
	}
	for _, t := range in.Tagging.TagSet {
		m[aws2s(t.Key)] = aws2s(t.Value)
	}
	return m
}

// An object above the ceiling is tagged skipped WITHOUT a GetObject: paying
// egress to fetch bytes clamd will refuse is pure waste, and the size is
// already on the listing page.
func TestClamAVSkipsOversizeWithoutFetching(t *testing.T) {
	var wrote map[string]string
	s3 := &stubScanS3{t: t} // getObject deliberately unset: calling it fails the test
	s3.allowTagRead()
	s3.putTagging = func(in *awss3.PutObjectTaggingInput) (*awss3.PutObjectTaggingOutput, error) {
		wrote = tagsOf(in)
		return &awss3.PutObjectTaggingOutput{}, nil
	}
	eng := &fakeEngine{}
	c := &clamavScanner{signer: testSigner(t, "test-key", ""), s3: s3, engine: eng, maxBytes: 10}

	v, _, err := c.Scan(context.Background(), "b", "big.iso", "abc", 11)
	if err != nil {
		t.Fatal(err)
	}
	if v != VerdictSkipped {
		t.Fatalf("verdict = %q, want skipped", v)
	}
	if eng.calls != 0 {
		t.Fatal("an oversize object was streamed to the engine anyway")
	}
	if wrote[TagVerdict] != string(VerdictSkipped) {
		t.Fatalf("tags = %v", wrote)
	}
}

// A verdict is written with the etag it judged, so a later rewrite reads as
// unknown rather than inheriting the old verdict.
func TestClamAVTagsVerdictWithETag(t *testing.T) {
	var wrote map[string]string
	s3 := &stubScanS3{t: t}
	s3.getObject = func(*awss3.GetObjectInput) (*awss3.GetObjectOutput, error) {
		return objectBody("hello"), nil
	}
	s3.allowTagRead()
	s3.putTagging = func(in *awss3.PutObjectTaggingInput) (*awss3.PutObjectTaggingOutput, error) {
		wrote = tagsOf(in)
		return &awss3.PutObjectTaggingOutput{}, nil
	}
	eng := &fakeEngine{res: clamd.Result{Status: clamd.StatusClean}}
	c := &clamavScanner{signer: testSigner(t, "test-key", ""), s3: s3, engine: eng, maxBytes: 1 << 20}

	v, _, err := c.Scan(context.Background(), "b", "k", "d41d8cd9@2026-09-01T10:00:00Z", 5)
	if err != nil {
		t.Fatal(err)
	}
	if v != VerdictClean {
		t.Fatalf("verdict = %q", v)
	}
	if wrote[TagVersion] != "d41d8cd9@2026-09-01T10:00:00Z" {
		t.Fatalf("verdict is not bound to the object version it judged: %v", wrote)
	}
	if wrote[TagAt] == "" {
		t.Fatalf("no scan time recorded: %v", wrote)
	}
	if !bytes.Equal(eng.read, []byte("hello")) {
		t.Fatalf("engine saw %q", eng.read)
	}
}

// clamd being unreachable is a PASS error, not thousands of failed tags.
// "The engine is unreachable" and "the engine read this object and failed on
// it" are different facts and only the second belongs in a tag.
func TestClamAVDialFailureWritesNoVerdict(t *testing.T) {
	s3 := &stubScanS3{t: t} // putTagging unset: any write fails the test
	s3.getObject = func(*awss3.GetObjectInput) (*awss3.GetObjectOutput, error) {
		return objectBody("hello"), nil
	}
	eng := &fakeEngine{err: errors.New("dial unix /run/clamav/clamd.ctl: connection refused")}
	c := &clamavScanner{signer: testSigner(t, "test-key", ""), s3: s3, engine: eng, maxBytes: 1 << 20}

	v, _, err := c.Scan(context.Background(), "b", "k", "e", 5)
	if err == nil {
		t.Fatal("an unreachable engine reported success")
	}
	if v != VerdictUnknown {
		t.Fatalf("verdict = %q, want unknown", v)
	}
}

// A mid-stream size refusal maps to skipped, not failed.
func TestClamAVSizeExceededMapsToSkipped(t *testing.T) {
	var wrote map[string]string
	s3 := &stubScanS3{t: t}
	s3.getObject = func(*awss3.GetObjectInput) (*awss3.GetObjectOutput, error) {
		return objectBody("hello"), nil
	}
	s3.allowTagRead()
	s3.putTagging = func(in *awss3.PutObjectTaggingInput) (*awss3.PutObjectTaggingOutput, error) {
		wrote = tagsOf(in)
		return &awss3.PutObjectTaggingOutput{}, nil
	}
	eng := &fakeEngine{res: clamd.Result{Status: clamd.StatusSizeExceeded}}
	c := &clamavScanner{signer: testSigner(t, "test-key", ""), s3: s3, engine: eng, maxBytes: 0} // no local ceiling: the engine's own

	v, _, err := c.Scan(context.Background(), "b", "k", "e", 999)
	if err != nil {
		t.Fatal(err)
	}
	if v != VerdictSkipped {
		t.Fatalf("verdict = %q, want skipped", v)
	}
	if wrote[TagVerdict] != string(VerdictSkipped) {
		t.Fatalf("tags = %v", wrote)
	}
}

// An engine ERROR on an object it did read IS a per-object fact, and does get
// recorded — the mirror of the unreachable case above.
func TestClamAVEngineErrorIsRecordedAsFailed(t *testing.T) {
	var wrote map[string]string
	s3 := &stubScanS3{t: t}
	s3.getObject = func(*awss3.GetObjectInput) (*awss3.GetObjectOutput, error) {
		return objectBody("hello"), nil
	}
	s3.allowTagRead()
	s3.putTagging = func(in *awss3.PutObjectTaggingInput) (*awss3.PutObjectTaggingOutput, error) {
		wrote = tagsOf(in)
		return &awss3.PutObjectTaggingOutput{}, nil
	}
	eng := &fakeEngine{res: clamd.Result{Status: clamd.StatusError, Raw: "stream: broken ERROR"}}
	c := &clamavScanner{signer: testSigner(t, "test-key", ""), s3: s3, engine: eng, maxBytes: 1 << 20}

	v, _, err := c.Scan(context.Background(), "b", "k", "e", 5)
	if err != nil {
		t.Fatal(err)
	}
	if v != VerdictFailed || wrote[TagVerdict] != string(VerdictFailed) {
		t.Fatalf("verdict = %q, tags = %v", v, wrote)
	}
}

// An infected object carries the verdict the store policy denies on. If these
// two strings ever drift apart, enforcement silently stops matching anything.
func TestClamAVInfectedVerdictMatchesEnforcementTag(t *testing.T) {
	s3 := &stubScanS3{t: t}
	s3.getObject = func(*awss3.GetObjectInput) (*awss3.GetObjectOutput, error) {
		return objectBody("x5O!P%"), nil
	}
	s3.allowTagRead()
	var wrote map[string]string
	s3.putTagging = func(in *awss3.PutObjectTaggingInput) (*awss3.PutObjectTaggingOutput, error) {
		wrote = tagsOf(in)
		return &awss3.PutObjectTaggingOutput{}, nil
	}
	eng := &fakeEngine{res: clamd.Result{Status: clamd.StatusInfected, Signature: "Eicar-Signature"}}
	c := &clamavScanner{signer: testSigner(t, "test-key", ""), s3: s3, engine: eng, maxBytes: 1 << 20}

	if _, _, err := c.Scan(context.Background(), "b", "k", "e", 6); err != nil {
		t.Fatal(err)
	}
	deny := c.Enforcement()
	if wrote[deny.Key] != deny.Value {
		t.Fatalf("the tag written (%v) does not match the tag denied on (%+v)", wrote, deny)
	}
}

func taggingOutput(pairs ...string) *awss3.GetObjectTaggingOutput {
	var set []s3types.Tag
	for i := 0; i+1 < len(pairs); i += 2 {
		k, v := pairs[i], pairs[i+1]
		set = append(set, s3types.Tag{Key: &k, Value: &v})
	}
	return &awss3.GetObjectTaggingOutput{TagSet: set}
}

// A verdict is bound to the bytes it judged. An object rewritten under the
// same key reads as unknown, so the sweep scans it again — otherwise anyone
// could launder a payload through a key that once held something clean.
func TestClamAVVerdictIgnoresStaleETag(t *testing.T) {
	s3 := &stubScanS3{t: t}
	s3.getTagging = func(*awss3.GetObjectTaggingInput) (*awss3.GetObjectTaggingOutput, error) {
		return taggingOutput(TagVerdict, "clean", TagVersion, "old-etag"), nil
	}
	c := &clamavScanner{signer: testSigner(t, "test-key", ""), s3: s3, engine: &fakeEngine{}}

	v, err := c.Verdict(context.Background(), "b", "k", "new-etag")
	if err != nil {
		t.Fatal(err)
	}
	if v != VerdictUnknown {
		t.Fatalf("a rewritten object inherited its old verdict: %q", v)
	}
}

// S3 hands back a QUOTED ETag. Normalisation happens once, where the version
// token is built, so a stored tag and a fresh listing compare equal. If they
// never did, every object would re-scan on every sweep forever — an invisible
// bug that costs money rather than correctness.
func TestScanObjectVersionNormalisesQuoting(t *testing.T) {
	at := time.Date(2026, 9, 1, 10, 0, 0, 0, time.UTC)
	quoted := ScanObject{Key: "k", ETag: `"d41d8cd9"`, Modified: at}
	bare := ScanObject{Key: "k", ETag: "d41d8cd9", Modified: at}
	if quoted.Version() != bare.Version() {
		t.Fatalf("quoting changed the version token: %q vs %q", quoted.Version(), bare.Version())
	}
}

// The version moves when the object is overwritten, even if the bytes hash to
// the same ETag. That is the whole point of folding last-modified in: a
// single-part upload's ETag is the content MD5, and chosen-prefix MD5
// collisions are practical — an attacker controlling both uploads could give a
// payload the ETag of the clean object it replaces and inherit its verdict.
// Last-modified is not theirs to choose.
func TestScanObjectVersionChangesOnOverwriteWithCollidingETag(t *testing.T) {
	etag := "d41d8cd9" // same hash, as a collision would give
	before := ScanObject{Key: "k", ETag: etag, Modified: time.Date(2026, 9, 1, 10, 0, 0, 0, time.UTC)}
	after := ScanObject{Key: "k", ETag: etag, Modified: time.Date(2026, 9, 1, 11, 0, 0, 0, time.UTC)}
	if before.Version() == after.Version() {
		t.Fatalf("an overwrite kept the same version token (%q) — the old verdict would be inherited",
			before.Version())
	}
}

// An untagged object is unknown, not clean. Defaulting the other way would
// report a bucket nobody has ever scanned as fully clean.
func TestClamAVUntaggedObjectIsUnknown(t *testing.T) {
	s3 := &stubScanS3{t: t}
	s3.getTagging = func(*awss3.GetObjectTaggingInput) (*awss3.GetObjectTaggingOutput, error) {
		return taggingOutput(), nil
	}
	c := &clamavScanner{signer: testSigner(t, "test-key", ""), s3: s3, engine: &fakeEngine{}}

	v, err := c.Verdict(context.Background(), "b", "k", "e")
	if err != nil {
		t.Fatal(err)
	}
	if v != VerdictUnknown {
		t.Fatalf("an object nobody scanned reported %q", v)
	}
}

// Nothing is attached to the bucket, so nothing is consumed and nothing leaks.
// This is the whole reason ClamAV is the primary engine rather than the bought
// alternative, whose per-bucket plans cap at 25 per account per Region.
func TestClamAVAttachesNothingToTheBucket(t *testing.T) {
	ctx := context.Background()
	c := &clamavScanner{signer: testSigner(t, "test-key", ""), s3: &stubScanS3{t: t}, engine: &fakeEngine{}}

	if err := c.EnsureEnabled(ctx, "b"); err != nil {
		t.Fatal(err)
	}
	if err := c.EnsureDisabled(ctx, "b"); err != nil {
		t.Fatal(err)
	}
	remaining, capacity, err := c.Capacity(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if remaining != -1 || capacity != -1 {
		t.Fatalf("capacity = %d/%d, want -1/-1 (unbounded)", remaining, capacity)
	}
}

// Provenance carries the signature database, not just the engine name. "Clean"
// without a database version is not evidence of anything.
func TestClamAVProvenanceCarriesSignatureVersion(t *testing.T) {
	eng := &fakeEngine{version: "ClamAV 1.4.1/27500/Mon Sep  1 08:00:00 2026"}
	c := &clamavScanner{signer: testSigner(t, "test-key", ""), s3: &stubScanS3{t: t}, engine: eng}

	engine, version, err := c.Provenance(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	if engine != "clamav" {
		t.Fatalf("engine = %q", engine)
	}
	if !strings.Contains(version, "27500") {
		t.Fatalf("version = %q, want the signature database in it", version)
	}
}

// apiErr is an AWS API error with a given code, which is what isDenied and
// isGone classify on.
func apiErr(code string) error {
	return &smithy.GenericAPIError{Code: code, Message: code}
}

// An object we are not allowed to read is DENIED, and the sweep goes on.
//
// This is not tidiness. A plain error here fails the whole page unwritten; the
// next tick re-lists the same page, meets the same object and fails
// identically, and the bucket's sweep never advances past it. One SSE-KMS
// object outside the console's key policy would stall a store's coverage
// forever while every surface said "in progress".
func TestClamAVUnreadableObjectIsDeniedNotAPassFailure(t *testing.T) {
	for _, code := range []string{"AccessDenied", "AccessDeniedException", "Forbidden", "InvalidObjectState"} {
		t.Run(code, func(t *testing.T) {
			var wrote map[string]string
			s3 := &stubScanS3{t: t}
			s3.getObject = func(*awss3.GetObjectInput) (*awss3.GetObjectOutput, error) {
				return nil, apiErr(code)
			}
			s3.allowTagRead()
			s3.putTagging = func(in *awss3.PutObjectTaggingInput) (*awss3.PutObjectTaggingOutput, error) {
				wrote = tagsOf(in)
				return &awss3.PutObjectTaggingOutput{}, nil
			}
			c := &clamavScanner{signer: testSigner(t, "test-key", ""), s3: s3, engine: &fakeEngine{}, maxBytes: 1 << 20}

			v, _, err := c.Scan(context.Background(), "b", "k", "e", 5)
			if err != nil {
				t.Fatalf("an unreadable object failed the pass: %v", err)
			}
			if v != VerdictDenied {
				t.Fatalf("verdict = %q, want denied", v)
			}
			// The tag is an audit trace, not an answer: the next sweep
			// re-issues denied rather than reading it back. See
			// TestClamAVDeniedAndFailedAreReIssuedNotInherited.
			if wrote[TagVerdict] != string(VerdictDenied) {
				t.Fatalf("tags = %v", wrote)
			}
		})
	}
}

// An object deleted between the listing and the fetch is GONE — neither
// covered nor a gap. Routine on any bucket in use, and it must not stall the
// page either.
func TestClamAVDeletedObjectIsGone(t *testing.T) {
	s3 := &stubScanS3{t: t} // putTagging unset: nothing may be written
	s3.getObject = func(*awss3.GetObjectInput) (*awss3.GetObjectOutput, error) {
		return nil, apiErr("NoSuchKey")
	}
	c := &clamavScanner{signer: testSigner(t, "test-key", ""), s3: s3, engine: &fakeEngine{}, maxBytes: 1 << 20}

	if _, _, err := c.Scan(context.Background(), "b", "k", "e", 5); !errors.Is(err, ErrObjectGone) {
		t.Fatalf("err = %v, want ErrObjectGone", err)
	}
}

// Unreadable TAGS are denied too, by the same argument: Verdict is called for
// every object on every page, so an error there stalls just as hard.
func TestClamAVUnreadableTagsAreDenied(t *testing.T) {
	s3 := &stubScanS3{t: t}
	s3.getTagging = func(*awss3.GetObjectTaggingInput) (*awss3.GetObjectTaggingOutput, error) {
		return nil, apiErr("AccessDenied")
	}
	c := &clamavScanner{signer: testSigner(t, "test-key", ""), s3: s3, engine: &fakeEngine{}}

	v, err := c.Verdict(context.Background(), "b", "k", "e")
	if err != nil {
		t.Fatalf("an unreadable tag set failed the pass: %v", err)
	}
	if v != VerdictDenied {
		t.Fatalf("verdict = %q, want denied", v)
	}
}

// A genuine transport failure is still a pass failure — the classification
// must not swallow everything.
func TestClamAVTransportErrorStillFailsThePass(t *testing.T) {
	s3 := &stubScanS3{t: t}
	s3.getObject = func(*awss3.GetObjectInput) (*awss3.GetObjectOutput, error) {
		return nil, apiErr("InternalError")
	}
	c := &clamavScanner{signer: testSigner(t, "test-key", ""), s3: s3, engine: &fakeEngine{}, maxBytes: 1 << 20}

	v, _, err := c.Scan(context.Background(), "b", "k", "e", 5)
	if err == nil {
		t.Fatal("a transport failure was recorded as a verdict")
	}
	if v != VerdictUnknown {
		t.Fatalf("verdict = %q, want unknown", v)
	}
}

// An adopted bucket's objects can carry tags written long before CARLOS ever
// scoped credentials to it — lifecycle rules, cost allocation. PutObjectTagging
// replaces the whole set, so those have to be carried across explicitly.
// Deleting a tenant's tags silently on first scan would be a real loss with no
// trace of what went.
func TestClamAVPreservesForeignTags(t *testing.T) {
	var wrote map[string]string
	s3 := &stubScanS3{t: t}
	s3.getObject = func(*awss3.GetObjectInput) (*awss3.GetObjectOutput, error) {
		return objectBody("hello"), nil
	}
	s3.allowTagRead("cost-centre", "research", TagVerdict, "clean", TagVersion, "stale")
	s3.putTagging = func(in *awss3.PutObjectTaggingInput) (*awss3.PutObjectTaggingOutput, error) {
		wrote = tagsOf(in)
		return &awss3.PutObjectTaggingOutput{}, nil
	}
	c := &clamavScanner{signer: testSigner(t, "test-key", ""), s3: s3, engine: &fakeEngine{res: clamd.Result{Status: clamd.StatusClean}}, maxBytes: 1 << 20}

	if _, _, err := c.Scan(context.Background(), "b", "k", "fresh", 5); err != nil {
		t.Fatal(err)
	}
	if wrote["cost-centre"] != "research" {
		t.Fatalf("a tag this scanner does not own was destroyed: %v", wrote)
	}
	// Its own tags ARE replaced, not duplicated.
	if wrote[TagVersion] != "fresh" {
		t.Fatalf("the scanner did not overwrite its own stale etag tag: %v", wrote)
	}
}

// denied and failed are RE-ISSUED every sweep, never carried forward.
//
// Both are transient — a permissions fix, a Glacier restore, an engine having
// a bad day. Reading them back as answers would make them permanent: the
// object is not rewritten, so its version never changes, so it would never be
// retried and the repair would never take. The SSE-KMS case is the one that
// bites, because tagging needs no KMS access and so the tag does land.
func TestClamAVDeniedAndFailedAreReIssuedNotInherited(t *testing.T) {
	for _, stuck := range []Verdict{VerdictDenied, VerdictFailed} {
		t.Run(string(stuck), func(t *testing.T) {
			sg := testSigner(t, "test-key", "")
			s3 := &stubScanS3{t: t}
			s3.allowTagRead(signedTagPairs(t, sg, "b", "k", "v1", stuck)...)
			c := &clamavScanner{signer: sg, s3: s3, engine: &fakeEngine{}}

			v, err := c.Verdict(context.Background(), "b", "k", "v1")
			if err != nil {
				t.Fatal(err)
			}
			if v != VerdictUnknown {
				t.Fatalf("verdict = %q, want unknown — %q must be retried, not inherited", v, stuck)
			}
		})
	}

	// clean and infected ARE carried forward: they are answers about bytes
	// that have not changed.
	for _, keep := range []Verdict{VerdictClean, VerdictInfected, VerdictSkipped} {
		t.Run(string(keep), func(t *testing.T) {
			sg := testSigner(t, "test-key", "")
			s3 := &stubScanS3{t: t}
			s3.allowTagRead(signedTagPairs(t, sg, "b", "k", "v1", keep)...)
			c := &clamavScanner{signer: sg, s3: s3, engine: &fakeEngine{}}
			v, err := c.Verdict(context.Background(), "b", "k", "v1")
			if err != nil {
				t.Fatal(err)
			}
			if v != keep {
				t.Fatalf("verdict = %q, want %q carried forward", v, keep)
			}
		})
	}
}

// A tag set we cannot READ must never be written over. PutObjectTagging
// replaces the whole set, so writing blind here destroys exactly the foreign
// tags the merge exists to protect — on the one path where we know least about
// what is there.
func TestClamAVDoesNotBlindWriteWhenTagReadFails(t *testing.T) {
	s3 := &stubScanS3{t: t} // putTagging unset: any write fails the test
	s3.getObject = func(*awss3.GetObjectInput) (*awss3.GetObjectOutput, error) {
		return objectBody("hello"), nil
	}
	s3.getTagging = func(*awss3.GetObjectTaggingInput) (*awss3.GetObjectTaggingOutput, error) {
		return nil, apiErr("SlowDown")
	}
	c := &clamavScanner{signer: testSigner(t, "test-key", ""), s3: s3, engine: &fakeEngine{res: clamd.Result{Status: clamd.StatusClean}}, maxBytes: 1 << 20}

	if _, _, err := c.Scan(context.Background(), "b", "k", "v1", 5); err == nil {
		t.Fatal("an unreadable tag set was written over anyway")
	}
}

// An object already carrying eight foreign tags cannot also hold our three —
// S3 caps an object at ten. Erroring there would abandon the page unwritten
// every tick forever, which is the exact permanent stall the denied
// classification was added to remove. Record the verdict, skip the write.
func TestClamAVSkipsTagWriteRatherThanStallAtTheTagLimit(t *testing.T) {
	var foreign []string
	for i := 0; i < 8; i++ {
		foreign = append(foreign, "own-"+string(rune('a'+i)), "v")
	}
	s3 := &stubScanS3{t: t} // putTagging unset: an 11-tag write fails the test
	s3.getObject = func(*awss3.GetObjectInput) (*awss3.GetObjectOutput, error) {
		return objectBody("hello"), nil
	}
	s3.allowTagRead(foreign...)
	c := &clamavScanner{signer: testSigner(t, "test-key", ""), s3: s3, engine: &fakeEngine{res: clamd.Result{Status: clamd.StatusClean}}, maxBytes: 1 << 20}

	v, _, err := c.Scan(context.Background(), "b", "k", "v1", 5)
	if err != nil {
		t.Fatalf("a heavily tagged object stalled the sweep: %v", err)
	}
	if v != VerdictClean {
		t.Fatalf("verdict = %q, want clean — it was scanned, just not recorded", v)
	}
}

// Status finds the CURRENT version itself and hands back the object, so a
// caller holding only a key can go on to Scan without a HeadObject of its own.
//
// That second return is the whole point: Scan needs a version and a size, the
// on-demand drain has no listing page, and asking it to reconstruct the token
// would leak an internal format and hand it the laundering bug as a
// correctness bug.
func TestStatusReturnsObjectSoCallerCanScan(t *testing.T) {
	mod := time.Date(2026, 9, 2, 8, 0, 0, 0, time.UTC)
	s3 := &stubScanS3{t: t}
	s3.headOK(`"d41d8cd9"`, mod, 4096)
	s3.getTagging = func(*awss3.GetObjectTaggingInput) (*awss3.GetObjectTaggingOutput, error) {
		return taggingOutput(), nil // no tag yet
	}
	c := &clamavScanner{signer: testSigner(t, "test-key", ""), s3: s3, engine: &fakeEngine{}}

	v, obj, err := c.Status(context.Background(), "b", "k")
	if err != nil {
		t.Fatal(err)
	}
	if v != VerdictUnknown {
		t.Fatalf("verdict = %q, want unknown for an untagged object", v)
	}
	if obj.Size != 4096 || obj.ETag != "d41d8cd9" || !obj.Modified.Equal(mod) {
		t.Fatalf("object = %+v — a caller cannot Scan without these", obj)
	}
	if obj.Version() == "" {
		t.Fatal("no version token to scan against")
	}
}

// Status reports what the tag SAYS. Verdict decides what the SWEEP does about
// it. Conflating them makes a failed scan read as pending forever: the drain
// deletes the item as scanned, and nothing would ever change the answer.
func TestStatusDoesNotApplyTheReIssueRule(t *testing.T) {
	mod := time.Date(2026, 9, 2, 8, 0, 0, 0, time.UTC)
	version := ScanObject{ETag: "abc", Modified: mod}.Version()

	for _, stuck := range []Verdict{VerdictDenied, VerdictFailed} {
		t.Run(string(stuck), func(t *testing.T) {
			sg := testSigner(t, "test-key", "")
			s3 := &stubScanS3{t: t}
			s3.headOK("abc", mod, 10)
			s3.allowTagRead(signedTagPairs(t, sg, "b", "k", version, stuck)...)
			c := &clamavScanner{signer: sg, s3: s3, engine: &fakeEngine{}}

			// Status: as written.
			got, _, err := c.Status(context.Background(), "b", "k")
			if err != nil {
				t.Fatal(err)
			}
			if got != stuck {
				t.Fatalf("Status = %q, want %q reported as itself", got, stuck)
			}

			// Verdict: re-issued, so the sweep retries it.
			sweepView, err := c.Verdict(context.Background(), "b", "k", version)
			if err != nil {
				t.Fatal(err)
			}
			if sweepView != VerdictUnknown {
				t.Fatalf("Verdict = %q, want unknown so the sweep re-scans", sweepView)
			}
		})
	}
}

// A tag about bytes that have since changed is not an answer about the bytes
// that are there now.
func TestStatusIgnoresStaleTag(t *testing.T) {
	mod := time.Date(2026, 9, 2, 8, 0, 0, 0, time.UTC)
	s3 := &stubScanS3{t: t}
	s3.headOK("new-etag", mod, 10)
	s3.getTagging = func(*awss3.GetObjectTaggingInput) (*awss3.GetObjectTaggingOutput, error) {
		return taggingOutput(TagVerdict, "clean",
			TagVersion, ScanObject{ETag: "old-etag", Modified: mod}.Version()), nil
	}
	c := &clamavScanner{signer: testSigner(t, "test-key", ""), s3: s3, engine: &fakeEngine{}}

	v, _, err := c.Status(context.Background(), "b", "k")
	if err != nil {
		t.Fatal(err)
	}
	if v != VerdictUnknown {
		t.Fatalf("a rewritten object reported %q from its predecessor's tag", v)
	}

	// THE case the version binding exists for, and the one the check above
	// does not make: the ETag is IDENTICAL and only the timestamp moved.
	// Single-part ETag is the content MD5, and chosen-prefix MD5 collisions
	// are practical — so an attacker who can get one file scanned can upload a
	// second with the same ETag and inherit its clean verdict. Comparing only
	// the half before the "@" passes the first assertion and fails this one.
	s3.headOK("same-etag", mod.Add(time.Hour), 10)
	s3.getTagging = func(*awss3.GetObjectTaggingInput) (*awss3.GetObjectTaggingOutput, error) {
		return taggingOutput(TagVerdict, "clean",
			TagVersion, ScanObject{ETag: "same-etag", Modified: mod}.Version()), nil
	}
	v, _, err = c.Status(context.Background(), "b", "k")
	if err != nil {
		t.Fatal(err)
	}
	if v != VerdictUnknown {
		t.Fatalf("an object rewritten with a colliding ETag reported %q — the timestamp half of the version is not being compared", v)
	}
}

// A denied HeadObject leaves no version and no size, so the caller must not go
// on to Scan — the zero ScanObject is how it knows.
func TestStatusOnDeniedHead(t *testing.T) {
	s3 := &stubScanS3{t: t}
	s3.headObject = func(*awss3.HeadObjectInput) (*awss3.HeadObjectOutput, error) {
		return nil, apiErr("AccessDenied")
	}
	c := &clamavScanner{signer: testSigner(t, "test-key", ""), s3: s3, engine: &fakeEngine{}}

	v, obj, err := c.Status(context.Background(), "b", "k")
	if err != nil {
		t.Fatal(err)
	}
	if v != VerdictDenied {
		t.Fatalf("verdict = %q, want denied", v)
	}
	if obj != (ScanObject{}) {
		t.Fatalf("object = %+v, want the zero value so nobody tries to scan it", obj)
	}
}

func TestStatusOnMissingObject(t *testing.T) {
	s3 := &stubScanS3{t: t}
	s3.headObject = func(*awss3.HeadObjectInput) (*awss3.HeadObjectOutput, error) {
		return nil, apiErr("NotFound")
	}
	c := &clamavScanner{signer: testSigner(t, "test-key", ""), s3: s3, engine: &fakeEngine{}}

	if _, _, err := c.Status(context.Background(), "b", "k"); !errors.Is(err, ErrObjectGone) {
		t.Fatalf("err = %v, want ErrObjectGone", err)
	}
}

// A verdict reached but not persisted says so. Without the flag the on-demand
// drain would report clean, delete the item, and the status door would answer
// pending forever — the app's source of truth never arriving while a callback
// said otherwise.
func TestScanReportsUnrecordedAtTheTagLimit(t *testing.T) {
	var foreign []string
	for i := 0; i < 8; i++ {
		foreign = append(foreign, "own-"+string(rune('a'+i)), "v")
	}
	s3 := &stubScanS3{t: t} // putTagging unset: an 11-tag write fails the test
	s3.getObject = func(*awss3.GetObjectInput) (*awss3.GetObjectOutput, error) {
		return objectBody("hello"), nil
	}
	s3.allowTagRead(foreign...)
	c := &clamavScanner{signer: testSigner(t, "test-key", ""), s3: s3, engine: &fakeEngine{res: clamd.Result{Status: clamd.StatusClean}}, maxBytes: 1 << 20}

	v, recorded, err := c.Scan(context.Background(), "b", "k", "v1", 5)
	if err != nil {
		t.Fatal(err)
	}
	if v != VerdictClean {
		t.Fatalf("verdict = %q, want clean — it WAS scanned", v)
	}
	if recorded {
		t.Fatal("recorded = true, but nothing was written to the object")
	}
}

// The ordinary path records, so `recorded` is not simply always false.
func TestScanReportsRecordedOnTheOrdinaryPath(t *testing.T) {
	s3 := &stubScanS3{t: t}
	s3.getObject = func(*awss3.GetObjectInput) (*awss3.GetObjectOutput, error) {
		return objectBody("hello"), nil
	}
	s3.allowTagRead()
	s3.putTagging = func(*awss3.PutObjectTaggingInput) (*awss3.PutObjectTaggingOutput, error) {
		return &awss3.PutObjectTaggingOutput{}, nil
	}
	c := &clamavScanner{signer: testSigner(t, "test-key", ""), s3: s3, engine: &fakeEngine{res: clamd.Result{Status: clamd.StatusClean}}, maxBytes: 1 << 20}

	v, recorded, err := c.Scan(context.Background(), "b", "k", "v1", 5)
	if err != nil || v != VerdictClean || !recorded {
		t.Fatalf("= %q, %v, %v", v, recorded, err)
	}
}

// A denied fetch is never "recorded", whether or not the best-effort tag
// landed: nothing may depend on a trace we could not verify.
func TestScanDeniedIsNeverRecorded(t *testing.T) {
	s3 := &stubScanS3{t: t}
	s3.getObject = func(*awss3.GetObjectInput) (*awss3.GetObjectOutput, error) {
		return nil, apiErr("AccessDenied")
	}
	s3.allowTagRead()
	s3.putTagging = func(*awss3.PutObjectTaggingInput) (*awss3.PutObjectTaggingOutput, error) {
		return &awss3.PutObjectTaggingOutput{}, nil
	}
	c := &clamavScanner{signer: testSigner(t, "test-key", ""), s3: s3, engine: &fakeEngine{}, maxBytes: 1 << 20}

	v, recorded, err := c.Scan(context.Background(), "b", "k", "v1", 5)
	if err != nil || v != VerdictDenied {
		t.Fatalf("= %q, %v", v, err)
	}
	if recorded {
		t.Fatal("a denied object reported its verdict as recorded")
	}
}

// AN INFECTED VERDICT MUST ALWAYS BE RECORDABLE, AND FOREIGN TAGS DO NOT
// OUTRANK IT.
//
// The tag limit tolerance above is right for a clean verdict: the object is
// simply re-scanned next tick, and an app's own tags — which on an ADOPTED
// bucket may predate CARLOS entirely — are worth more than one avoided
// re-scan. For infected it is a bypass of the whole enforcement mechanism.
//
// `scan enforce on` writes a Deny keyed on `s3:ExistingObjectTag/carlos-scan =
// infected` (aws.go). That Deny only ever engages if the tag gets WRITTEN. An
// app that attaches eight tags of its own to an upload makes its object
// permanently unmarkable: the verdict is reached, the write is skipped, the
// Deny never matches, and the malware stays readable through the app's own
// credential. Nothing alarms — the sweep counts the object as reached, the
// result object says recorded:false, and the status door reads no tag and
// answers unknown.
//
// So: for infected, foreign tags are evicted to make room. That is a real
// cost to a tenant's metadata and it is the right trade, because the
// alternative is a known-infected object that enforcement cannot touch.
func TestInfectedIsRecordedEvenWhenForeignTagsCrowdItOut(t *testing.T) {
	var foreign []string
	for i := 0; i < 8; i++ {
		foreign = append(foreign, "own-"+string(rune('a'+i)), "v")
	}
	var written []s3types.Tag
	s3 := &stubScanS3{t: t}
	s3.getObject = func(*awss3.GetObjectInput) (*awss3.GetObjectOutput, error) {
		return objectBody("hello"), nil
	}
	s3.allowTagRead(foreign...)
	s3.putTagging = func(in *awss3.PutObjectTaggingInput) (*awss3.PutObjectTaggingOutput, error) {
		written = in.Tagging.TagSet
		return &awss3.PutObjectTaggingOutput{}, nil
	}
	c := &clamavScanner{signer: testSigner(t, "test-key", ""),
		s3:       s3,
		engine:   &fakeEngine{res: clamd.Result{Status: clamd.StatusInfected, Signature: "Eicar-Test"}},
		maxBytes: 1 << 20,
	}

	v, recorded, err := c.Scan(context.Background(), "b", "k", "v1", 5)
	if err != nil {
		t.Fatal(err)
	}
	if v != VerdictInfected {
		t.Fatalf("verdict = %q, want infected", v)
	}
	if !recorded {
		t.Fatal("an infected verdict went unrecorded — `scan enforce on` denies on the TAG, so an object that cannot be tagged cannot be quarantined")
	}
	if len(written) == 0 {
		t.Fatal("nothing was written to the object at all")
	}
	if len(written) > maxObjectTags {
		t.Fatalf("wrote %d tags, over S3's limit of %d — the write would be rejected", len(written), maxObjectTags)
	}
	var gotVerdict string
	for _, tg := range written {
		if aws2s(tg.Key) == TagVerdict {
			gotVerdict = aws2s(tg.Value)
		}
	}
	if gotVerdict != string(VerdictInfected) {
		t.Fatalf("the written set carries carlos-scan=%q, want infected:\n%v", gotVerdict, written)
	}
}

// The eviction is only as large as it needs to be: our three go in, and as
// many foreign tags as still fit are kept. Dropping all of them would be a
// simpler implementation and a gratuitous one — the object is a tenant's, and
// this takes the minimum the limit forces.
func TestInfectedEvictsOnlyAsManyForeignTagsAsItMustNot(t *testing.T) {
	var foreign []string
	for i := 0; i < 8; i++ {
		foreign = append(foreign, "own-"+string(rune('a'+i)), "v")
	}
	var written []s3types.Tag
	s3 := &stubScanS3{t: t}
	s3.getObject = func(*awss3.GetObjectInput) (*awss3.GetObjectOutput, error) {
		return objectBody("hello"), nil
	}
	s3.allowTagRead(foreign...)
	s3.putTagging = func(in *awss3.PutObjectTaggingInput) (*awss3.PutObjectTaggingOutput, error) {
		written = in.Tagging.TagSet
		return &awss3.PutObjectTaggingOutput{}, nil
	}
	c := &clamavScanner{signer: testSigner(t, "test-key", ""),
		s3:       s3,
		engine:   &fakeEngine{res: clamd.Result{Status: clamd.StatusInfected, Signature: "Eicar-Test"}},
		maxBytes: 1 << 20,
	}
	if _, _, err := c.Scan(context.Background(), "b", "k", "v1", 5); err != nil {
		t.Fatal(err)
	}

	var kept int
	for _, tg := range written {
		if !isScanTag(aws2s(tg.Key)) {
			kept++
		}
	}
	// Ten minus however many WE own — derived, because that number went from
	// three to four when the signature tag landed and a literal would have
	// silently started asserting an eleven-tag write.
	var owned int
	for _, tg := range written {
		if isScanTag(aws2s(tg.Key)) {
			owned++
		}
	}
	if want := maxObjectTags - owned; kept != want {
		t.Fatalf("kept %d foreign tags, want %d — evict the minimum the limit forces, not all of them:\n%v", kept, want, written)
	}
}

// And a CLEAN verdict still yields, which is the behaviour the two tests above
// this one pin. The split is deliberate: clean costs one re-scan, infected
// costs the enforcement Deny never engaging.
func TestCleanStillYieldsToForeignTagsAtTheLimit(t *testing.T) {
	var foreign []string
	for i := 0; i < 8; i++ {
		foreign = append(foreign, "own-"+string(rune('a'+i)), "v")
	}
	s3 := &stubScanS3{t: t} // putTagging unset: any write fails the test
	s3.getObject = func(*awss3.GetObjectInput) (*awss3.GetObjectOutput, error) {
		return objectBody("hello"), nil
	}
	s3.allowTagRead(foreign...)
	c := &clamavScanner{signer: testSigner(t, "test-key", ""), s3: s3, engine: &fakeEngine{res: clamd.Result{Status: clamd.StatusClean}}, maxBytes: 1 << 20}

	v, recorded, err := c.Scan(context.Background(), "b", "k", "v1", 5)
	if err != nil {
		t.Fatal(err)
	}
	if v != VerdictClean || recorded {
		t.Fatalf("clean = %q recorded=%v, want clean and unrecorded — a tenant's tags outrank one avoided re-scan", v, recorded)
	}
}

// THE FORGERY THE IAM DENYS CANNOT REACH, refused at the only place that
// matters: the read.
//
// A POST-policy form upload sets carlos-scan freely on MinIO with every Deny
// in place, because no IAM condition key describes a form upload. So the app
// CAN write these three tags. What it cannot do is make them mean anything:
// without a signature this deployment issued, readTag reports no verdict and
// the object is scanned like any other.
//
// This is the test that would have caught all four of the tagging defects at
// once, and it does not care which route the tag arrived by.
func TestForgedTagWithoutASignatureIsNotAVerdict(t *testing.T) {
	s3 := &stubScanS3{t: t}
	s3.getObject = func(*awss3.GetObjectInput) (*awss3.GetObjectOutput, error) {
		return objectBody("hello"), nil
	}
	s3.headObject = func(*awss3.HeadObjectInput) (*awss3.HeadObjectOutput, error) {
		return &awss3.HeadObjectOutput{ETag: s2aws(`"etag1"`)}, nil
	}
	// Exactly what an app can write for itself, and nothing else.
	s3.allowTagRead(TagVerdict, string(VerdictClean), TagVersion, "v1", TagAt, "2026-09-05T00:00:00Z")
	c := &clamavScanner{
		s3:       s3,
		engine:   &fakeEngine{res: clamd.Result{Status: clamd.StatusClean}},
		maxBytes: 1 << 20,
		signer:   testSigner(t, "deployment-key", ""),
	}

	v, err := c.Verdict(context.Background(), "b", "k", "v1")
	if err != nil {
		t.Fatal(err)
	}
	if v != VerdictUnknown {
		t.Fatalf("verdict = %q for an unsigned tag, want unknown — an app just marked its own object clean", v)
	}

	// And Status, the door's read, must agree. Two readers of one tag that
	// disagree about whether it is a verdict is the drift readTag exists to
	// prevent.
	st, _, err := c.Status(context.Background(), "b", "k")
	if err != nil {
		t.Fatal(err)
	}
	if st != VerdictUnknown {
		t.Fatalf("Status = %q for an unsigned tag, want unknown", st)
	}
}

// A signature that is present but not ours — a random value, or one lifted
// from a different object — is the same answer.
func TestForgedSignatureIsNotAVerdict(t *testing.T) {
	signer := testSigner(t, "deployment-key", "")
	// A signature legitimately issued for a DIFFERENT object, copied across.
	lifted := signer.Sign(scanFacts{
		bucket: "b", key: "other", version: "v1",
		verdict: string(VerdictClean), at: "2026-09-05T00:00:00Z",
	})

	for _, tc := range []struct{ name, sig string }{
		{"rubbish", "not-a-signature"},
		{"another object's", lifted},
		{"an empty value", ""},
	} {
		t.Run(tc.name, func(t *testing.T) {
			s3 := &stubScanS3{t: t}
			s3.getObject = func(*awss3.GetObjectInput) (*awss3.GetObjectOutput, error) {
				return objectBody("hello"), nil
			}
			s3.allowTagRead(
				TagVerdict, string(VerdictClean),
				TagVersion, "v1",
				TagAt, "2026-09-05T00:00:00Z",
				TagSignature, tc.sig,
			)
			c := &clamavScanner{
				s3: s3, engine: &fakeEngine{res: clamd.Result{Status: clamd.StatusClean}},
				maxBytes: 1 << 20, signer: signer,
			}
			v, err := c.Verdict(context.Background(), "b", "k", "v1")
			if err != nil {
				t.Fatal(err)
			}
			if v != VerdictUnknown {
				t.Fatalf("verdict = %q, want unknown", v)
			}
		})
	}
}

// The positive half: a verdict this scanner wrote is read back and trusted, so
// the whole mechanism does not simply refuse everything. Without this the two
// tests above are satisfied by a readTag that always returns unknown.
func TestAVerdictThisScannerWroteIsTrusted(t *testing.T) {
	signer := testSigner(t, "deployment-key", "")
	var written []s3types.Tag
	s3 := &stubScanS3{t: t}
	s3.getObject = func(*awss3.GetObjectInput) (*awss3.GetObjectOutput, error) {
		return objectBody("hello"), nil
	}
	s3.allowTagRead()
	s3.putTagging = func(in *awss3.PutObjectTaggingInput) (*awss3.PutObjectTaggingOutput, error) {
		written = in.Tagging.TagSet
		return &awss3.PutObjectTaggingOutput{}, nil
	}
	c := &clamavScanner{
		s3: s3, engine: &fakeEngine{res: clamd.Result{Status: clamd.StatusClean}},
		maxBytes: 1 << 20, signer: signer,
	}

	v, recorded, err := c.Scan(context.Background(), "b", "k", "v1", 5)
	if err != nil || v != VerdictClean || !recorded {
		t.Fatalf("Scan = %q recorded=%v err=%v", v, recorded, err)
	}
	if len(written) != 4 {
		t.Fatalf("wrote %d tags, want four — the verdict, the version, the time and the signature over them:\n%v", len(written), written)
	}

	// Read it back through the real path: what was written IS what the next
	// read sees, which is the round trip the whole mechanism depends on.
	s3.getTagging = func(*awss3.GetObjectTaggingInput) (*awss3.GetObjectTaggingOutput, error) {
		return &awss3.GetObjectTaggingOutput{TagSet: written}, nil
	}
	got, err := c.Verdict(context.Background(), "b", "k", "v1")
	if err != nil {
		t.Fatal(err)
	}
	if got != VerdictClean {
		t.Fatalf("reading back a verdict this scanner just wrote = %q, want clean", got)
	}
}

// ROTATION. A verdict signed under the previous key is still a verdict — the
// alternative is that rotating the key rescans every object in every store at
// once — and the sweep re-signs it under the current key so the old one can
// eventually be dropped.
//
// The re-sign must NOT move carlos-scan-at. That timestamp is when the bytes
// were judged, and a rotation judges nothing; moving it would silently make
// every verdict look freshly scanned and destroy the one field an auditor
// reads to ask "how old is this answer".
func TestRotationResignsWithoutRescanningOrMovingTheTimestamp(t *testing.T) {
	old := testSigner(t, "old-key", "")
	rotated := testSigner(t, "new-key", "old-key")

	const at = "2026-09-05T00:00:00Z"
	stale := old.Sign(scanFacts{
		bucket: "b", key: "k", version: "v1", verdict: string(VerdictClean), at: at,
	})

	var written []s3types.Tag
	s3 := &stubScanS3{t: t}
	s3.allowTagRead(
		TagVerdict, string(VerdictClean),
		TagVersion, "v1",
		TagAt, at,
		TagSignature, stale,
	)
	s3.putTagging = func(in *awss3.PutObjectTaggingInput) (*awss3.PutObjectTaggingOutput, error) {
		written = in.Tagging.TagSet
		return &awss3.PutObjectTaggingOutput{}, nil
	}
	// getObject is deliberately unset: reaching the engine at all fails the
	// test, which is how "without rescanning" is asserted rather than assumed.
	c := &clamavScanner{s3: s3, engine: nil, maxBytes: 1 << 20, signer: rotated}

	v, err := c.Verdict(context.Background(), "b", "k", "v1")
	if err != nil {
		t.Fatal(err)
	}
	if v != VerdictClean {
		t.Fatalf("verdict = %q under the previous key, want clean — a rotation must not invalidate stored verdicts", v)
	}
	if written == nil {
		t.Fatal("nothing was re-signed, so the previous key could never be dropped")
	}

	var gotAt, gotSig string
	for _, tg := range written {
		switch aws2s(tg.Key) {
		case TagAt:
			gotAt = aws2s(tg.Value)
		case TagSignature:
			gotSig = aws2s(tg.Value)
		}
	}
	if gotAt != at {
		t.Fatalf("carlos-scan-at moved to %q — a re-sign judges nothing and must not claim a fresh scan", gotAt)
	}
	if gotSig == stale {
		t.Fatal("the signature was rewritten unchanged, still under the old key")
	}
	if ok, wasStale := rotated.Verify(scanFacts{
		bucket: "b", key: "k", version: "v1", verdict: string(VerdictClean), at: at,
	}, gotSig); !ok || wasStale {
		t.Fatalf("the re-signed value does not verify under the current key (ok=%v stale=%v)", ok, wasStale)
	}
}

// A re-sign that cannot be written must not fail the sweep. The verdict is
// still valid under the previous key; the worst case is that it is re-signed
// on a later tick, and failing the page over it would stall every object
// behind this one.
func TestAFailedResignDoesNotFailTheSweep(t *testing.T) {
	old := testSigner(t, "old-key", "")
	rotated := testSigner(t, "new-key", "old-key")
	const at = "2026-09-05T00:00:00Z"

	s3 := &stubScanS3{t: t}
	s3.allowTagRead(
		TagVerdict, string(VerdictClean), TagVersion, "v1", TagAt, at,
		TagSignature, old.Sign(scanFacts{bucket: "b", key: "k", version: "v1", verdict: string(VerdictClean), at: at}),
	)
	s3.putTagging = func(*awss3.PutObjectTaggingInput) (*awss3.PutObjectTaggingOutput, error) {
		return nil, errors.New("the bucket had a bad moment")
	}
	c := &clamavScanner{s3: s3, engine: nil, maxBytes: 1 << 20, signer: rotated}

	v, err := c.Verdict(context.Background(), "b", "k", "v1")
	if err != nil {
		t.Fatalf("a failed re-sign failed the sweep: %v", err)
	}
	if v != VerdictClean {
		t.Fatalf("verdict = %q, want clean", v)
	}
}

// Four owned tags, not three, so the room left for a tenant's own is six.
// 63e80cc computed this from a hardcoded 3; a fourth tag silently made that
// wrong in the direction of writing an eleven-tag set S3 would reject.
func TestInfectedEvictionAccountsForTheSignatureTag(t *testing.T) {
	var foreign []string
	for i := 0; i < 8; i++ {
		foreign = append(foreign, "own-"+string(rune('a'+i)), "v")
	}
	var written []s3types.Tag
	s3 := &stubScanS3{t: t}
	s3.getObject = func(*awss3.GetObjectInput) (*awss3.GetObjectOutput, error) {
		return objectBody("hello"), nil
	}
	s3.allowTagRead(foreign...)
	s3.putTagging = func(in *awss3.PutObjectTaggingInput) (*awss3.PutObjectTaggingOutput, error) {
		written = in.Tagging.TagSet
		return &awss3.PutObjectTaggingOutput{}, nil
	}
	c := &clamavScanner{
		s3:       s3,
		engine:   &fakeEngine{res: clamd.Result{Status: clamd.StatusInfected, Signature: "Eicar-Test"}},
		maxBytes: 1 << 20,
		signer:   testSigner(t, "deployment-key", ""),
	}
	if _, _, err := c.Scan(context.Background(), "b", "k", "v1", 5); err != nil {
		t.Fatal(err)
	}
	if len(written) > maxObjectTags {
		t.Fatalf("wrote %d tags, over S3's limit of %d", len(written), maxObjectTags)
	}
	var kept int
	for _, tg := range written {
		if !isScanTag(aws2s(tg.Key)) {
			kept++
		}
	}
	if want := maxObjectTags - 4; kept != want {
		t.Fatalf("kept %d foreign tags, want %d — four owned tags now, not three:\n%v", kept, want, written)
	}
}

// The signature tag is the scanner's own, so a re-scan replaces it rather than
// treating it as a tenant's and preserving it. A stale signature left beside a
// fresh verdict is a verdict that will not verify.
func TestSignatureTagIsOwnedByTheScanner(t *testing.T) {
	if !isScanTag(TagSignature) {
		t.Fatal("the signature tag is not in isScanTag — a re-scan would preserve the old one beside a new verdict")
	}
}

// signedTagPairs is a tag set as the scanner itself would have written it:
// verdict, version, time, and a signature over all of it.
//
// Every test that asserts on how a verdict READS has to build one of these
// now. An unsigned set is not a weaker verdict, it is not a verdict — which is
// the entire point of the mechanism, and is why hand-written tag sets stopped
// working the moment signing landed.
func signedTagPairs(t *testing.T, sg *TagSigner, bucket, key, version string, v Verdict) []string {
	t.Helper()
	const at = "2026-09-05T00:00:00Z"
	return []string{
		TagVerdict, string(v),
		TagVersion, version,
		TagAt, at,
		TagSignature, sg.Sign(scanFacts{
			bucket: bucket, key: key, version: version, verdict: string(v), at: at,
		}),
	}
}