from __future__ import annotations
import warnings
from io import BytesIO
from zipfile import ZIP_DEFLATED, ZipFile
import pytest
from PIL import Image
from app.masking.detector import RegexDetector
from app.media.docx_sanitizer import DocxSanitizer
from app.media.image_sanitizer import ImageSanitizer
from app.media.pdf_sanitizer import PdfSanitizer
from app.media.text_redactor import TextRedactor
from app.media.types import MediaSanitizationError
from app.policies.policy_engine import PolicyEngine
CONTENT_TYPES = (
b''
b''
b''
b''
b''
b""
)
ROOT_RELS = (
b''
b''
b''
b""
)
DOCUMENT = b"""
Safe synthetic fixture
"""
class NoTextOCR:
def extract(self, image: Image.Image) -> list[object]:
return []
class NoFaces:
def detect(self, image: Image.Image) -> list[tuple[int, int, int, int]]:
return []
@pytest.fixture
def redactor(settings) -> TextRedactor:
policy = PolicyEngine(
settings.policy_file,
block_api_keys=False,
block_secrets=False,
block_credit_cards=False,
)
return TextRedactor(RegexDetector(), policy)
@pytest.fixture
def image_sanitizer(redactor: TextRedactor) -> ImageSanitizer:
return ImageSanitizer(redactor, NoTextOCR(), NoFaces())
def _docx(
extras: tuple[tuple[str, bytes], ...] = (),
*,
duplicate_document: bool = False,
) -> bytes:
output = BytesIO()
with warnings.catch_warnings():
warnings.simplefilter("ignore", UserWarning)
with ZipFile(output, "w", ZIP_DEFLATED) as archive:
archive.writestr("[Content_Types].xml", CONTENT_TYPES)
archive.writestr("_rels/.rels", ROOT_RELS)
archive.writestr("word/document.xml", DOCUMENT)
if duplicate_document:
archive.writestr("word/document.xml", DOCUMENT)
for name, content in extras:
archive.writestr(name, content)
return output.getvalue()
def _pdf(page_count: int = 1, size: tuple[int, int] = (100, 100)) -> bytes:
pages = [Image.new("RGB", size, "white") for _ in range(page_count)]
output = BytesIO()
try:
pages[0].save(output, format="PDF", save_all=True, append_images=pages[1:])
finally:
for page in pages:
page.close()
return output.getvalue()
def test_docx_rejects_duplicate_zip_entries(redactor: TextRedactor) -> None:
sanitizer = DocxSanitizer(redactor)
with pytest.raises(MediaSanitizationError, match="duplicate ZIP entries") as exc_info:
sanitizer.sanitize(_docx(duplicate_document=True), "duplicate.docx")
assert exc_info.value.error_type == "unsafe_docx"
def test_docx_rejects_entry_count_over_budget(redactor: TextRedactor) -> None:
sanitizer = DocxSanitizer(redactor, max_entries=3)
payload = _docx((("word/styles.xml", b"",),))
with pytest.raises(MediaSanitizationError, match="too many ZIP entries") as exc_info:
sanitizer.sanitize(payload, "too-many-parts.docx")
assert exc_info.value.error_type == "unsafe_docx"
def test_docx_rejects_uncompressed_content_over_budget(redactor: TextRedactor) -> None:
sanitizer = DocxSanitizer(redactor, max_uncompressed_bytes=128)
with pytest.raises(MediaSanitizationError, match="uncompressed content") as exc_info:
sanitizer.sanitize(_docx(), "expanded.docx")
assert exc_info.value.error_type == "unsafe_docx"
def test_docx_rejects_unsafe_compression_ratio(redactor: TextRedactor) -> None:
sanitizer = DocxSanitizer(redactor, max_compression_ratio=2)
compressible_xml = (
b""
)
payload = _docx((("word/styles.xml", compressible_xml),))
with pytest.raises(MediaSanitizationError, match="compression ratio") as exc_info:
sanitizer.sanitize(payload, "compression-bomb.docx")
assert exc_info.value.error_type == "unsafe_docx"
def test_docx_rejects_unknown_opaque_parts(redactor: TextRedactor) -> None:
sanitizer = DocxSanitizer(redactor)
payload = _docx((("attachments/private.bin", b"synthetic opaque bytes"),))
with pytest.raises(MediaSanitizationError, match="opaque or unsupported") as exc_info:
sanitizer.sanitize(payload, "opaque.docx")
assert exc_info.value.error_type == "unsupported_docx_content"
assert exc_info.value.status_code == 415
def test_image_rejects_dimensions_over_pixel_budget(redactor: TextRedactor) -> None:
source = BytesIO()
Image.new("RGB", (20, 20), "white").save(source, format="PNG")
sanitizer = ImageSanitizer(redactor, NoTextOCR(), NoFaces(), max_pixels=399)
with pytest.raises(MediaSanitizationError, match="dimensions") as exc_info:
sanitizer.sanitize(source.getvalue(), "oversized.png")
assert exc_info.value.error_type == "unsafe_image"
assert exc_info.value.status_code == 413
def test_image_rejects_multiframe_input(redactor: TextRedactor) -> None:
source = BytesIO()
frames = [Image.new("RGB", (16, 16), color) for color in ("white", "black")]
try:
frames[0].save(source, format="TIFF", save_all=True, append_images=frames[1:])
finally:
for frame in frames:
frame.close()
sanitizer = ImageSanitizer(redactor, NoTextOCR(), NoFaces())
with pytest.raises(MediaSanitizationError, match="multi-frame") as exc_info:
sanitizer.sanitize(source.getvalue(), "multipage.tiff")
assert exc_info.value.error_type == "unsupported_media_type"
assert exc_info.value.status_code == 415
def test_pdf_rejects_page_count_over_budget(image_sanitizer: ImageSanitizer) -> None:
sanitizer = PdfSanitizer(image_sanitizer, max_pages=1, render_scale=1)
with pytest.raises(MediaSanitizationError, match="between 1 and 1 pages") as exc_info:
sanitizer.sanitize(_pdf(page_count=2), "too-many-pages.pdf")
assert exc_info.value.error_type == "unsafe_pdf"
assert exc_info.value.status_code == 413
def test_pdf_rejects_page_over_pixel_budget(image_sanitizer: ImageSanitizer) -> None:
sanitizer = PdfSanitizer(
image_sanitizer,
render_scale=1,
max_page_pixels=1_000,
max_total_pixels=1_000_000,
)
with pytest.raises(MediaSanitizationError, match="page exceeds") as exc_info:
sanitizer.sanitize(_pdf(size=(100, 100)), "wide-page.pdf")
assert exc_info.value.error_type == "unsafe_pdf"
assert exc_info.value.status_code == 413
def test_pdf_rejects_total_pixels_over_budget(image_sanitizer: ImageSanitizer) -> None:
sanitizer = PdfSanitizer(
image_sanitizer,
max_pages=2,
render_scale=1,
max_page_pixels=20_000,
max_total_pixels=15_000,
)
with pytest.raises(MediaSanitizationError, match="total pixel") as exc_info:
sanitizer.sanitize(_pdf(page_count=2, size=(100, 100)), "pixel-budget.pdf")
assert exc_info.value.error_type == "unsafe_pdf"
assert exc_info.value.status_code == 413