diff options
Diffstat (limited to 'tests/test_report.py')
| -rw-r--r-- | tests/test_report.py | 224 |
1 files changed, 224 insertions, 0 deletions
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() |