# Copyright (C) 2026 Danilo M. # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License version 2 as # published by the Free Software Foundation. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. """Dispatch tests for the command line: exit codes and wiring, not prompts.""" import io import pathlib import tempfile import unittest from contextlib import redirect_stderr, redirect_stdout from abusectl import cli FIXTURES = pathlib.Path(__file__).parent / "fixtures" class TestInitNonInteractive(unittest.TestCase): 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 _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_it_writes_a_config_from_flags_alone(self): code, _, _ = self._run( "--config", str(self.config), "init", "--non-interactive", "--trusted-relays", "192.0.2.0/24", ) self.assertEqual(code, 0) self.assertIn("192.0.2.0/24", self.config.read_text()) def test_a_missing_required_flag_fails_rather_than_prompting(self): # An agent cannot answer a prompt, so this must not block. code, _, err = self._run( "--config", str(self.config), "init", "--non-interactive" ) self.assertEqual(code, cli.EXIT_ERROR) self.assertIn("--trusted-relays", err) self.assertFalse(self.config.exists()) def test_a_provider_name_supplies_the_relays(self): code, _, _ = self._run( "--config", str(self.config), "init", "--non-interactive", "--provider", "gmail", ) self.assertEqual(code, 0) self.assertIn("74.125.0.0/16", self.config.read_text()) def test_an_unknown_provider_is_an_error(self): code, _, err = self._run( "--config", str(self.config), "init", "--non-interactive", "--provider", "nosuchprovider", ) self.assertEqual(code, cli.EXIT_ERROR) self.assertIn("nosuchprovider", err) def test_an_existing_config_is_refused_without_force(self): self.config.write_text("[general]\n", encoding="utf-8") code, _, err = self._run( "--config", str(self.config), "init", "--non-interactive", "--trusted-relays", "192.0.2.0/24", ) self.assertEqual(code, cli.EXIT_ERROR) self.assertIn("--force", err) def test_force_overwrites_an_existing_config(self): self.config.write_text('[general]\ntrusted_relays = ["10.0.0.0/8"]\n', encoding="utf-8") code, _, _ = self._run( "--config", str(self.config), "init", "--non-interactive", "--trusted-relays", "192.0.2.0/24", "--force", ) self.assertEqual(code, 0) self.assertIn("192.0.2.0/24", self.config.read_text()) class TestParse(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.root = pathlib.Path(self._tmp.name) self.config = self.root / "config.toml" self.cases = self.root / "cases" self.config.write_text( f'[general]\ntrusted_relays = ["192.0.2.0/24"]\n' f'cases = "{self.cases}"\n', encoding="utf-8", ) def tearDown(self): self._tmp.cleanup() 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_it_creates_a_case_and_prints_its_path(self): code, out, _ = self._run( "--config", str(self.config), "parse", str(FIXTURES / "forged-chain.eml"), ) self.assertEqual(code, 0) created = pathlib.Path(out.strip()) self.assertTrue(created.is_dir()) self.assertTrue((created / "manifest.json").is_file()) self.assertTrue((created / "source.eml").is_file()) def test_the_manifest_holds_the_iocs_and_auth_verdicts(self): import json _, out, _ = self._run( "--config", str(self.config), "parse", str(FIXTURES / "simple.eml"), ) manifest = json.loads( (pathlib.Path(out.strip()) / "manifest.json").read_text() ) values = [i["value"] for i in manifest["iocs"]] self.assertIn("203.0.113.42", values) self.assertEqual(manifest["auth"]["spf"], "fail") def test_no_recipient_address_reaches_the_manifest(self): _, out, _ = self._run( "--config", str(self.config), "parse", str(FIXTURES / "simple.eml"), ) text = (pathlib.Path(out.strip()) / "manifest.json").read_text() self.assertNotIn("you@example.org", text) # The bare domain is still barred everywhere the IOCs live. The # headers block is the one exception and it is a NARROW one: the # whitelist publishes the boundary Received line and # Authentication-Results, and both name our own receiving relay in a # "by"/authserv-id clause. That is the user's mail host, not the # user's identity, and a desk learns it from the report's own From # regardless. The address itself must still be absent, which the # assertion above and report_headers' own tests cover. import json manifest = json.loads(text) headers = manifest.pop("headers") self.assertNotIn("example.org", json.dumps(manifest)) # And nothing shaped like an address survives in the exception. # Both spellings: you%40example.org is not a hypothetical, it is why # leaky.eml exists, and docs/plans/2026-09-09-contacts.md records a # From of phish@victim%40example.org.invalid. blob = json.dumps(headers) self.assertNotIn("@example.org", blob) self.assertNotIn("you%40example.org", blob) names = [name for name, _ in headers] for name in ("To", "Cc", "Delivered-To", "X-Original-To"): self.assertNotIn(name, names) def test_a_missing_config_points_at_init(self): code, _, err = self._run( "--config", str(self.root / "absent.toml"), "parse", str(FIXTURES / "simple.eml"), ) self.assertEqual(code, cli.EXIT_NOT_CONFIGURED) self.assertIn("abusectl init", err) def test_a_missing_message_is_an_error_not_a_traceback(self): code, _, err = self._run( "--config", str(self.config), "parse", str(self.root / "absent.eml"), ) self.assertEqual(code, cli.EXIT_ERROR) 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()