diff options
| -rw-r--r-- | abusectl/__main__.py | 19 | ||||
| -rw-r--r-- | abusectl/cli.py | 215 | ||||
| -rw-r--r-- | tests/test_cli.py | 167 |
3 files changed, 401 insertions, 0 deletions
diff --git a/abusectl/__main__.py b/abusectl/__main__.py new file mode 100644 index 0000000..e56d416 --- /dev/null +++ b/abusectl/__main__.py @@ -0,0 +1,19 @@ +# Copyright (C) 2026 Danilo M. <danix@danix.xyz> +# +# 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. +"""`python3 -m abusectl`.""" + +from abusectl.cli import main + +raise SystemExit(main()) diff --git a/abusectl/cli.py b/abusectl/cli.py new file mode 100644 index 0000000..d43cccd --- /dev/null +++ b/abusectl/cli.py @@ -0,0 +1,215 @@ +# Copyright (C) 2026 Danilo M. <danix@danix.xyz> +# +# 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. +"""The command line: dispatch and exit codes only, no logic of its own. + +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. +""" + +import argparse +import sys +from pathlib import Path + +from abusectl import __version__ +from abusectl import case +from abusectl import config +from abusectl import init as init_module +from abusectl import parse as parse_module + +EXIT_OK = 0 +EXIT_ERROR = 1 +EXIT_NOT_CONFIGURED = 3 + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="abusectl") + parser.add_argument("--version", action="version", version=__version__) + parser.add_argument( + "--config", type=Path, default=None, + help="path to config.toml (default: XDG config location)", + ) + + subparsers = parser.add_subparsers(dest="command", required=True) + + init_parser = subparsers.add_parser("init", help="write the first-run config") + init_parser.add_argument("--trusted-relays", nargs="*", default=None) + init_parser.add_argument("--provider", default=None) + init_parser.add_argument("--cases", default=None) + init_parser.add_argument("--from-sample", type=Path, default=None) + init_parser.add_argument("--non-interactive", action="store_true") + init_parser.add_argument("--force", action="store_true") + + parse_parser = subparsers.add_parser("parse", help="parse a message into a case") + parse_parser.add_argument("message", type=Path) + + return parser + + +def _resolve_relays(args) -> list[str]: + """Combine --trusted-relays and --provider into one relay list. + + Union rather than either-or: a user with a known provider plus one + extra relay of their own should not have to pick. Order is preserved + and duplicates removed so the written config is stable and readable. + """ + relays: list[str] = [] + + if args.trusted_relays: + relays.extend(args.trusted_relays) + + if args.provider: + provider_relays = init_module.provider_relays(args.provider) + if not provider_relays: + known = ", ".join(sorted(init_module.PROVIDERS)) + raise ValueError( + f"unknown provider {args.provider!r}; known providers: {known}" + ) + relays.extend(provider_relays) + + if not relays: + raise ValueError( + "no trusted relays given; pass --trusted-relays or --provider" + ) + + seen = set() + deduped = [] + for relay in relays: + if relay not in seen: + seen.add(relay) + deduped.append(relay) + return deduped + + +def _prompt_answers(sample: Path | None) -> dict: + """Ask the interactive questions and return an answers dict for init.build. + + Deliberately simple: this is hand-tested, not unit-tested, so the value + here is asking the right questions, not the exact wording. + """ + trusted_relays: list[str] = [] + + if sample is not None: + raw = sample.read_bytes() + hops = init_module.hops_from_sample(raw) + print("Received-chain hops in the sample, outermost first:") + for i, ip in enumerate(hops, start=1): + print(f" {i}. {ip}") + picks = input("Which are your own infrastructure? (comma-separated numbers): ") + for token in picks.split(","): + token = token.strip() + if not token: + continue + index = int(token) - 1 + trusted_relays.append(f"{hops[index]}/32") + else: + answer = input( + "Trusted relay CIDRs (comma-separated), or leave blank to name a provider: " + ).strip() + if answer: + trusted_relays = [r.strip() for r in answer.split(",") if r.strip()] + else: + provider = input( + f"Provider name ({', '.join(sorted(init_module.PROVIDERS))}): " + ).strip() + trusted_relays = init_module.provider_relays(provider) + if not trusted_relays: + raise ValueError(f"unknown provider {provider!r}") + + cases = input( + f"Cases directory [{config.DEFAULT_CASES}]: " + ).strip() + + return {"trusted_relays": trusted_relays, "cases": cases} + + +def _cmd_init(args) -> int: + target = args.config if args.config is not None else config.path() + + if args.non_interactive: + try: + relays = _resolve_relays(args) + answers = {"trusted_relays": relays, "cases": args.cases or ""} + init_module.write(target, answers, force=args.force) + except (ValueError, FileExistsError) as exc: + print(f"abusectl init: {exc}", file=sys.stderr) + return EXIT_ERROR + print(target) + return EXIT_OK + + force = args.force + if target.exists() and not force: + summary = init_module.existing_summary(target) + print(f"Already configured: {summary}") + print("Sections this run does not set are kept, and the current file is backed up.") + answer = input("Set it up again? [y/N]: ").strip().lower() + if answer not in ("y", "yes"): + print("left unchanged") + return EXIT_OK + force = True + + try: + answers = _prompt_answers(args.from_sample) + init_module.write(target, answers, force=force) + except (ValueError, FileExistsError) as exc: + print(f"abusectl init: {exc}", file=sys.stderr) + return EXIT_ERROR + + print(target) + return EXIT_OK + + +def _cmd_parse(args) -> int: + 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 parse: {exc}", file=sys.stderr) + return EXIT_NOT_CONFIGURED + + try: + raw = args.message.read_bytes() + except OSError as exc: + print(f"abusectl parse: {args.message}: {exc}", file=sys.stderr) + return EXIT_ERROR + + try: + created = case.create(settings.cases, raw) + manifest = case.load(created.path) + manifest["iocs"] = parse_module.iocs(raw, trusted=settings.trusted_relays) + manifest["auth"] = parse_module.auth_results(raw) + case.save(created.path, manifest) + except parse_module.NoTrustBoundary as exc: + print(f"abusectl parse: {exc}", file=sys.stderr) + return EXIT_NOT_CONFIGURED + + print(created.path) + return EXIT_OK + + +def main(argv: list[str] | None = None) -> int: + parser = _build_parser() + args = parser.parse_args(argv) + + if args.command == "init": + return _cmd_init(args) + if args.command == "parse": + return _cmd_parse(args) + + parser.error(f"unknown command {args.command!r}") + return EXIT_ERROR diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..fc4a6e7 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,167 @@ +# Copyright (C) 2026 Danilo M. <danix@danix.xyz> +# +# 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("example.org", text) + + 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) + + +if __name__ == "__main__": + unittest.main() |