From be392fe195f8f15dbe263b4c39d609bce95a8615 Mon Sep 17 00:00:00 2001 From: "Danilo M." Date: Thu, 10 Sep 2026 10:20:55 +0200 Subject: feat: write report bodies into the case directory case.py stays the only writer of the manifest, so generate returns it and the caller saves. A modified body is backed up with a timestamp before --force overwrites it. Bodies are written and read as BYTES, not text. The plan used read_text and write_text, which fail two ways at once. The locale encoding is not the utf-8 body_hash pins, so a body hashed differently under a non-UTF-8 LANG; and text mode applies universal newlines, so a report delimited by CRLF came back with every CRLF collapsed to LF. The second was live on every platform: regenerating an untouched case raised Modified, because the body read back never hashed to what was written. For a sent destination that hash is the record of what was disclosed, so it has to be the hash of the bytes on disk. Text mode would also translate the bare LFs inside the machine-readable part on a CRLF platform, corrupting the field block a desk's parser reads. The refusal ordering is now pinned by tests rather than left implicit: a frozen or modified case must leave the bodies directory byte-for-byte as it was, asserted over the whole directory rather than one file. bodies is created without parents, so a missing case directory raises rather than scattering bodies into a tree generate invented. The backup test pins the filename, not just the count of .orig files. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01LByBnw83xr9YP85nskzkyE --- tests/test_report.py | 224 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 224 insertions(+) (limited to 'tests/test_report.py') diff --git a/tests/test_report.py b/tests/test_report.py index b36252c..e87956f 100644 --- a/tests/test_report.py +++ b/tests/test_report.py @@ -3,7 +3,10 @@ import email import email.policy import hashlib import json +import re +import tempfile import unittest +from pathlib import Path from abusectl import report @@ -1706,5 +1709,226 @@ class Hashes(unittest.TestCase): self.assertEqual(len(report.body_hash(text)), 64) +class Writing(unittest.TestCase): + """generate() writing bodies into a case directory. + + A body is a byte-exact RFC 5322 document, not a text file the platform + may reflow, which is what most of these tests are really about. + """ + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.path = Path(self.tmp.name) + (self.path / "bodies").mkdir() + self.addCleanup(self.tmp.cleanup) + + def _generate(self, manifest=None, **kwargs): + if manifest is None: + manifest = copy.deepcopy(MANIFEST) + return report.generate(manifest, self.path, IDENTITY, **kwargs) + + def test_it_writes_one_body_per_destination_and_records_the_hash(self): + manifest = self._generate() + destination = manifest["destinations"][0] + body = self.path / destination["body"] + self.assertTrue(body.exists()) + self.assertEqual( + destination["body_sha256"], + report.body_hash(body.read_bytes().decode("utf-8")), + ) + + def test_the_body_path_is_named_for_the_destination_id(self): + """The id names the file, because the edit check pairs the two.""" + manifest = self._generate() + destination = manifest["destinations"][0] + self.assertEqual(destination["body"], + f"bodies/{destination['id']}.xarf") + + def test_the_recorded_hash_is_the_hash_of_the_bytes_on_disk(self): + """The round trip must be byte-exact, and this is the test that + says so for every destination rather than the first. + + A report is CRLF-delimited per RFC 5322 while its feedback part + carries bare LFs, so a text-mode write can translate the bare LFs + and a text-mode read collapses every CRLF. Either one makes the + recorded hash disagree with the file, which reads as a hand edit + and, for a sent destination, is a false record of what was + disclosed. + """ + manifest = self._generate() + for destination in manifest["destinations"]: + with self.subTest(destination=destination["id"]): + raw = (self.path / destination["body"]).read_bytes() + self.assertEqual(destination["body_sha256"], + hashlib.sha256(raw).hexdigest()) + + def test_the_written_body_keeps_its_crlf_line_endings(self): + """Verified against the document, not assumed from the policy.""" + manifest = self._generate() + raw = (self.path / manifest["destinations"][0]["body"]).read_bytes() + self.assertIn(b"\r\n", raw) + # The header block is pure CRLF: a bare LF there would mean the + # write translated line endings, which is the corruption a + # text-mode write causes on a platform whose linesep is CRLF. + header, _, _ = raw.partition(b"\r\n\r\n") + self.assertNotIn(b"\n", header.replace(b"\r\n", b"")) + # And the bare LFs inside the feedback part survive untranslated, + # which is the same defect seen from the other side. + self.assertIn(b"Feedback-Type: abuse\nUser-Agent:", raw) + + def test_an_untouched_case_regenerates_without_complaint(self): + """The regression that the plan's read_text() caused. + + Nothing was edited, so a second run must simply rewrite the + bodies. Under a lossy round trip every body reads as hand-edited + and the second run refuses, which makes report unusable twice. + """ + manifest = self._generate() + again = report.generate(copy.deepcopy(manifest), self.path, IDENTITY) + self.assertEqual([d["id"] for d in again["destinations"]], + [d["id"] for d in manifest["destinations"]]) + self.assertEqual(list((self.path / "bodies").glob("*.orig")), []) + + def test_it_records_the_unreportable_indicators(self): + manifest = copy.deepcopy(MANIFEST) + manifest["contacts"] = manifest["contacts"] + [ + {"iocs": ["ioc-4"], "query": "x.invalid", "abuse": [], + "source": "rdap", "error": "no abuse role published"}, + ] + result = self._generate(manifest) + self.assertEqual( + result["unreportable"], + [{"ioc": "ioc-4", "reason": "no abuse role published"}], + ) + + def test_a_modified_body_is_backed_up_before_being_overwritten(self): + manifest = self._generate() + destination_id = manifest["destinations"][0]["id"] + body = self.path / manifest["destinations"][0]["body"] + body.write_bytes(b"hand edited during review\r\n") + + with self.assertRaises(report.Modified): + report.generate(copy.deepcopy(manifest), self.path, IDENTITY) + + report.generate(copy.deepcopy(manifest), self.path, IDENTITY, + force=True) + backups = list((self.path / "bodies").glob("*.orig")) + self.assertEqual(len(backups), 1) + # Pin the NAME, not just the count: a backup written beside the + # wrong destination still globs as one file. The stamp sits + # between the extension and .orig so the id stays readable. + self.assertTrue( + re.fullmatch(rf"{destination_id}\.xarf\.\d{{8}}T\d{{6}}Z\.orig", + backups[0].name), + backups[0].name, + ) + self.assertEqual(backups[0].read_bytes(), + b"hand edited during review\r\n") + # And the body itself was replaced rather than left as the edit. + self.assertNotEqual(body.read_bytes(), + b"hand edited during review\r\n") + + def test_a_backup_preserves_the_edited_bytes_exactly(self): + """The backup is the user's twenty minutes of review, so it is + copied verbatim rather than through a text-mode round trip. + """ + manifest = self._generate() + body = self.path / manifest["destinations"][0]["body"] + edited = "line one\r\nline two\nnon-ascii: Örg\r\n" + body.write_bytes(edited.encode("utf-8")) + report.generate(copy.deepcopy(manifest), self.path, IDENTITY, + force=True) + backups = list((self.path / "bodies").glob("*.orig")) + self.assertEqual(backups[0].read_bytes(), edited.encode("utf-8")) + + def test_a_deleted_body_regenerates_without_complaint(self): + manifest = self._generate() + (self.path / manifest["destinations"][0]["body"]).unlink() + again = report.generate(copy.deepcopy(manifest), self.path, IDENTITY) + self.assertTrue((self.path / again["destinations"][0]["body"]).exists()) + + def test_a_frozen_case_writes_nothing_at_all(self): + """A refusal must leave the bodies directory exactly as it was. + + check_regenerable runs after the scan and before every write, so + no body is rewritten and no backup is renamed. Asserting the + directory's full state is what makes that ordering testable: a + partial write on a frozen case corrupts an evidence record a desk + already holds. + """ + manifest = self._generate() + before = {p.name: p.read_bytes() + for p in (self.path / "bodies").iterdir()} + manifest["frozen"] = {"at": "2026-09-07T10:00:00Z", "by": "abusedb"} + + with self.assertRaises(report.Frozen): + report.generate(copy.deepcopy(manifest), self.path, IDENTITY) + with self.assertRaises(report.Frozen): + report.generate(copy.deepcopy(manifest), self.path, IDENTITY, + force=True) + + after = {p.name: p.read_bytes() + for p in (self.path / "bodies").iterdir()} + self.assertEqual(after, before) + + def test_a_modified_refusal_writes_nothing_at_all(self): + """Same ordering, the other refusal. Without --force an edited + body must survive untouched, including every body that was NOT + edited: a partial rewrite would discard part of a review. + """ + manifest = self._generate() + body = self.path / manifest["destinations"][0]["body"] + body.write_bytes(b"hand edited during review\r\n") + before = {p.name: p.read_bytes() + for p in (self.path / "bodies").iterdir()} + + with self.assertRaises(report.Modified): + report.generate(copy.deepcopy(manifest), self.path, IDENTITY) + + after = {p.name: p.read_bytes() + for p in (self.path / "bodies").iterdir()} + self.assertEqual(after, before) + + def test_it_refuses_a_case_directory_that_does_not_exist(self): + """generate does not conjure a case. case.py creates a case, and a + missing directory means a wrong path rather than a first run, so + creating the tree here would scatter bodies into an empty dir. + """ + with self.assertRaises(FileNotFoundError): + report.generate(copy.deepcopy(MANIFEST), self.path / "absent", + IDENTITY) + + def test_no_recipient_identifier_reaches_a_written_body(self): + """Property 1, checked against the bytes that land on disk. + + The manifest is given a recipient address in every place a body + draws from, and none of it may appear in a file. + """ + manifest = copy.deepcopy(MANIFEST) + secret = "victim@example.org" + manifest["headers"] = manifest["headers"] + [ + ("To", f"Victim <{secret}>"), + ("Delivered-To", secret), + ] + manifest["iocs"] = manifest["iocs"] + [ + {"id": "ioc-3", "type": "url", + "value": f"http://kit.invalid/verify?e={secret}", + "origin": "body"}, + ] + result = self._generate(manifest) + for destination in result["destinations"]: + raw = (self.path / destination["body"]).read_bytes() + with self.subTest(destination=destination["id"]): + self.assertNotIn(secret.encode("utf-8"), raw) + self.assertNotIn(b"victim", raw.lower()) + + def test_it_returns_the_manifest_rather_than_saving_it(self): + """case.py is the only writer of a manifest.""" + result = self._generate() + self.assertNotIn("manifest.json", + [p.name for p in self.path.iterdir()]) + self.assertIn("destinations", result) + + if __name__ == "__main__": unittest.main() -- cgit v1.2.3