diff options
| -rw-r--r-- | abusectl/report.py | 98 | ||||
| -rw-r--r-- | tests/test_report.py | 224 |
2 files changed, 322 insertions, 0 deletions
diff --git a/abusectl/report.py b/abusectl/report.py index 38a50ff..20eb58a 100644 --- a/abusectl/report.py +++ b/abusectl/report.py @@ -28,10 +28,12 @@ itself is a module that can disclose it in a code path nobody reviewed. import hashlib +from datetime import datetime, timezone from email.header import Header from email.headerregistry import Address from email.message import EmailMessage from email.policy import SMTP +from pathlib import Path VERSION = "0.1.0" @@ -954,3 +956,99 @@ def check_regenerable(manifest: dict, modified: list[str], ". Re-run with --force to discard those edits; a timestamped " "backup is kept." ) + + +def _read_body(path: Path) -> str: + """Read a body back EXACTLY as it was written. + + Deliberately not read_text(). A body is an RFC 5322 document delimited + by CRLF, while the feedback part inside it carries bare LFs that + set_content chose; text mode applies universal newlines and collapses + every CRLF to LF on the way in. The text that comes back then hashes + differently from the text that was written even though not one byte on + disk changed, so an untouched case reads as hand-edited and refuses to + regenerate, and for a sent destination the recorded hash is a false + record of what was disclosed. + + The encoding is pinned to utf-8 for the same reason body_hash pins it: + the locale of the run that reads a body is not the locale of the run + that wrote it. + """ + return path.read_bytes().decode("utf-8") + + +def _write_body(path: Path, text: str) -> None: + """Write a body byte-exactly, in the encoding body_hash assumes. + + Binary rather than write_text for the mirror of the reason above: text + mode translates a bare LF to os.linesep, which on a CRLF platform + would rewrite the bare LFs inside the machine-readable part and + corrupt the field block a desk's parser reads. The bytes written here + are the bytes hashed, which is what makes the hash a record of what + was disclosed rather than of what was intended. + """ + path.write_bytes(text.encode("utf-8")) + + +def generate(manifest: dict, case_path: Path, identity: dict, + force: bool = False) -> dict: + """Write every body and return the manifest with destinations[] set. + + The manifest is RETURNED rather than saved: case.py is the only writer + of a case directory's manifest, so the caller saves. The bodies are + this module's to write because they are not the manifest. + + The case directory must already exist. `bodies` is created if it is + missing, but its parents are NOT, because a missing case directory + means a wrong path rather than a first run and creating the tree would + scatter bodies into an empty directory the user never made. + + ORDER MATTERS AND IS THE POINT. The whole manifest is scanned for + edited bodies first, then check_regenerable decides, and only then does + anything on disk change. A refusal therefore leaves the bodies + directory byte-for-byte as it was: a frozen case is an evidence record + a desk already holds, and a half-rewritten one is worse than a stale + one. Scanning every destination before refusing is also what makes the + error name every edited body rather than the first. + """ + case_path = Path(case_path) + bodies = case_path / "bodies" + bodies.mkdir(exist_ok=True) + + modified = [] + for existing in manifest.get("destinations") or []: + recorded = existing.get("body_sha256") + if not recorded or not existing.get("body"): + continue + path = case_path / existing["body"] + # A deleted body is not a modified one. It regenerates silently, + # because deleting a body is how a user asks for a fresh one. + if path.exists() and body_hash(_read_body(path)) != recorded: + modified.append(existing["id"]) + + check_regenerable(manifest, modified, force=force) + + stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + for destination_id in modified: + existing = next(d for d in manifest["destinations"] + if d["id"] == destination_id) + path = case_path / existing["body"] + # The stamp goes AFTER the full name rather than replacing the + # extension, so the destination id stays legible in a directory + # listing and two forced runs in different seconds keep both + # edits. with_suffix would be wrong here: it replaces the last + # suffix, so it depends on the name having exactly the one this + # writer gave it. + path.rename(path.with_name(f"{path.name}.{stamp}.orig")) + + destinations = email_destinations(manifest.get("contacts", [])) + for destination in destinations: + text = build(manifest, destination, identity) + relative = f"bodies/{destination['id']}.xarf" + _write_body(case_path / relative, text) + destination["body"] = relative + destination["body_sha256"] = body_hash(text) + + manifest["destinations"] = destinations + manifest["unreportable"] = unreportable(manifest.get("contacts", [])) + return manifest 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() |