path: root/tests/test_cli.py

# 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("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)


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()