diff options
| author | Danilo M. <danix@danix.xyz> | 2026-09-10 11:10:08 +0200 |
|---|---|---|
| committer | Danilo M. <danix@danix.xyz> | 2026-09-10 11:10:08 +0200 |
| commit | 45a68fbbe8da80a7bdf399b2e042f1982ca6a69d (patch) | |
| tree | d56a1f42cc1507b185cbfcc60f2aee63a8ddf1d5 | |
| parent | fb45dbc03650bd9ed7f5629fd01fffdaa951f109 (diff) | |
| download | abusectl-45a68fbbe8da80a7bdf399b2e042f1982ca6a69d.tar.gz abusectl-45a68fbbe8da80a7bdf399b2e042f1982ca6a69d.zip | |
feat: add the report subcommand
Refuses without a configured reporter identity rather than filing a report
with no reply address, and turns a frozen or edited case into an error
message rather than a traceback.
The identity rule is EMAIL REQUIRED, NAME AND ORG OPTIONAL, not the plan's
"all three required". Each [reporter] key is individually skippable at init
and config drops a skipped one rather than storing "", so a partial identity
is a normal shape; text_part() already renders whatever subset is present.
Requiring all three would refuse a config init itself writes without
complaint. The address is different in kind because it becomes the From.
That closes backlog item 5: report.build() read identity["email"] directly
and raised a bare KeyError on a name-without-address identity, which is
exactly what skipping one init prompt while answering another produces. The
check lives in the command rather than the builder, since that is what
decides whether the user sees an exit code and a sentence or a traceback.
Two further defects in the plan's code, both fixed here:
- It called config.load() with no argument, so the global --config option
was silently ignored for this subcommand alone. It now resolves the path
the way _cmd_parse does. Reporting against the wrong identity is the
failure that would have caused.
- Its summary printed counts and no path. Nothing sends these bodies yet, so
reviewing them by hand is the entire point of the command, and the case's
bodies directory is now named in the output.
case.py stays the only writer of the manifest: generate() returns it and
case.save() writes it atomically. Bodies are written before that save, so a
failed save leaves bodies the manifest does not record; the next run finds no
recorded hash, treats them as regenerable rather than hand-edited and
overwrites them, and nothing has been sent, so no machinery is warranted.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LByBnw83xr9YP85nskzkyE
| -rw-r--r-- | abusectl/cli.py | 92 | ||||
| -rw-r--r-- | docs/BACKLOG.md | 10 | ||||
| -rw-r--r-- | tests/test_cli.py | 200 |
3 files changed, 302 insertions, 0 deletions
diff --git a/abusectl/cli.py b/abusectl/cli.py index 3065c34..299d3a1 100644 --- a/abusectl/cli.py +++ b/abusectl/cli.py @@ -35,6 +35,7 @@ from abusectl import contacts as contacts_module from abusectl import init as init_module from abusectl import parse as parse_module from abusectl import rdap as rdap_module +from abusectl import report as report_module EXIT_OK = 0 EXIT_ERROR = 1 @@ -67,6 +68,16 @@ def _build_parser() -> argparse.ArgumentParser: ) contacts_parser.add_argument("case", type=Path) + report_parser = subparsers.add_parser( + "report", help="build report bodies for a case" + ) + report_parser.add_argument("case", type=Path) + report_parser.add_argument( + "--force", + action="store_true", + help="discard hand edits to bodies, keeping a timestamped backup", + ) + return parser @@ -369,6 +380,85 @@ def _cmd_contacts(args) -> int: return EXIT_OK +def _cmd_report(args) -> int: + """Build report bodies and rewrite the manifest. + + Offline and irreversible-free: nothing here sends anything. The output is + what the user reviews before submit does something that cannot be + recalled, so the summary names the directory holding it. A count alone + tells a user something happened and not where to look, and until `submit` + exists reviewing those files by hand is the whole point of running this. + + THE IDENTITY RULE IS: EMAIL REQUIRED, NAME AND ORG OPTIONAL. Each + [reporter] key is individually skippable at init and config drops a + skipped one rather than storing "", so a partial identity is a normal + shape rather than a broken one, and report.text_part() already renders + whatever subset is present. Requiring all three would refuse a config + `init` itself writes without complaint. The address is different in kind: + it becomes the From, so a report built without one either raises + KeyError out of build() (backlog item 5) or, worse, is sent with an + empty addr-spec and cannot be replied to, which defeats the reason the + identity is disclosed at all. + + Checked HERE rather than in report.build(), for the reason backlog item 5 + names: where the check lives decides whether the user gets an exit code + and a sentence or a traceback. It is the same not-configured condition + config.load() raises for a missing trust boundary, so it gets the same + exit code and the same pointer at `init`. + + case.py stays the only writer of the manifest: generate() returns it and + case.save() writes it atomically. Bodies are report.py's own, and they + are written before that save, so a save that fails leaves bodies on disk + that the manifest does not record. That is ACCEPTABLE and needs no + machinery: the next run finds no recorded hash for them, treats them as + regenerable rather than hand-edited, and overwrites them. Nothing has + been sent, and a body whose hash nobody recorded is claimed by nobody. + """ + config_path = args.config if args.config is not None else config.path() + + try: + settings = config.load(config_path) + except config.NotConfigured as exc: + print(f"abusectl report: {exc}", file=sys.stderr) + return EXIT_NOT_CONFIGURED + + identity = settings.reporter + if not identity.get("email"): + print( + "abusectl report: no reporter email configured, and a report " + "with no reply address is one an abuse desk cannot answer. " + f"Set email under [reporter] in {config_path}, or run " + "`abusectl init`.", + file=sys.stderr, + ) + return EXIT_NOT_CONFIGURED + + try: + manifest = case.load(args.case) + except FileNotFoundError: + print(f"abusectl report: no case at {args.case}", file=sys.stderr) + return EXIT_ERROR + except (ValueError, OSError) as exc: + print(f"abusectl report: {args.case}: {exc}", file=sys.stderr) + return EXIT_ERROR + + try: + manifest = report_module.generate( + manifest, args.case, identity, force=args.force + ) + except (report_module.Frozen, report_module.Modified) as exc: + print(f"abusectl report: {exc}", file=sys.stderr) + return EXIT_ERROR + + case.save(args.case, manifest) + + count = len(manifest["destinations"]) + orphans = len(manifest["unreportable"]) + print(f"{count} destinations, {orphans} indicators with no abuse desk") + print(f"bodies in {Path(args.case) / 'bodies'}; review before submitting") + return EXIT_OK + + def main(argv: list[str] | None = None) -> int: parser = _build_parser() args = parser.parse_args(argv) @@ -380,6 +470,8 @@ def main(argv: list[str] | None = None) -> int: return _cmd_parse(args) if args.command == "contacts": return _cmd_contacts(args) + if args.command == "report": + return _cmd_report(args) except KeyboardInterrupt: # Ctrl+C, or EOF at a prompt. Setup writes nothing until every # answer is in hand, so abandoning it leaves no half-written config. diff --git a/docs/BACKLOG.md b/docs/BACKLOG.md index 63ea52c..3ab642f 100644 --- a/docs/BACKLOG.md +++ b/docs/BACKLOG.md @@ -9,6 +9,7 @@ number and gains a status rather than being renumbered. | 2 | An IDN indicator resolves to no contact | S | open | | 3 | Expose kept cases so qtmaildir can tag spam | ? | open, unsized | | 4 | `Report-Type: phishing` is unverified against x-arf | XS | open | +| 5 | `report.build()` raises KeyError on an identity with no email | XS | closed | ## 1. Skip boilerplate namespace URLs @@ -137,6 +138,15 @@ expects. ## 5. `report.build()` raises KeyError on an identity with no email +**Closed** by the `report` subcommand (Task 11). `cli._cmd_report()` refuses +before building, with the not-configured exit code and a sentence naming the +file to edit, rather than letting `build()` raise. The rule it applies is +EMAIL REQUIRED, NAME AND ORG OPTIONAL: the address becomes the `From` and a +report without one cannot be answered, while the other two are individually +skippable at `init` and `text_part()` already renders whatever subset is +present. `tests/test_cli.py` covers both halves, the refusal and the +email-only identity that must still succeed. + **Observed.** `report.build(manifest, destination, identity)` reads `identity.get("name", "")` defensively but `identity["email"]` directly, so an identity carrying a name and no address raises `KeyError: 'email'` rather than diff --git a/tests/test_cli.py b/tests/test_cli.py index b8fffea..89da6b0 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -259,5 +259,205 @@ class ContactsCommand(unittest.TestCase): self.assertEqual(code, cli.EXIT_ERROR) +class ReportCommand(unittest.TestCase): + """The report command's dispatch: exit codes, and what it refuses. + + The identity rule this asserts is EMAIL REQUIRED, NAME AND ORG OPTIONAL, + which is narrower than the plan's "all three". Every [reporter] key is + individually skippable by init and config drops a skipped one, so + demanding all three would refuse a config init itself is happy to write. + report.text_part() renders whatever subset is present, and build() puts + the address in the From, so only the address is load-bearing: a report + with no reply address is one an abuse desk cannot answer. + """ + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.root = pathlib.Path(self._tmp.name) + self.config = self.root / "config.toml" + + def tearDown(self): + self._tmp.cleanup() + + def _write_config(self, reporter: str) -> None: + self.config.write_text( + '[general]\ntrusted_relays = ["192.0.2.0/24"]\n' + reporter, + encoding="utf-8", + ) + + def _make_case(self): + from abusectl import case + + created = case.create( + self.root / "cases", b"From: sender@example.invalid\r\n\r\nbody\r\n" + ) + manifest = case.load(created.path) + manifest["iocs"] = [ + {"id": "ioc-1", "type": "ipv4", "value": "198.51.100.7", + "origin": "received-chain"} + ] + manifest["contacts"] = [{ + "iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["abuse@example.invalid"], "source": "rdap", + }] + case.save(created.path, manifest) + return created.path + + def _run(self, *args): + out, err = io.StringIO(), io.StringIO() + with redirect_stdout(out), redirect_stderr(err): + code = cli.main(list(args)) + return code, out.getvalue(), err.getvalue() + + def test_a_missing_case_is_an_error_not_a_traceback(self): + self._write_config('[reporter]\nemail = "r@example.org"\n') + code, _, err = self._run( + "--config", str(self.config), "report", "/nonexistent/case" + ) + self.assertEqual(code, cli.EXIT_ERROR) + self.assertIn("/nonexistent/case", err) + + def test_a_case_with_no_reporter_configured_says_so(self): + # An unconfigured identity is not-configured, not a crash: the + # report would otherwise be filed with no reply address. + self._write_config("") + case_path = self._make_case() + code, _, err = self._run( + "--config", str(self.config), "report", str(case_path) + ) + self.assertEqual(code, cli.EXIT_NOT_CONFIGURED) + self.assertIn("abusectl init", err) + # Nothing was written. A refusal must leave the case as it was. + self.assertEqual(list((case_path / "bodies").iterdir()), []) + + def test_an_identity_with_no_email_is_refused_rather_than_crashing(self): + # Backlog item 5: build() reads identity["email"] directly, so this + # shape raised KeyError. A name without an address is exactly what + # skipping one init prompt and answering another produces. + self._write_config('[reporter]\nname = "A Reporter"\n') + case_path = self._make_case() + code, _, err = self._run( + "--config", str(self.config), "report", str(case_path) + ) + self.assertEqual(code, cli.EXIT_NOT_CONFIGURED) + self.assertIn("email", err) + self.assertEqual(list((case_path / "bodies").iterdir()), []) + + def test_an_email_alone_is_enough_to_build_a_report(self): + # name and org are genuinely optional, and text_part renders the + # subset that is present. Refusing here would reject a config init + # writes without complaint. + self._write_config('[reporter]\nemail = "r@example.org"\n') + case_path = self._make_case() + code, out, err = self._run( + "--config", str(self.config), "report", str(case_path) + ) + self.assertEqual(code, cli.EXIT_OK, err) + + from abusectl import case + + manifest = case.load(case_path) + self.assertEqual(len(manifest["destinations"]), 1) + body = case_path / manifest["destinations"][0]["body"] + text = body.read_bytes().decode("utf-8") + self.assertIn("From: r@example.org", text) + self.assertIn("Reported by: <r@example.org>", text) + # The summary tells the user where to look: nothing sends these yet, + # so review is the next step and it needs a path. + self.assertIn("1 destinations", out) + self.assertIn(str(case_path / "bodies"), out) + + def test_the_global_config_option_is_honoured(self): + # --config is global and every other subcommand honours it. Calling + # config.load() with no argument would read the user's real config + # and report against the wrong identity, or refuse a configured run. + self._write_config('[reporter]\nemail = "r@example.org"\n') + case_path = self._make_case() + code, _, err = self._run( + "--config", str(self.root / "absent.toml"), "report", str(case_path) + ) + self.assertEqual(code, cli.EXIT_NOT_CONFIGURED) + self.assertIn("absent.toml", err) + + def test_a_frozen_case_is_an_error_not_a_traceback(self): + from abusectl import case + + self._write_config('[reporter]\nemail = "r@example.org"\n') + case_path = self._make_case() + manifest = case.load(case_path) + manifest["frozen"] = {"by": "abuse@example.invalid", "at": "2026-09-10"} + case.save(case_path, manifest) + + code, _, err = self._run( + "--config", str(self.config), "report", str(case_path) + ) + self.assertEqual(code, cli.EXIT_ERROR) + self.assertIn("cannot be regenerated", err) + + def test_an_edited_body_is_refused_and_force_clears_it(self): + from abusectl import case + + self._write_config('[reporter]\nemail = "r@example.org"\n') + case_path = self._make_case() + self._run("--config", str(self.config), "report", str(case_path)) + + body = case_path / case.load(case_path)["destinations"][0]["body"] + body.write_bytes(b"hand edited\n") + + code, _, err = self._run( + "--config", str(self.config), "report", str(case_path) + ) + self.assertEqual(code, cli.EXIT_ERROR) + self.assertIn("--force", err) + self.assertEqual(body.read_bytes(), b"hand edited\n") + + code, _, err = self._run( + "--config", str(self.config), "report", "--force", str(case_path) + ) + self.assertEqual(code, cli.EXIT_OK, err) + self.assertIn("From: r@example.org", body.read_bytes().decode("utf-8")) + backups = list((case_path / "bodies").glob("*.orig")) + self.assertEqual(len(backups), 1) + self.assertEqual(backups[0].read_bytes(), b"hand edited\n") + + def test_a_second_run_on_an_untouched_case_succeeds(self): + self._write_config('[reporter]\nemail = "r@example.org"\n') + case_path = self._make_case() + self._run("--config", str(self.config), "report", str(case_path)) + code, _, err = self._run( + "--config", str(self.config), "report", str(case_path) + ) + self.assertEqual(code, cli.EXIT_OK, err) + + def test_a_corrupt_manifest_is_an_error_not_a_traceback(self): + self._write_config('[reporter]\nemail = "r@example.org"\n') + case_path = self._make_case() + (case_path / "manifest.json").write_text("{not json") + code, _, err = self._run( + "--config", str(self.config), "report", str(case_path) + ) + self.assertEqual(code, cli.EXIT_ERROR) + + def test_a_case_with_no_contacts_reports_nothing_rather_than_failing(self): + # generate() always sets both keys, so the summary can index them. + from abusectl import case + + self._write_config('[reporter]\nemail = "r@example.org"\n') + case_path = self._make_case() + manifest = case.load(case_path) + manifest["contacts"] = [] + del manifest["destinations"] + case.save(case_path, manifest) + + code, out, err = self._run( + "--config", str(self.config), "report", str(case_path) + ) + self.assertEqual(code, cli.EXIT_OK, err) + self.assertIn("0 destinations", out) + written = case.load(case_path) + self.assertEqual(written["destinations"], []) + self.assertEqual(written["unreportable"], []) + + if __name__ == "__main__": unittest.main() |
