from __future__ import annotations import json import os import stat import subprocess import sys import tempfile import unittest from io import BytesIO from pathlib import Path from unittest import mock from typing import Any, cast from PIL import Image import capture import independent_verify HERE = Path(__file__).resolve().parent def image_bytes(fmt: str = "PNG", mode: str = "RGB", size: tuple[int, int] = (3, 2)) -> bytes: out = BytesIO() Image.new(mode, size, 1).save(out, format=fmt) return out.getvalue() class CaptureTests(unittest.TestCase): def __init__(self, methodName: str = "runTest") -> None: super().__init__(methodName) self.temp = cast(tempfile.TemporaryDirectory[str], cast(object, None)) self.root = Path() self.contract = Path() self.patches: list[Any] = [] def setUp(self) -> None: self.temp = tempfile.TemporaryDirectory() self.root = Path(self.temp.name) self.contract = self.root / "capture-contract.json" self.contract.write_bytes((HERE / "capture-contract.json").read_bytes()) self.contract.chmod(0o600) self.patches = [ mock.patch.object(capture, "ROOT", self.root), mock.patch.object(capture, "CONTRACT", self.contract), mock.patch.object(capture, "VAULT", self.root / "vault"), mock.patch.object(capture, "RECEIPT", self.root / "capture-receipt.redacted.json"), ] for patch in self.patches: patch.start() def tearDown(self) -> None: for patch in reversed(self.patches): patch.stop() self.temp.cleanup() def source(self, data: bytes, name: str = "input.png", mode: int = 0o600) -> Path: path = self.root / name path.write_bytes(data) path.chmod(mode) return path def test_png_capture_exact_bytes_mode_receipt_and_privacy(self) -> None: raw = image_bytes() receipt = capture.capture(self.source(raw), capture.ATTESTATION) target = self.root / receipt["capture"]["vault_path"] self.assertEqual(target.read_bytes(), raw) self.assertEqual(stat.S_IMODE(target.stat().st_mode), 0o600) self.assertEqual(receipt["capture"]["image_width"], 3) self.assertEqual(receipt["capture"]["image_height"], 2) serialized = json.dumps(receipt) self.assertNotIn("body_weight", serialized) self.assertNotIn("체크인 감사합니다", serialized) self.assertFalse(receipt["privacy"]["ocr_performed"]) def test_jpeg_capture(self) -> None: receipt = capture.capture(self.source(image_bytes("JPEG"), "input.jpg"), capture.ATTESTATION) self.assertEqual(receipt["capture"]["image_format"], "JPEG") self.assertEqual(receipt["capture"]["image_mode"], "RGB") def test_invalid_and_polyglot_rejected(self) -> None: with self.assertRaises(capture.CaptureError): capture.capture(self.source(b"not an image"), capture.ATTESTATION) self.source(image_bytes() + b"PK\x03\x04", "polyglot.png") with self.assertRaisesRegex(capture.CaptureError, "polyglot"): capture.capture(self.root / "polyglot.png", capture.ATTESTATION) def test_symlink_and_hardlink_rejected_by_nofollow_policy(self) -> None: real = self.source(image_bytes(), "real.png") link = self.root / "link.png" link.symlink_to(real) with self.assertRaises(capture.CaptureError): capture.capture(link, capture.ATTESTATION) hard = self.root / "hard.png" os.link(real, hard) with self.assertRaisesRegex(capture.CaptureError, "single-link"): capture.capture(real, capture.ATTESTATION) def test_oversize_rejected_without_reading(self) -> None: path = self.root / "huge.png" with path.open("wb") as stream: stream.truncate(capture.MAX_BYTES + 1) path.chmod(0o600) with self.assertRaisesRegex(capture.CaptureError, "image size"): capture.capture(path, capture.ATTESTATION) def test_unsafe_source_mode_rejected(self) -> None: with self.assertRaisesRegex(capture.CaptureError, "group/world-writable"): capture.capture(self.source(image_bytes(), mode=0o666), capture.ATTESTATION) with self.assertRaisesRegex(capture.CaptureError, "executable"): capture.capture(self.source(image_bytes(), "exec.png", 0o700), capture.ATTESTATION) def test_disallowed_decoded_mode_rejected(self) -> None: with self.assertRaisesRegex(capture.CaptureError, "mode is not allowed"): capture.capture(self.source(image_bytes("JPEG", "CMYK"), "cmyk.jpg"), capture.ATTESTATION) def test_attestation_mismatch_rejected(self) -> None: with self.assertRaisesRegex(capture.CaptureError, "exact operator attestation"): capture.capture(self.source(image_bytes()), "yes") self.assertFalse((self.root / "vault").exists()) def test_one_use_capture(self) -> None: capture.capture(self.source(image_bytes()), capture.ATTESTATION) second = self.source(image_bytes(size=(4, 4)), "second.png") with self.assertRaisesRegex(capture.CaptureError, "one-use"): capture.capture(second, capture.ATTESTATION) def test_complete_write_handles_short_writes_and_rejects_zero(self) -> None: written = bytearray() def short(_fd: int, data: bytes) -> int: count = min(2, len(data)); written.extend(data[:count]); return count capture._write_all(9, b"abcdef", short) self.assertEqual(written, b"abcdef") with self.assertRaisesRegex(capture.CaptureError, "incomplete write"): capture._write_all(9, b"x", lambda _fd, _data: 0) def test_atomic_write_fsyncs_file_and_directory(self) -> None: calls: list[int] = [] real_fsync = os.fsync def recording(fd: int) -> None: calls.append(stat.S_IFMT(os.fstat(fd).st_mode)); real_fsync(fd) with mock.patch.object(capture.os, "fsync", side_effect=recording): capture.atomic_write(self.root / "atomic", b"bytes") self.assertIn(stat.S_IFREG, calls) self.assertIn(stat.S_IFDIR, calls) self.assertEqual((self.root / "atomic").read_bytes(), b"bytes") def test_receipt_binding_mismatch_is_rejected_independently(self) -> None: capture.capture(self.source(image_bytes()), capture.ATTESTATION) contract_raw = self.contract.read_bytes() entry = {"bytes": len(contract_raw), "mode": "0600", "path": "capture-contract.json", "sha256": capture.digest(contract_raw)} inventory = {"entries": [entry], "schema": "task26-customer-surface-capture-inventory-v1"} inventory_raw = capture.canonical(inventory) (self.root / "inventory.json").write_bytes(inventory_raw) (self.root / "inventory.json").chmod(0o600) seal = {"contract_sha256": capture.digest(contract_raw), "inventory_sha256": capture.digest(inventory_raw), "schema": "task26-customer-surface-capture-seal-v1"} (self.root / "SEAL.json").write_bytes(capture.canonical(seal)) (self.root / "SEAL.json").chmod(0o600) receipt_path = self.root / "capture-receipt.redacted.json" receipt = json.loads(receipt_path.read_text()) receipt["binding"]["expected_provider_message_id"] = "999" receipt_path.write_bytes(capture.canonical(receipt)); receipt_path.chmod(0o600) with mock.patch.object(independent_verify, "ROOT", self.root): with self.assertRaisesRegex(RuntimeError, "receipt binding mismatch"): independent_verify.verify() def test_dry_run_has_no_mutation(self) -> None: result = capture.dry_run() self.assertEqual(result["status"], "PASS_AWAITING_FINAL_RUN_IMAGE") self.assertEqual(result["live_actions"], 0) self.assertFalse((self.root / "vault").exists()) def test_cli_rejects_relative_path(self) -> None: result = subprocess.run( [sys.executable, str(HERE / "capture.py"), "relative.png", "--attest", capture.ATTESTATION], text=True, capture_output=True, check=False, ) self.assertEqual(result.returncode, 2) self.assertIn("absolute Linux", result.stderr) if __name__ == "__main__": unittest.main()