from __future__ import annotations import hashlib import importlib.util import io import json import tempfile import threading import unittest from contextlib import nullcontext from pathlib import Path from types import SimpleNamespace from unittest.mock import AsyncMock, patch from PIL import Image from starlette.datastructures import Headers, UploadFile from . import upload_storage as upload_storage_module from .upload_storage import ( UploadWriteFreezeGate, UploadWriteFrozen, assert_path_without_reparse, build_privacy_safe_manifest, database_target_sha256, inspect_public_avatar_image, is_runtime_generated_public_avatar_relative_path, normalize_public_avatar_relative_path, public_avatar_relative_path_sha256, validate_current_avatar_references, validate_upload_manifest, ) DATABASE_TARGET_SHA256 = hashlib.sha256(b"connected-database-target").hexdigest() def _valid_png_bytes() -> bytes: buffer = io.BytesIO() Image.new("RGBA", (1, 1), (0, 0, 0, 0)).save(buffer, format="PNG") return buffer.getvalue() class _AsyncValueContext: def __init__(self, value): self.value = value async def __aenter__(self): return self.value async def __aexit__(self, exc_type, exc, traceback): return False class _RuntimeDatabaseConnection: def __init__(self, *, avatar_urls: list[str], database_name: str = "vignette"): self.avatar_urls = avatar_urls self.database_name = database_name self.transaction_options: list[dict[str, object]] = [] self.executed: list[str] = [] self.codec_names: list[str] = [] def transaction(self, **options): self.transaction_options.append(options) return _AsyncValueContext(self) async def fetchrow(self, _query: str): return { "database_name": self.database_name, "database_role": "vignette_app", "server_address": "10.0.0.5", "server_port": 5432, } async def execute(self, query: str): self.executed.append(query) async def set_type_codec(self, name: str, **_options): self.codec_names.append(name) async def fetch(self, _query: str): return [{"avatar_url": value} for value in self.avatar_urls] class _RuntimeDatabasePool: def __init__(self, connection: _RuntimeDatabaseConnection): self.connection = connection self.acquire_count = 0 def acquire(self): self.acquire_count += 1 return _AsyncValueContext(self.connection) class UploadStorageContractTest(unittest.TestCase): def test_broken_png_crc_is_decode_invalid_not_an_initializer_error(self) -> None: content = bytearray(_valid_png_bytes()) idat = content.index(b"IDAT") + 4 content[idat] ^= 0x01 proof = inspect_public_avatar_image( "profile-avatars/broken.png", bytes(content), ) self.assertFalse(proof.valid) def test_api_and_release_probe_share_connected_database_identity_digest(self) -> None: helper_path = ( Path(__file__).resolve().parents[3] / "scripts" / "public_runtime_database_identity.py" ) spec = importlib.util.spec_from_file_location( "public_runtime_database_identity_contract_test", helper_path, ) self.assertIsNotNone(spec) self.assertIsNotNone(spec.loader if spec is not None else None) module = importlib.util.module_from_spec(spec) assert spec is not None and spec.loader is not None spec.loader.exec_module(module) release_database_target_sha256 = module.database_target_sha256 identity = { "database_name": "vignette", "database_role": "vignette_app", "server_address": "10.0.0.5", "server_port": 5432, } self.assertEqual( release_database_target_sha256(**identity), database_target_sha256(**identity), ) def test_manifest_proves_required_object_without_recording_its_path(self) -> None: with tempfile.TemporaryDirectory(prefix="vignette-upload-manifest-") as raw: root = Path(raw) / "uploads" object_path = root / "profile-avatars" / "private-object.jpg" object_path.parent.mkdir(parents=True) object_path.write_bytes(b"avatar-bytes") token_sha = hashlib.sha256(b"freeze-token").hexdigest() payload = build_privacy_safe_manifest( upload_root=root, references=[("profile-avatars/private-object.jpg", "profile_avatar")], preserved_paths=["profile-avatars/private-object.jpg"], database_target_sha256=DATABASE_TARGET_SHA256, write_freeze_token_sha256=token_sha, ) serialized = json.dumps(payload, sort_keys=True) self.assertNotIn("private-object", serialized) self.assertNotIn('"relative_path":', serialized) manifest_dir = Path(raw) / "private-state" / "manifests" manifest_dir.mkdir(parents=True) manifest = manifest_dir / "manifest.json" manifest.write_text(serialized, encoding="utf-8") expected_sha = hashlib.sha256(manifest.read_bytes()).hexdigest() proof = validate_upload_manifest( upload_root=root, manifest_path=manifest, expected_manifest_sha256=expected_sha, expected_write_freeze_path=Path(raw) / "upload-write.freeze", expected_database_target_sha256=DATABASE_TARGET_SHA256, require_freeze_path_binding=False, ) self.assertEqual(1, proof.required_object_count) public_manifest = root / "manifest.json" public_manifest.write_text(serialized, encoding="utf-8") public_manifest_sha = hashlib.sha256( public_manifest.read_bytes() ).hexdigest() with self.assertRaisesRegex(ValueError, "outside the public upload root"): validate_upload_manifest( upload_root=root, manifest_path=public_manifest, expected_manifest_sha256=public_manifest_sha, expected_write_freeze_path=Path(raw) / "upload-write.freeze", expected_database_target_sha256=DATABASE_TARGET_SHA256, require_freeze_path_binding=False, ) object_path.write_bytes(b"drift") with self.assertRaisesRegex(ValueError, "content hash"): validate_upload_manifest( upload_root=root, manifest_path=manifest, expected_manifest_sha256=expected_sha, expected_write_freeze_path=Path(raw) / "upload-write.freeze", expected_database_target_sha256=DATABASE_TARGET_SHA256, require_freeze_path_binding=False, ) def test_duplicate_db_references_are_aggregated_without_raw_paths(self) -> None: with tempfile.TemporaryDirectory(prefix="vignette-upload-duplicates-") as raw: root = Path(raw) / "uploads" object_path = root / "profile-avatars" / "shared-avatar.png" object_path.parent.mkdir(parents=True) object_path.write_bytes(b"shared-avatar") token_sha = hashlib.sha256(b"freeze-token").hexdigest() payload = build_privacy_safe_manifest( upload_root=root, references=[ ("profile-avatars/shared-avatar.png", "profile_avatar"), ("profile-avatars/shared-avatar.png", "profile_avatar"), ], preserved_paths=["profile-avatars/shared-avatar.png"], database_target_sha256=DATABASE_TARGET_SHA256, write_freeze_token_sha256=token_sha, ) self.assertEqual(1, payload["required_object_count"]) self.assertEqual(2, payload["required_reference_count"]) self.assertEqual( 2, payload["preserved_objects"][0]["reference_count"] ) serialized = json.dumps(payload, sort_keys=True) self.assertNotIn("shared-avatar", serialized) manifest_dir = Path(raw) / "private-state" manifest_dir.mkdir() manifest = manifest_dir / "manifest.json" manifest.write_text(serialized, encoding="utf-8") proof = validate_upload_manifest( upload_root=root, manifest_path=manifest, expected_manifest_sha256=hashlib.sha256( manifest.read_bytes() ).hexdigest(), expected_write_freeze_path=Path(raw) / "write-freeze.json", expected_database_target_sha256=DATABASE_TARGET_SHA256, require_freeze_path_binding=False, ) current = validate_current_avatar_references( upload_root=root, avatar_urls=[ "/uploads/profile-avatars/shared-avatar.png", "/uploads/profile-avatars/shared-avatar.png", ], initial_manifest=proof, expected_database_target_sha256=DATABASE_TARGET_SHA256, ) self.assertEqual(1, current.object_count) self.assertEqual(2, current.reference_count) self.assertEqual(proof.reference_set_sha256, current.reference_set_sha256) def test_runtime_generated_current_references_redecode_and_count_corruption( self, ) -> None: with tempfile.TemporaryDirectory(prefix="vignette-upload-current-decode-") as raw: root = Path(raw) / "uploads" avatar_root = root / "profile-avatars" avatar_root.mkdir(parents=True) preserved = avatar_root / "preserved.png" preserved.write_bytes(_valid_png_bytes()) payload = build_privacy_safe_manifest( upload_root=root, references=[], preserved_paths=["profile-avatars/preserved.png"], database_target_sha256=DATABASE_TARGET_SHA256, write_freeze_token_sha256=hashlib.sha256(b"freeze-token").hexdigest(), ) private = Path(raw) / "private" private.mkdir() manifest = private / "manifest.json" manifest.write_text(json.dumps(payload), encoding="utf-8") proof = validate_upload_manifest( upload_root=root, manifest_path=manifest, expected_manifest_sha256=hashlib.sha256( manifest.read_bytes() ).hexdigest(), expected_write_freeze_path=private / "write-freeze.json", expected_database_target_sha256=DATABASE_TARGET_SHA256, require_freeze_path_binding=False, ) valid_name = ( "8da1b8fc-3bdd-4e72-86f7-d17fc315c4cb-" "AbCdEf_12345.png" ) corrupt_name = ( "1265087b-d3d0-4e85-82bc-499d993a3f89-" "QrStUv_67890.png" ) (avatar_root / valid_name).write_bytes(_valid_png_bytes()) (avatar_root / corrupt_name).write_bytes(b"not-a-png") current = validate_current_avatar_references( upload_root=root, avatar_urls=[ f"/uploads/profile-avatars/{valid_name}", f"/uploads/profile-avatars/{corrupt_name}", f"/uploads/profile-avatars/{corrupt_name}", ], initial_manifest=proof, expected_database_target_sha256=DATABASE_TARGET_SHA256, ) self.assertEqual(2, current.object_count) self.assertEqual(3, current.reference_count) self.assertEqual(1, current.decode_invalid_object_count) self.assertEqual(2, current.decode_invalid_reference_count) arbitrary = avatar_root / "unbound.png" arbitrary.write_bytes(_valid_png_bytes()) with self.assertRaisesRegex(ValueError, "not authorized"): validate_current_avatar_references( upload_root=root, avatar_urls=["/uploads/profile-avatars/unbound.png"], initial_manifest=proof, expected_database_target_sha256=DATABASE_TARGET_SHA256, ) def test_manifest_v3_preserves_zero_ref_upload_and_binds_connected_db(self) -> None: with tempfile.TemporaryDirectory(prefix="vignette-upload-preserved-v2-") as raw: root = Path(raw) / "uploads" avatar_root = root / "profile-avatars" avatar_root.mkdir(parents=True) referenced = avatar_root / "referenced.png" pending = avatar_root / "pending.webp" referenced.write_bytes(b"referenced") pending.write_bytes(b"pending") payload = build_privacy_safe_manifest( upload_root=root, references=[("profile-avatars/referenced.png", "profile_avatar")], preserved_paths=[ "profile-avatars/referenced.png", "profile-avatars/pending.webp", ], database_target_sha256=DATABASE_TARGET_SHA256, write_freeze_token_sha256=hashlib.sha256(b"freeze").hexdigest(), ) self.assertEqual( "vignette.public-avatar-upload-manifest.v3", payload["schema_version"], ) self.assertEqual(2, payload["preserved_object_count"]) self.assertEqual(0, payload["preserved_decode_valid_count"]) self.assertEqual(2, payload["preserved_decode_invalid_count"]) self.assertEqual(1, payload["required_decode_invalid_object_count"]) self.assertEqual( 1, payload["required_decode_invalid_reference_count"], ) self.assertEqual( len(b"referenced") + len(b"pending"), payload["preserved_total_size_bytes"], ) self.assertEqual(1, payload["required_object_count"]) self.assertEqual( [0, 1], sorted( int(record["reference_count"]) for record in payload["preserved_objects"] ), ) self.assertEqual( 1, payload["database_reference_contract"]["reference_count"], ) serialized = json.dumps(payload, sort_keys=True) self.assertNotIn("referenced.png", serialized) self.assertNotIn("pending.webp", serialized) private = Path(raw) / "private" private.mkdir() manifest = private / "manifest.json" manifest.write_text(serialized, encoding="utf-8") manifest_sha = hashlib.sha256(manifest.read_bytes()).hexdigest() proof = validate_upload_manifest( upload_root=root, manifest_path=manifest, expected_manifest_sha256=manifest_sha, expected_write_freeze_path=private / "freeze.json", expected_database_target_sha256=DATABASE_TARGET_SHA256, require_freeze_path_binding=False, ) self.assertEqual(2, proof.preserved_object_count) self.assertEqual(0, len(proof.preserved_records)) self.assertEqual(2, proof.preserved_decode_invalid_count) self.assertEqual( len(b"referenced") + len(b"pending"), proof.preserved_total_size_bytes, ) self.assertEqual(DATABASE_TARGET_SHA256, proof.database_target_sha256) with self.assertRaisesRegex(ValueError, "database target binding"): validate_upload_manifest( upload_root=root, manifest_path=manifest, expected_manifest_sha256=manifest_sha, expected_write_freeze_path=private / "freeze.json", expected_database_target_sha256=hashlib.sha256( b"other-database" ).hexdigest(), require_freeze_path_binding=False, ) (avatar_root / "unknown.txt").write_bytes(b"must-not-be-served") with self.assertRaisesRegex(ValueError, "invalid extra"): validate_upload_manifest( upload_root=root, manifest_path=manifest, expected_manifest_sha256=manifest_sha, expected_write_freeze_path=private / "freeze.json", expected_database_target_sha256=DATABASE_TARGET_SHA256, require_freeze_path_binding=False, ) def test_manifest_hash_and_json_parse_share_one_byte_read(self) -> None: source = Path(__file__).with_name("upload_storage.py").read_text( encoding="utf-8" ) validator = source[ source.index("def validate_upload_manifest(") : source.index("class CurrentAvatarReferenceProof") ] self.assertIn("manifest_bytes = manifest.read_bytes()", validator) self.assertIn("_sha256_bytes(manifest_bytes)", validator) self.assertIn('json.loads(manifest_bytes.decode("utf-8"))', validator) self.assertNotIn("manifest.read_text", validator) def test_routine_manifest_validation_rehashes_preserved_and_allows_only_generated_extras( self, ) -> None: with tempfile.TemporaryDirectory( prefix="vignette-upload-routine-validation-" ) as raw: root = Path(raw) / "uploads" avatar_root = root / "profile-avatars" avatar_root.mkdir(parents=True) preserved = avatar_root / "legacy-avatar.png" preserved.write_bytes(_valid_png_bytes()) payload = build_privacy_safe_manifest( upload_root=root, references=[("profile-avatars/legacy-avatar.png", "profile_avatar")], preserved_paths=["profile-avatars/legacy-avatar.png"], database_target_sha256=DATABASE_TARGET_SHA256, write_freeze_token_sha256=hashlib.sha256(b"freeze").hexdigest(), ) private = Path(raw) / "private" private.mkdir() manifest = private / "manifest.json" manifest.write_text(json.dumps(payload), encoding="utf-8") manifest_sha256 = hashlib.sha256(manifest.read_bytes()).hexdigest() generated_name = ( "8da1b8fc-3bdd-4e72-86f7-d17fc315c4cb-" "AbCdEf_12345.webp" ) (avatar_root / generated_name).write_bytes(b"new-runtime-upload") proof = validate_upload_manifest( upload_root=root, manifest_path=manifest, expected_manifest_sha256=manifest_sha256, expected_write_freeze_path=private / "freeze.json", expected_database_target_sha256=DATABASE_TARGET_SHA256, require_freeze_path_binding=False, verify_preserved_objects=True, reject_unbound_extras=False, ) self.assertEqual(1, len(proof.preserved_records)) self.assertTrue( is_runtime_generated_public_avatar_relative_path( f"profile-avatars/{generated_name}" ) ) with self.assertRaisesRegex(ValueError, "unbound extra"): validate_upload_manifest( upload_root=root, manifest_path=manifest, expected_manifest_sha256=manifest_sha256, expected_write_freeze_path=private / "freeze.json", expected_database_target_sha256=DATABASE_TARGET_SHA256, require_freeze_path_binding=False, verify_preserved_objects=True, reject_unbound_extras=True, ) arbitrary = avatar_root / "looks-valid-but-unbound.png" arbitrary.write_bytes(b"unauthorized") with self.assertRaisesRegex(ValueError, "unauthorized extra"): validate_upload_manifest( upload_root=root, manifest_path=manifest, expected_manifest_sha256=manifest_sha256, expected_write_freeze_path=private / "freeze.json", expected_database_target_sha256=DATABASE_TARGET_SHA256, require_freeze_path_binding=False, verify_preserved_objects=True, reject_unbound_extras=False, ) arbitrary.unlink() preserved.write_bytes(b"tampered") with self.assertRaisesRegex(ValueError, "content hash"): validate_upload_manifest( upload_root=root, manifest_path=manifest, expected_manifest_sha256=manifest_sha256, expected_write_freeze_path=private / "freeze.json", expected_database_target_sha256=DATABASE_TARGET_SHA256, require_freeze_path_binding=False, verify_preserved_objects=True, reject_unbound_extras=False, ) def test_preserved_immutable_cache_has_a_fail_closed_total_size_limit(self) -> None: from . import upload_storage with tempfile.TemporaryDirectory(prefix="vignette-avatar-cache-cap-") as raw: root = Path(raw) / "uploads" avatar = root / "profile-avatars" / "legacy.png" avatar.parent.mkdir(parents=True) avatar.write_bytes(_valid_png_bytes()) payload = build_privacy_safe_manifest( upload_root=root, references=[("profile-avatars/legacy.png", "profile_avatar")], preserved_paths=["profile-avatars/legacy.png"], database_target_sha256=DATABASE_TARGET_SHA256, write_freeze_token_sha256=hashlib.sha256(b"freeze").hexdigest(), ) private = Path(raw) / "private" private.mkdir() manifest = private / "manifest.json" manifest.write_text(json.dumps(payload), encoding="utf-8") with ( patch.object(upload_storage, "MAX_PRESERVED_AVATAR_CACHE_BYTES", 1), self.assertRaisesRegex(ValueError, "cache size limit"), ): validate_upload_manifest( upload_root=root, manifest_path=manifest, expected_manifest_sha256=hashlib.sha256( manifest.read_bytes() ).hexdigest(), expected_write_freeze_path=private / "freeze.json", expected_database_target_sha256=DATABASE_TARGET_SHA256, require_freeze_path_binding=False, ) def test_public_avatar_normalizer_rejects_encoded_control_and_non_image_paths( self, ) -> None: for value in ( "profile-avatars/encoded%2fpath.png", "profile-avatars/control\x01.png", "profile-avatars/not-an-image.txt", "profile-avatars/nested/file.png", "profile-avatars/.hidden.png", ): with self.subTest(value=value), self.assertRaises(ValueError): normalize_public_avatar_relative_path(value) def test_current_inventory_cannot_be_empty_after_nonempty_initialization( self, ) -> None: with tempfile.TemporaryDirectory( prefix="vignette-upload-empty-current-" ) as raw: root = Path(raw) / "uploads" object_path = root / "profile-avatars" / "initial.png" object_path.parent.mkdir(parents=True) object_path.write_bytes(b"initial") payload = build_privacy_safe_manifest( upload_root=root, references=[("profile-avatars/initial.png", "profile_avatar")], preserved_paths=["profile-avatars/initial.png"], database_target_sha256=DATABASE_TARGET_SHA256, write_freeze_token_sha256=hashlib.sha256(b"token").hexdigest(), ) private = Path(raw) / "private" private.mkdir() manifest = private / "manifest.json" manifest.write_text(json.dumps(payload), encoding="utf-8") proof = validate_upload_manifest( upload_root=root, manifest_path=manifest, expected_manifest_sha256=hashlib.sha256( manifest.read_bytes() ).hexdigest(), expected_write_freeze_path=Path(raw) / "freeze.json", expected_database_target_sha256=DATABASE_TARGET_SHA256, require_freeze_path_binding=False, ) with self.assertRaisesRegex(ValueError, "explicit reset receipt"): validate_current_avatar_references( upload_root=root, avatar_urls=[], initial_manifest=proof, expected_database_target_sha256=DATABASE_TARGET_SHA256, ) def test_manifest_and_freeze_state_are_rejected_inside_public_root(self) -> None: with tempfile.TemporaryDirectory( prefix="vignette-upload-private-state-" ) as raw: root = Path(raw) / "uploads" object_path = root / "profile-avatars" / "avatar.png" object_path.parent.mkdir(parents=True) object_path.write_bytes(b"avatar") token_sha = hashlib.sha256(b"token").hexdigest() with self.assertRaisesRegex(ValueError, "outside the public upload root"): build_privacy_safe_manifest( upload_root=root, references=[("profile-avatars/avatar.png", "profile_avatar")], preserved_paths=["profile-avatars/avatar.png"], database_target_sha256=DATABASE_TARGET_SHA256, write_freeze_token_sha256=token_sha, write_freeze_path=root / ".private-freeze.json", ) def test_static_mount_exposes_only_profile_avatar_directory(self) -> None: source = Path(__file__).with_name("main.py").read_text(encoding="utf-8") runtime_source = Path(__file__).with_name("upload_runtime.py").read_text( encoding="utf-8" ) self.assertIn('app.mount(\n "/uploads/profile-avatars"', source) self.assertIn("FlatPublicAvatarStaticFiles(directory=", source) self.assertNotIn("\n StaticFiles(directory=", source) self.assertNotIn('app.mount("/uploads",', source) self.assertNotIn('name="uploads"', source) self.assertIn( "assert_path_without_reparse(\n public_avatar_root,", runtime_source, ) def test_public_avatar_root_reparse_is_rejected(self) -> None: with tempfile.TemporaryDirectory(prefix="vignette-avatar-reparse-") as raw: root = Path(raw) / "uploads" avatar_root = root / "profile-avatars" avatar_root.mkdir(parents=True) with patch.object( upload_storage_module, "_is_reparse_or_symlink", side_effect=lambda candidate: candidate == avatar_root, ): with self.assertRaisesRegex(ValueError, "reparse point"): assert_path_without_reparse( avatar_root, "initialized public avatar directory", ) def test_write_gate_blocks_new_writes_and_counts_inflight_leases(self) -> None: with tempfile.TemporaryDirectory(prefix="vignette-upload-freeze-") as raw: freeze = Path(raw) / "upload-write.freeze" gate = UploadWriteFreezeGate(freeze) with gate.write_lease(): self.assertEqual(1, gate.status().in_flight) self.assertEqual(0, gate.status().in_flight) freeze.write_text( json.dumps( { "schema_version": "vignette.public-upload-write-freeze.v1", "token": "opaque-token-0123456789abcdef0123456789", } ), encoding="utf-8", ) with self.assertRaises(UploadWriteFrozen): with gate.write_lease(): pass status = gate.status() self.assertTrue(status.active) self.assertTrue(status.valid) self.assertEqual(0, status.in_flight) def test_write_gate_status_reads_sentinel_and_counter_under_one_lock(self) -> None: with tempfile.TemporaryDirectory(prefix="vignette-upload-freeze-race-") as raw: freeze = Path(raw) / "upload-write.freeze" gate = UploadWriteFreezeGate(freeze) status_read_started = threading.Event() allow_status_read = threading.Event() lease_entered = threading.Event() statuses = [] def blocking_entry_probe(candidate: Path) -> bool: if candidate == freeze: status_read_started.set() if not allow_status_read.wait(2): raise AssertionError("status sentinel read did not resume") return False return candidate.exists() def read_status() -> None: statuses.append(gate.status()) def enter_lease() -> None: with gate.write_lease(): lease_entered.set() with patch.object( upload_storage_module, "_path_entry_exists_without_following", side_effect=blocking_entry_probe, ): status_thread = threading.Thread(target=read_status) status_thread.start() self.assertTrue(status_read_started.wait(1)) lease_thread = threading.Thread(target=enter_lease) lease_thread.start() self.assertFalse(lease_entered.wait(0.1)) allow_status_read.set() status_thread.join(2) lease_thread.join(2) self.assertFalse(status_thread.is_alive()) self.assertFalse(lease_thread.is_alive()) self.assertEqual(1, len(statuses)) self.assertEqual(0, statuses[0].in_flight) self.assertTrue(lease_entered.is_set()) def test_unreadable_or_dangling_freeze_entry_blocks_writes(self) -> None: with tempfile.TemporaryDirectory(prefix="vignette-upload-freeze-entry-") as raw: freeze = Path(raw) / "dangling-freeze" gate = UploadWriteFreezeGate(freeze) with patch.object( upload_storage_module, "_path_entry_exists_without_following", return_value=True, ): with self.assertRaises(UploadWriteFrozen): with gate.write_lease(): pass status = gate.status() self.assertTrue(status.active) self.assertFalse(status.valid) class RuntimeUploadDatabaseContractTest(unittest.IsolatedAsyncioTestCase): @staticmethod def _manifest_proof(raw: str): root = Path(raw) / "uploads" avatar = root / "profile-avatars" / "initial.png" avatar.parent.mkdir(parents=True) avatar.write_bytes(b"initial") target_sha256 = database_target_sha256( database_name="vignette", database_role="vignette_app", server_address="10.0.0.5", server_port=5432, ) payload = build_privacy_safe_manifest( upload_root=root, references=[("profile-avatars/initial.png", "profile_avatar")], preserved_paths=["profile-avatars/initial.png"], database_target_sha256=target_sha256, write_freeze_token_sha256=hashlib.sha256(b"freeze").hexdigest(), ) private = Path(raw) / "private" private.mkdir() manifest = private / "manifest.json" manifest.write_text(json.dumps(payload), encoding="utf-8") proof = validate_upload_manifest( upload_root=root, manifest_path=manifest, expected_manifest_sha256=hashlib.sha256(manifest.read_bytes()).hexdigest(), expected_write_freeze_path=private / "freeze.json", expected_database_target_sha256=target_sha256, require_freeze_path_binding=False, ) return root, proof, target_sha256 async def test_lifespan_database_proof_uses_one_repeatable_read_pool_snapshot( self, ) -> None: from . import upload_runtime with tempfile.TemporaryDirectory(prefix="vignette-runtime-db-proof-") as raw: root, proof, target_sha256 = self._manifest_proof(raw) connection = _RuntimeDatabaseConnection( avatar_urls=["/uploads/profile-avatars/initial.png"] ) pool = _RuntimeDatabasePool(connection) with ( patch.object(upload_runtime, "upload_root", root), patch.object(upload_runtime, "upload_manifest_proof", proof), patch.object( upload_runtime.settings, "public_runtime_db_target_sha256", target_sha256, ), ): runtime_proof = ( await upload_runtime.validate_runtime_upload_database_state(pool) ) self.assertEqual(1, pool.acquire_count) self.assertEqual( [{"isolation": "repeatable_read", "readonly": True}], connection.transaction_options, ) self.assertEqual(target_sha256, runtime_proof.database_target_sha256) self.assertEqual(1, runtime_proof.current_references.object_count) self.assertEqual(1, runtime_proof.current_references.reference_count) self.assertTrue(connection.executed) main_source = Path(__file__).with_name("main.py").read_text(encoding="utf-8") init_index = main_source.index("await init_pool()") proof_index = main_source.index("await validate_runtime_upload_database_state") first_mutation_index = main_source.index("await ensure_runtime_tables()") self.assertLess(init_index, proof_index) self.assertLess(proof_index, first_mutation_index) async def test_lifespan_database_proof_rejects_connected_target_drift(self) -> None: from . import upload_runtime with tempfile.TemporaryDirectory(prefix="vignette-runtime-db-drift-") as raw: root, proof, target_sha256 = self._manifest_proof(raw) connection = _RuntimeDatabaseConnection( avatar_urls=["/uploads/profile-avatars/initial.png"], database_name="other_database", ) pool = _RuntimeDatabasePool(connection) with ( patch.object(upload_runtime, "upload_root", root), patch.object(upload_runtime, "upload_manifest_proof", proof), patch.object( upload_runtime.settings, "public_runtime_db_target_sha256", target_sha256, ), ): with self.assertRaisesRegex(RuntimeError, "database target drift"): await upload_runtime.validate_runtime_upload_database_state(pool) async def test_every_new_pool_connection_is_bound_before_codec_setup(self) -> None: from . import db target_sha256 = database_target_sha256( database_name="vignette", database_role="vignette_app", server_address="10.0.0.5", server_port=5432, ) connection = _RuntimeDatabaseConnection(avatar_urls=[]) with ( patch.object(db.settings, "user_upload_manifest_required", True), patch.object( db.settings, "public_runtime_db_target_sha256", target_sha256, ), ): await db._init_connection(connection) self.assertEqual(["jsonb", "json"], connection.codec_names) drifted = _RuntimeDatabaseConnection( avatar_urls=[], database_name="other_database" ) with ( patch.object(db.settings, "user_upload_manifest_required", True), patch.object( db.settings, "public_runtime_db_target_sha256", target_sha256, ), ): with self.assertRaisesRegex(RuntimeError, "database target drift"): await db._init_connection(drifted) self.assertEqual([], drifted.codec_names) class AvatarWriteSafetyTest(unittest.IsolatedAsyncioTestCase): @staticmethod def _profile(users, *, avatar_url: str): return users.UserProfileResponse( user_id="user-1", email="user@example.test", display_name="사용자", role="learner", cohort_ids=[], affiliation="한신대학교", avatar_url=avatar_url, ) async def test_static_avatar_handler_rejects_nested_hidden_and_unknown_files( self, ) -> None: from starlette.exceptions import HTTPException as StarletteHTTPException from . import upload_runtime with tempfile.TemporaryDirectory(prefix="vignette-avatar-static-") as raw: upload_root = Path(raw) / "uploads" root = upload_root / "profile-avatars" root.mkdir(parents=True) generated_name = ( "8da1b8fc-3bdd-4e72-86f7-d17fc315c4cb-" "AbCdEf_12345.png" ) valid = root / generated_name valid_bytes = _valid_png_bytes() valid.write_bytes(valid_bytes) arbitrary = root / "avatar.png" arbitrary.write_bytes(b"unbound") nested = root / "nested" nested.mkdir() (nested / "avatar.png").write_bytes(b"nested") (root / ".hidden.png").write_bytes(b"hidden") (root / "notes.txt").write_bytes(b"private") files = upload_runtime.FlatPublicAvatarStaticFiles(directory=str(root)) scope = {"method": "GET", "headers": []} with patch.object(upload_runtime, "upload_manifest_proof", None): response = await files.get_response(generated_name, scope) self.assertEqual(200, response.status_code) self.assertEqual(valid_bytes, response.body) self.assertEqual( f'"{hashlib.sha256(valid_bytes).hexdigest()}"', response.headers["etag"], ) with patch.object(upload_runtime, "upload_manifest_proof", None): head = await files.get_response( generated_name, {"method": "HEAD", "headers": []}, ) self.assertEqual(b"", head.body) self.assertEqual(str(len(valid_bytes)), head.headers["content-length"]) with ( patch.object(upload_runtime, "upload_root", upload_root), patch.object(upload_runtime, "upload_manifest_proof", None), ): upload_runtime.assert_public_avatar_url_exists( f"/uploads/profile-avatars/{generated_name}" ) with self.assertRaisesRegex(ValueError, "not authorized"): upload_runtime.assert_public_avatar_url_exists( "/uploads/profile-avatars/avatar.png" ) valid.write_bytes(b"damaged-after-upload") with patch.object(upload_runtime, "upload_manifest_proof", None): with self.assertRaises(StarletteHTTPException) as damaged: await files.get_response(generated_name, scope) self.assertEqual(404, damaged.exception.status_code) corrupt_name = ( "1265087b-d3d0-4e85-82bc-499d993a3f89-" "QrStUv_67890.png" ) (root / corrupt_name).write_bytes(b"not-a-png") with patch.object(upload_runtime, "upload_manifest_proof", None): with self.assertRaises(StarletteHTTPException) as corrupt: await files.get_response(corrupt_name, scope) self.assertEqual(404, corrupt.exception.status_code) mismatch_name = ( "02f01847-1d0c-48c8-9559-4ba791d00275-" "MnOpQr_24680.png" ) mismatch_buffer = io.BytesIO() Image.new("RGB", (1, 1), (255, 255, 255)).save( mismatch_buffer, format="JPEG", ) (root / mismatch_name).write_bytes(mismatch_buffer.getvalue()) with patch.object(upload_runtime, "upload_manifest_proof", None): with self.assertRaises(StarletteHTTPException) as mismatch: await files.get_response(mismatch_name, scope) self.assertEqual(404, mismatch.exception.status_code) for invalid in ( "avatar.png", "nested/avatar.png", ".hidden.png", "notes.txt", ): with self.subTest(path=invalid): with patch.object(upload_runtime, "upload_manifest_proof", None): with self.assertRaises(StarletteHTTPException) as raised: await files.get_response(invalid, scope) self.assertEqual(404, raised.exception.status_code) legacy_name = "legacy-avatar.png" legacy = root / legacy_name legacy.write_bytes(b"legacy") path_sha256 = public_avatar_relative_path_sha256( f"profile-avatars/{legacy_name}" ) proof = SimpleNamespace( preserved_records=( ( path_sha256, hashlib.sha256(b"legacy").hexdigest(), len(b"legacy"), "image/png", b"legacy", ), ), preserved_path_decode_records=((path_sha256, True),), ) with patch.object(upload_runtime, "upload_manifest_proof", proof): response = await files.get_response(legacy_name, scope) self.assertEqual(200, response.status_code) self.assertEqual(b"legacy", response.body) self.assertEqual("6", response.headers["content-length"]) self.assertEqual("image/png", response.headers["content-type"]) self.assertEqual( f'"{hashlib.sha256(b"legacy").hexdigest()}"', response.headers["etag"], ) legacy.write_bytes(b"drift") cached = await files.get_response(legacy_name, scope) self.assertEqual(b"legacy", cached.body) head = await files.get_response( legacy_name, {"method": "HEAD", "headers": []}, ) self.assertEqual(b"", head.body) self.assertEqual("6", head.headers["content-length"]) forensic_name = "forensic-avatar.png" forensic_path_sha256 = public_avatar_relative_path_sha256( f"profile-avatars/{forensic_name}" ) (root / forensic_name).write_bytes(b"\x89PNG\r\n\x1a\ncorrupt") forensic_proof = SimpleNamespace( preserved_records=(), preserved_path_decode_records=( (forensic_path_sha256, False), ), ) with patch.object( upload_runtime, "upload_manifest_proof", forensic_proof, ): with self.assertRaises(StarletteHTTPException) as raised: await files.get_response(forensic_name, scope) self.assertEqual(404, raised.exception.status_code) async def test_upload_root_is_revalidated_immediately_before_write(self) -> None: from fastapi import HTTPException from .routes import users with tempfile.TemporaryDirectory(prefix="vignette-avatar-root-drift-") as raw: root = Path(raw) with ( patch.object(users, "public_avatar_root", root), patch.object( users, "assert_path_without_reparse", side_effect=ValueError("reparse point"), ), ): with self.assertRaises(HTTPException) as raised: users._upload_root() self.assertEqual(503, raised.exception.status_code) target = root / "avatar.png" with patch.object( users, "assert_path_without_reparse", side_effect=ValueError("reparse point"), ): with self.assertRaisesRegex(ValueError, "reparse point"): users._write_new_avatar_atomically(target, b"avatar") self.assertFalse(target.exists()) async def test_unrelated_profile_patch_does_not_restore_stale_avatar(self) -> None: from .deps import Principal, Role from .routes import users principal = Principal( user_id="user-1", role=Role.LEARNER, email="user@example.test", display_name="사용자", ) stale_profile = self._profile( users, avatar_url="/uploads/profile-avatars/stale.png" ) update = AsyncMock(return_value=object()) with ( patch.object(users, "_profile_for", AsyncMock(return_value=stale_profile)), patch.object(users, "update_managed_user", update), patch.object( users, "_profile_avatar_write_lease", side_effect=AssertionError("avatar lease must not be needed"), ), ): await users.patch_me(users.UserProfilePatch(display_name="새 이름"), principal) managed_patch = update.await_args.args[1] self.assertIsNone(managed_patch.avatar_url) async def test_explicit_same_avatar_patch_still_uses_write_lease(self) -> None: from .deps import Principal, Role from .routes import users principal = Principal( user_id="user-1", role=Role.LEARNER, email="user@example.test", display_name="사용자", ) avatar_url = "/uploads/profile-avatars/current.png" profile = self._profile(users, avatar_url=avatar_url) lease = patch.object( users, "_profile_avatar_write_lease", return_value=nullcontext() ) with ( patch.object(users, "_profile_for", AsyncMock(return_value=profile)), patch.object(users, "update_managed_user", AsyncMock(return_value=object())), patch.object(users, "assert_public_avatar_url_exists", return_value=None), lease as lease_mock, ): await users.patch_me(users.UserProfilePatch(avatar_url=avatar_url), principal) lease_mock.assert_called_once_with() async def test_onboarding_always_participates_in_avatar_write_freeze(self) -> None: from .deps import Principal, Role from .routes import users principal = Principal( user_id="user-1", role=Role.LEARNER, email="user@example.test", display_name="사용자", consent_at=1.0, ) profile = self._profile(users, avatar_url="") updated = SimpleNamespace( display_name="학습자", profile_completed_at=1.0 ) lease = patch.object( users, "_profile_avatar_write_lease", return_value=nullcontext() ) body = users.OnboardingRequest( legal_name="학습자", affiliation="한신대학교", department="상담심리학과", grade_level="3학년", phone="010-1234-5678", contact_address="경기도 오산시", nickname="학습자", self_introduction="소개", avatar_url="", terms_accepted=True, privacy_accepted=True, ) with ( patch.object(users, "_profile_for", AsyncMock(return_value=profile)), patch.object(users, "update_managed_user", AsyncMock(return_value=updated)), lease as lease_mock, ): await users.complete_onboarding(body, principal) lease_mock.assert_called_once_with() async def test_patch_failure_preserves_old_db_referenced_avatar(self) -> None: from .deps import Principal, Role from .routes import users with tempfile.TemporaryDirectory(prefix="vignette-avatar-write-") as raw: root = Path(raw) old = root / "user-1-old.png" old.write_bytes(b"old-avatar") user_id = "8da1b8fc-3bdd-4e72-86f7-d17fc315c4cb" principal = Principal( user_id=user_id, role=Role.LEARNER, email="user@example.test", display_name="사용자", ) upload = UploadFile( io.BytesIO(_valid_png_bytes()), filename="avatar.png", headers=Headers({"content-type": "image/png"}), ) with patch.object(users, "_upload_root", return_value=root): uploaded = await users.upload_my_avatar(principal, upload) new_name = uploaded.avatar_url.rsplit("/", 1)[-1] new_path = root / new_name self.assertTrue(new_path.is_file()) self.assertEqual(b"old-avatar", old.read_bytes()) profile = users.UserProfileResponse( user_id="user-1", email="user@example.test", display_name="사용자", role="learner", cohort_ids=[], affiliation="한신대학교", avatar_url="/uploads/profile-avatars/user-1-old.png", ) with ( patch.object(users, "_profile_for", AsyncMock(return_value=profile)), patch.object( users, "update_managed_user", AsyncMock(side_effect=RuntimeError("DB write failed")), ), patch.object( users, "assert_public_avatar_url_exists", return_value=None ), ): with self.assertRaisesRegex(RuntimeError, "DB write failed"): await users.patch_me( users.UserProfilePatch(avatar_url=uploaded.avatar_url), principal, ) self.assertEqual(b"old-avatar", old.read_bytes()) self.assertTrue(new_path.is_file()) async def test_avatar_upload_rejects_non_uuid_principal_before_file_creation( self, ) -> None: from fastapi import HTTPException from .deps import Principal, Role from .routes import users with tempfile.TemporaryDirectory(prefix="vignette-avatar-user-id-") as raw: root = Path(raw) principal = Principal( user_id="not-a-uuid", role=Role.LEARNER, email="user@example.test", display_name="사용자", ) upload = UploadFile( io.BytesIO(_valid_png_bytes()), filename="avatar.png", headers=Headers({"content-type": "image/png"}), ) with patch.object(users, "_upload_root", return_value=root): with self.assertRaises(HTTPException) as raised: await users.upload_my_avatar(principal, upload) self.assertEqual(503, raised.exception.status_code) self.assertEqual([], list(root.iterdir())) if __name__ == "__main__": unittest.main()