d591813f02
The document matcher decided by extension and fell through to "decode whatever this is as UTF-8" for everything that was not a .pdf. A shell script, an HTML page, a CSV or a JPEG all became a search query, and a file called report.pdf holding something else was read as a PDF. Three questions now, in order, before anything else touches the file: is it under 2 MB (read one byte past the cap, so a huge file is never held in memory to be measured); what is it, sniffed from the leading bytes; and what text is in it. PDF, DOCX and images, and nothing else — 415 with a sentence naming what to upload instead. DOCX is parsed from the zip with no new dependency and is checked against the one attack that shape allows, a member that unpacks to far more than the file's size suggests. An image is read by the tool model, and where no tool model is configured it says so rather than 500ing. On injection, since that is the question people mean: there is no path from an uploaded file to code that runs. The extracted text is a search query — bound parameter to Postgres, never concatenated into SQL — and the content of a message to a model. It is never rendered as HTML, never written to disk, never passed to a shell. Control characters are stripped because they make queries that match nothing, not because anything would interpret them. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TqXevQJhxFrM7jJg82cgZN
99 lines
4 KiB
Python
99 lines
4 KiB
Python
"""What may be uploaded, and what the bytes have to say about it.
|
|
|
|
Run: DATABASE_URL=sqlite:// PYTHONPATH=backend python -m unittest discover -s backend/tests
|
|
"""
|
|
import os
|
|
os.environ.setdefault("DATABASE_URL", "sqlite://")
|
|
|
|
import io
|
|
import unittest
|
|
import zipfile
|
|
|
|
from app.services import file_intake
|
|
|
|
|
|
def docx(text: str = "Febrile seizures are common between six months and five years.") -> bytes:
|
|
buffer = io.BytesIO()
|
|
with zipfile.ZipFile(buffer, "w") as bundle:
|
|
bundle.writestr("[Content_Types].xml", "<Types/>")
|
|
bundle.writestr(
|
|
"word/document.xml",
|
|
f"<w:document><w:body><w:p><w:r><w:t>{text}</w:t></w:r></w:p></w:body></w:document>")
|
|
return buffer.getvalue()
|
|
|
|
|
|
class Upload:
|
|
"""The two attributes the reader touches on a Starlette UploadFile."""
|
|
|
|
def __init__(self, data: bytes):
|
|
self.file = io.BytesIO(data)
|
|
|
|
|
|
PNG = b"\x89PNG\r\n\x1a\n" + b"0" * 40
|
|
|
|
|
|
class SniffingTests(unittest.TestCase):
|
|
def test_the_bytes_decide_not_the_extension(self):
|
|
self.assertEqual(file_intake.kind_of(b"%PDF-1.7\nstuff"), file_intake.PDF)
|
|
self.assertEqual(file_intake.kind_of(docx()), file_intake.DOCX)
|
|
self.assertEqual(file_intake.kind_of(PNG), file_intake.IMAGE)
|
|
self.assertEqual(file_intake.kind_of(b"RIFF\x00\x00\x00\x00WEBPmore"), file_intake.IMAGE)
|
|
|
|
def test_everything_else_is_refused(self):
|
|
# A shell script called report.pdf is not a PDF; a zip that is not a
|
|
# Word document is not one either.
|
|
self.assertIsNone(file_intake.kind_of(b"#!/bin/sh\nrm -rf /"))
|
|
self.assertIsNone(file_intake.kind_of(b"<html><script>alert(1)</script>"))
|
|
self.assertIsNone(file_intake.kind_of(b"id,name\n1,two\n"))
|
|
plain_zip = io.BytesIO()
|
|
with zipfile.ZipFile(plain_zip, "w") as bundle:
|
|
bundle.writestr("notes.txt", "hello")
|
|
self.assertIsNone(file_intake.kind_of(plain_zip.getvalue()))
|
|
self.assertIsNone(file_intake.kind_of(b""))
|
|
|
|
|
|
class SizeTests(unittest.TestCase):
|
|
def test_over_the_cap_is_refused_by_the_byte_past_it(self):
|
|
with self.assertRaises(file_intake.Rejected) as refusal:
|
|
file_intake.read(Upload(b"x" * (file_intake.MAX_BYTES + 1)))
|
|
self.assertIn("2 MB", str(refusal.exception))
|
|
|
|
def test_exactly_the_cap_is_allowed_and_empty_is_not(self):
|
|
self.assertEqual(len(file_intake.read(Upload(b"x" * file_intake.MAX_BYTES))),
|
|
file_intake.MAX_BYTES)
|
|
with self.assertRaises(file_intake.Rejected):
|
|
file_intake.read(Upload(b""))
|
|
|
|
|
|
class TextTests(unittest.TestCase):
|
|
def test_a_word_document_gives_up_its_words(self):
|
|
text = file_intake.text_from(docx(), file_intake.DOCX, 4000)
|
|
self.assertIn("Febrile seizures are common", text)
|
|
# Tags never reach the text, so no markup can be smuggled through one.
|
|
self.assertNotIn("<w:t>", text)
|
|
|
|
def test_a_zip_bomb_is_refused_before_it_is_read(self):
|
|
buffer = io.BytesIO()
|
|
with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as bundle:
|
|
bundle.writestr("[Content_Types].xml", "<Types/>")
|
|
bundle.writestr("word/document.xml", "<w:t>" + ("a" * (file_intake.MAX_UNZIPPED + 10)))
|
|
with self.assertRaises(file_intake.Rejected):
|
|
file_intake.text_from(buffer.getvalue(), file_intake.DOCX, 4000)
|
|
|
|
def test_control_characters_go_and_the_words_stay(self):
|
|
self.assertEqual(file_intake.clean("a\x00b\x07 c"), "a b c")
|
|
|
|
def test_an_image_without_a_reader_says_so_rather_than_failing(self):
|
|
with self.assertRaises(file_intake.Rejected) as refusal:
|
|
file_intake.text_from(PNG, file_intake.IMAGE, 4000, describe=None)
|
|
self.assertIn("tool model", str(refusal.exception))
|
|
|
|
def test_an_image_with_a_reader_is_read(self):
|
|
text = file_intake.text_from(
|
|
PNG, file_intake.IMAGE, 4000,
|
|
describe=lambda data, mime: f"A slide about croup ({mime})")
|
|
self.assertEqual(text, "A slide about croup (image/png)")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|