From 7fbd109ed8e00ca1184015ba50c30ee92cce1ad8 Mon Sep 17 00:00:00 2001 From: "Danilo M." Date: Wed, 9 Sep 2026 09:47:19 +0200 Subject: feat: add the contacts subcommand A re-run overwrites contacts[] wholesale rather than merging. A merge would let a contact resolved a week ago survive into a report filed today, which is the stale-address hazard the response caching policy already refuses, and overwriting makes a re-run always safe, which matters because a partial network failure makes re-running the natural next step. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Wrfqr2xqQfhtXCscU7zrdz --- abusectl/cli.py | 54 +++++++++++++++++++++++++++++++++++++++- tests/test_cli.py | 73 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 126 insertions(+), 1 deletion(-) diff --git a/abusectl/cli.py b/abusectl/cli.py index 3872a85..2200e14 100644 --- a/abusectl/cli.py +++ b/abusectl/cli.py @@ -18,7 +18,9 @@ Everything this needs (config, case, parse, init) is already built and tested elsewhere; this module wires argparse to those modules and decides what exit code a failure gets. init.py is imported as `init_module` because a subcommand handler here is also named for its subcommand, and importing -plainly would collide. +plainly would collide. contacts.py is imported the same way for the same +reason, and rdap.py follows the convention so the network modules read +alike. """ import argparse @@ -29,8 +31,10 @@ from pathlib import Path from abusectl import __version__ from abusectl import case from abusectl import config +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 EXIT_OK = 0 EXIT_ERROR = 1 @@ -58,6 +62,11 @@ def _build_parser() -> argparse.ArgumentParser: parse_parser = subparsers.add_parser("parse", help="parse a message into a case") parse_parser.add_argument("message", type=Path) + contacts_parser = subparsers.add_parser( + "contacts", help="resolve abuse contacts for a case" + ) + contacts_parser.add_argument("case", type=Path) + return parser @@ -263,6 +272,47 @@ def _cmd_parse(args) -> int: return EXIT_OK +def _cmd_contacts(args) -> int: + """Resolve abuse contacts and rewrite the manifest. + + A re-run overwrites contacts[] wholesale rather than merging. A merge + would let a contact resolved a week ago survive into a report filed + today, which is the stale-address hazard the response caching policy + already refuses, and a stale abuse address sends the report into a + mailbox nobody reads. Overwriting makes a re-run always safe and always + current, which matters because a partial network failure makes + re-running the natural next step. + """ + try: + manifest = case.load(args.case) + except FileNotFoundError: + print(f"abusectl contacts: no case at {args.case}", file=sys.stderr) + return EXIT_ERROR + except (ValueError, OSError) as exc: + # An unreadable or unknown-format manifest. The victim of a + # traceback here is the user mid-incident, who gets a stack trace + # instead of the path that is wrong. + print(f"abusectl contacts: {args.case}: {exc}", file=sys.stderr) + return EXIT_ERROR + + try: + bootstraps = { + name: rdap_module.bootstrap(name) for name in ("ipv4", "ipv6", "dns") + } + except rdap_module.BootstrapUnavailable as exc: + print(f"abusectl contacts: {exc}", file=sys.stderr) + return EXIT_ERROR + + resolved = contacts_module.resolve(manifest.get("iocs", []), bootstraps) + + manifest["contacts"] = resolved + case.save(args.case, manifest) + + unresolved = sum(1 for entry in resolved if not entry["abuse"]) + print(f"{len(resolved)} contacts, {unresolved} without an abuse address") + return EXIT_OK + + def main(argv: list[str] | None = None) -> int: parser = _build_parser() args = parser.parse_args(argv) @@ -272,6 +322,8 @@ def main(argv: list[str] | None = None) -> int: return _cmd_init(args) if args.command == "parse": return _cmd_parse(args) + if args.command == "contacts": + return _cmd_contacts(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/tests/test_cli.py b/tests/test_cli.py index fc4a6e7..1c023c3 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -163,5 +163,78 @@ class TestParse(unittest.TestCase): self.assertIn("absent.eml", err) +class ContactsCommand(unittest.TestCase): + def test_contacts_rewrites_the_manifest(self): + from unittest import mock + + from abusectl import case + + with tempfile.TemporaryDirectory() as tmp: + root = pathlib.Path(tmp) + created = case.create(root, 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"} + ] + case.save(created.path, manifest) + + fake_contacts = [{ + "iocs": ["ioc-1"], "query": "198.51.100.7", + "abuse": ["abuse@example.invalid"], "source": "rdap", + }] + + out = io.StringIO() + with mock.patch("abusectl.cli.contacts_module.resolve", + return_value=fake_contacts) as resolve, \ + mock.patch("abusectl.cli.rdap_module.bootstrap", + return_value={"services": []}), \ + redirect_stdout(out): + code = cli.main(["contacts", str(created.path)]) + + self.assertEqual(code, cli.EXIT_OK) + self.assertTrue(resolve.called) + written = case.load(created.path) + self.assertEqual(written["contacts"], fake_contacts) + + def test_a_missing_case_is_an_error_not_a_traceback(self): + err = io.StringIO() + with redirect_stderr(err): + code = cli.main(["contacts", "/nonexistent/case/path"]) + self.assertEqual(code, cli.EXIT_ERROR) + + def test_an_unavailable_bootstrap_is_an_error_not_a_traceback(self): + from unittest import mock + + from abusectl import case, rdap + + with tempfile.TemporaryDirectory() as tmp: + created = case.create( + pathlib.Path(tmp), b"From: sender@example.invalid\r\n\r\nbody\r\n" + ) + err = io.StringIO() + with mock.patch( + "abusectl.cli.rdap_module.bootstrap", + side_effect=rdap.BootstrapUnavailable("no bootstrap and no cache"), + ), redirect_stderr(err): + code = cli.main(["contacts", str(created.path)]) + + self.assertEqual(code, cli.EXIT_ERROR) + self.assertIn("no bootstrap and no cache", err.getvalue()) + + def test_a_corrupt_manifest_is_an_error_not_a_traceback(self): + from abusectl import case + + with tempfile.TemporaryDirectory() as tmp: + created = case.create( + pathlib.Path(tmp), b"From: sender@example.invalid\r\n\r\nbody\r\n" + ) + (created.path / "manifest.json").write_text("{not json") + err = io.StringIO() + with redirect_stderr(err): + code = cli.main(["contacts", str(created.path)]) + + self.assertEqual(code, cli.EXIT_ERROR) + + if __name__ == "__main__": unittest.main() -- cgit v1.2.3