test: add unit tests for idempotent RewrapDEK and E2E tests for key rotation

Unit tests (crypto/envelope_test.go):
- TestRewrapDEK_Idempotent: rewrap already-rewrapped object returns "" (skip)
- TestRewrapDEK_BothKEKsFail: neither KEK can unwrap → error
- TestEncodeDecodeBase64: round-trip for new helper functions

E2E tests (test_encryption.py):
- test_key_rotation_multiple_artifacts: rotate with 3 artifacts, verify all
  decryptable with new key, old key rejected
- test_key_rotation_clears_rotation_state: verify rotation_started_at and
  rotation_encrypted_secret are NULL after successful rotation

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: GenerQAQ

src/server/api/go/internal/infra/crypto/envelope_test.go

@@ -195,6 +195,71 @@ func TestRewrapDEK(t *testing.T) {
 	}
 }
 
+func TestRewrapDEK_Idempotent(t *testing.T) {
+	// Simulate the crash-safe key rotation: if an object is already rewrapped
+	// with newKEK, calling RewrapDEK again should return "" (skip) instead of erroring.
+	oldUserKEK, _ := DeriveUserKEK("old-key-for-idempotent", "pepper")
+	newUserKEK, _ := DeriveUserKEK("new-key-for-idempotent", "pepper")
+
+	plaintext := []byte("data for idempotent rewrap test")
+	ciphertext, meta, _ := EncryptData(oldUserKEK, plaintext)
+
+	// First rewrap: old → new
+	newWrapped, err := RewrapDEK(meta, oldUserKEK, newUserKEK)
+	if err != nil {
+		t.Fatal("first rewrap failed:", err)
+	}
+	if newWrapped == "" {
+		t.Fatal("first rewrap should not skip")
+	}
+	meta.UserWrappedDEK = newWrapped
+
+	// Second rewrap with same old/new KEKs — should skip (return "")
+	skipped, err := RewrapDEK(meta, oldUserKEK, newUserKEK)
+	if err != nil {
+		t.Fatal("second rewrap should not error:", err)
+	}
+	if skipped != "" {
+		t.Fatal("second rewrap should return empty string (skip already-rewrapped object)")
+	}
+
+	// Verify data is still decryptable with new KEK
+	decrypted, err := DecryptData(newUserKEK, ciphertext, meta)
+	if err != nil {
+		t.Fatal("decrypt with new KEK after idempotent rewrap:", err)
+	}
+	if !bytes.Equal(plaintext, decrypted) {
+		t.Fatal("plaintext should match after idempotent rewrap")
+	}
+}
+
+func TestRewrapDEK_BothKEKsFail(t *testing.T) {
+	// If the DEK is wrapped with a totally different KEK, RewrapDEK should error.
+	kek1, _ := DeriveUserKEK("kek-one", "pepper")
+	kek2, _ := DeriveUserKEK("kek-two", "pepper")
+	kekWrong, _ := DeriveUserKEK("kek-wrong", "pepper")
+
+	_, meta, _ := EncryptData(kekWrong, []byte("data"))
+
+	// Neither kek1 nor kek2 can unwrap — should return error
+	_, err := RewrapDEK(meta, kek1, kek2)
+	if err == nil {
+		t.Fatal("RewrapDEK should fail when neither KEK can unwrap")
+	}
+}
+
+func TestEncodeDecodeBase64(t *testing.T) {
+	original := []byte("hello base64 round trip")
+	encoded := EncodeBase64(original)
+	decoded, err := DecodeBase64(encoded)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Equal(original, decoded) {
+		t.Fatal("decoded should match original")
+	}
+}
+
 func TestMetadataMapRoundTrip(t *testing.T) {
 	meta := &EncryptedMeta{
 		Algo:           "AES-256-GCM",

src/server/tests/e2e/test_encryption.py

@@ -506,6 +506,70 @@ async def test_key_rotation_encrypted_project(db_conn, encrypted_project):
         assert resp.content == plaintext
 
 
+@pytest.mark.asyncio
+async def test_key_rotation_multiple_artifacts(db_conn, encrypted_project):
+    """Rotate key with multiple artifacts → ALL artifacts still accessible with new key."""
+    files = {
+        "file1.txt": b"first file for rotation test",
+        "file2.txt": b"second file for rotation test",
+        "file3.txt": b"third file for rotation test",
+    }
+
+    async with httpx.AsyncClient() as client:
+        disk_id = await create_disk(client, encrypted_project.headers)
+
+        # Upload multiple artifacts
+        for fname, content in files.items():
+            await upload_artifact(
+                client, disk_id, encrypted_project.headers,
+                filename=fname, content=content,
+            )
+
+        # Rotate the key
+        resp = await rotate_key_with_rewrap(client, encrypted_project.bearer_token)
+        assert resp.status_code == 200, f"Key rotation failed: {resp.text}"
+        new_key = resp.json()["data"]["secret_key"]
+        new_headers = {"Authorization": f"Bearer {new_key}"}
+
+        # Verify ALL artifacts are decryptable with new key
+        for fname, expected_content in files.items():
+            resp = await download_artifact(
+                client, disk_id, f"/{fname}", new_headers,
+            )
+            assert resp.status_code == 200, f"Download {fname} failed after rotation"
+            assert resp.content == expected_content, f"Content mismatch for {fname}"
+
+        # Verify old key no longer works
+        resp = await download_artifact(
+            client, disk_id, "/file1.txt", encrypted_project.headers,
+        )
+        assert resp.status_code in (401, 403, 404, 500), (
+            f"Old key should be rejected after rotation, got {resp.status_code}"
+        )
+
+
+@pytest.mark.asyncio
+async def test_key_rotation_clears_rotation_state(db_conn, encrypted_project):
+    """After successful rotation, rotation_started_at should be NULL."""
+    async with httpx.AsyncClient() as client:
+        disk_id = await create_disk(client, encrypted_project.headers)
+        await upload_artifact(
+            client, disk_id, encrypted_project.headers,
+            filename="state-check.txt", content=b"check rotation state",
+        )
+
+        resp = await rotate_key_with_rewrap(client, encrypted_project.bearer_token)
+        assert resp.status_code == 200
+
+        # Verify rotation state is cleared in DB
+        row = await db_conn.fetchrow(
+            "SELECT rotation_started_at, rotation_encrypted_secret FROM projects WHERE id = $1",
+            encrypted_project.project_id,
+        )
+        assert row["rotation_started_at"] is None, "rotation_started_at should be NULL after success"
+        assert row["rotation_encrypted_secret"] is None, "rotation_encrypted_secret should be NULL after success"
+
+
 @pytest.mark.asyncio
 async def test_key_rotation_plain_project(db_conn, plain_project):
     """Rotate key on plain project → works without re-wrap (no encrypted data)."""