diff options
| -rw-r--r-- | abusectl/cli.py | 59 | ||||
| -rw-r--r-- | abusectl/init.py | 21 | ||||
| -rw-r--r-- | tests/test_init.py | 139 |
3 files changed, 210 insertions, 9 deletions
diff --git a/abusectl/cli.py b/abusectl/cli.py index 76ed096..bdde4b0 100644 --- a/abusectl/cli.py +++ b/abusectl/cli.py @@ -26,6 +26,7 @@ alike. import argparse import ipaddress import sys +import tomllib from pathlib import Path from abusectl import __version__ @@ -337,11 +338,55 @@ def _ask_destinations() -> dict: return answers -def _prompt_answers(sample: Path | None) -> dict: - """Ask the interactive questions and return an answers dict for init.build. +def _confirm_dropped(target: Path, answers: dict) -> frozenset: + """Ask whether to keep each configured section the user just skipped. + + Skipping a question reads as "do not use this", but sections this run + does not render are preserved verbatim, which is what stops an init + setting only the relays from discarding a key set earlier. The two + rules collide exactly here: declining MISP at the prompt used to leave + the previous instance in the file, and `report` would then submit to an + instance the user had just said no to. + + Asked rather than assumed, because both answers are defensible: the + user may be re-running init for the relays alone and want the key kept. + Only sections that BOTH exist already and were skipped are asked about, + so a first run and a fully answered run ask nothing. + """ + if not target.exists(): + return frozenset() + + try: + with open(target, "rb") as handle: + existing = tomllib.load(handle) + except (OSError, tomllib.TOMLDecodeError): + # An unreadable file is not one to make removal decisions from. + # write() backs it up regardless, so nothing is lost either way. + return frozenset() + + dropped = [] + for section, keys in init_module.DESTINATION_ANSWER_KEYS.items(): + if section not in existing: + continue + if any(answers.get(key, "").strip() for key in keys): + continue + print(f"\n You skipped {section}, but the current config has a" + f" [{section}] section.") + keep = _ask(f" Keep the existing [{section}]? [Y/n]: ").lower() + if keep in ("n", "no"): + dropped.append(section) + return frozenset(dropped) + + +def _prompt_answers(sample: Path | None, + target: Path | None = None) -> tuple[dict, frozenset]: + """Ask the interactive questions, returning answers and sections to drop. Every answer is validated at the prompt that asked for it, so a mistake costs one retry rather than the whole run. + + The second return value names sections the user was asked about and + skipped, which write() must not preserve. See _confirm_dropped. """ if sample is not None: trusted_relays = _pick_hops(init_module.hops_from_sample(sample.read_bytes())) @@ -349,8 +394,10 @@ def _prompt_answers(sample: Path | None) -> dict: trusted_relays = _ask_relays() cases = _ask(f"\nCases directory (Enter for {config.DEFAULT_CASES}): ") - return {"trusted_relays": trusted_relays, "cases": cases, - **_ask_reporter(), **_ask_destinations()} + answers = {"trusted_relays": trusted_relays, "cases": cases, + **_ask_reporter(), **_ask_destinations()} + drop = _confirm_dropped(target, answers) if target is not None else frozenset() + return answers, drop def _cmd_init(args) -> int: @@ -379,8 +426,8 @@ def _cmd_init(args) -> int: force = True try: - answers = _prompt_answers(args.from_sample) - init_module.write(target, answers, force=force) + answers, drop = _prompt_answers(args.from_sample, target) + init_module.write(target, answers, force=force, drop=drop) except (ValueError, FileExistsError) as exc: print(f"abusectl init: {exc}", file=sys.stderr) return EXIT_ERROR diff --git a/abusectl/init.py b/abusectl/init.py index dd6ef03..c18e3a1 100644 --- a/abusectl/init.py +++ b/abusectl/init.py @@ -186,6 +186,16 @@ _DESTINATION_FIELDS = ( ) +# Which answer keys feed each destination section. Derived from +# _DESTINATION_FIELDS rather than written out again, so a section added to +# the builder cannot be forgotten by the prompt that asks whether to keep a +# skipped one. +DESTINATION_ANSWER_KEYS = { + section: tuple(answer_key for _, answer_key in fields) + for section, fields in _DESTINATION_FIELDS +} + + def build(answers: dict) -> str: """Render answers as config TOML for the [general] and [reporter] tables.""" relays = _validate_relays(answers.get("trusted_relays", [])) @@ -371,7 +381,8 @@ def _preserved_sections(path: pathlib.Path, built: set) -> str: return result -def write(path: pathlib.Path, answers: dict, force: bool = False) -> pathlib.Path: +def write(path: pathlib.Path, answers: dict, force: bool = False, + drop: frozenset = frozenset()) -> pathlib.Path: """Write a config built from answers to path. Refuses an existing file unless force=True: the refusal is what makes an @@ -387,7 +398,13 @@ def write(path: pathlib.Path, answers: dict, force: bool = False) -> pathlib.Pat # Only what this run actually rendered is dropped from the old file, so # an init that skips every reporter question keeps the identity the user # set by hand instead of silently deleting it. - preserved = _preserved_sections(path, _rendered_sections(text)) + # `drop` names sections the user was ASKED about and skipped. Without + # it a skip preserves whatever was there, so declining MISP at the + # prompt silently re-enabled the instance set on a previous run, and + # MISP is the gate for everything irreversible. An unasked section is + # still preserved: that is what stops an init setting only the relays + # from discarding a key set earlier. + preserved = _preserved_sections(path, _rendered_sections(text) | drop) if preserved: body = text[: -len(FOOTER)] if text.endswith(FOOTER) else text text = body.rstrip("\n") + "\n\n" + preserved + "\n" + FOOTER diff --git a/tests/test_init.py b/tests/test_init.py index 798ed26..2e2ebd5 100644 --- a/tests/test_init.py +++ b/tests/test_init.py @@ -14,13 +14,15 @@ # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. """Tests for the first-run config builder in abusectl.init.""" +import contextlib +import io import ipaddress import pathlib import tempfile import tomllib import unittest -from abusectl import config, init +from abusectl import cli, config, init class TestBuildConfig(unittest.TestCase): @@ -363,6 +365,141 @@ class BuildsDestinationSections(unittest.TestCase): {"misp", "abusedb", "urlhaus"}) +class DropsSkippedSections(unittest.TestCase): + """A section the user was ASKED about and skipped must not survive. + + Sections this run does not render are preserved verbatim, which is what + stops an init setting only the relays from discarding a key set + earlier. That rule and a deliberate skip collide: declining MISP at the + prompt left the previous instance in the file, and report would then + submit to an instance the user had just said no to. `drop` is how write + tells the two apart. + """ + + ANSWERS = {"trusted_relays": ["192.0.2.0/24"]} + + def _existing(self, tmp): + path = pathlib.Path(tmp) / "config.toml" + init.write(path, {**self.ANSWERS, + "misp_url": "https://old.example.invalid", + "misp_api_key": "old", + "abusedb_api_key": "olda"}) + return path + + def test_a_dropped_section_is_removed(self): + with tempfile.TemporaryDirectory() as tmp: + path = self._existing(tmp) + init.write(path, self.ANSWERS, force=True, drop=frozenset({"misp"})) + data = tomllib.loads(path.read_text()) + self.assertNotIn("misp", data) + + def test_dropping_one_leaves_the_others(self): + with tempfile.TemporaryDirectory() as tmp: + path = self._existing(tmp) + init.write(path, self.ANSWERS, force=True, drop=frozenset({"misp"})) + data = tomllib.loads(path.read_text()) + self.assertEqual(data["abusedb"]["api_key"], "olda") + + def test_without_drop_a_section_is_still_preserved(self): + # The rule drop exists to qualify, not to replace. A non-interactive + # caller passes no drop and must behave exactly as before. + with tempfile.TemporaryDirectory() as tmp: + path = self._existing(tmp) + init.write(path, self.ANSWERS, force=True) + data = tomllib.loads(path.read_text()) + self.assertEqual(data["misp"]["api_key"], "old") + + def test_a_dropped_section_that_is_answered_is_still_written(self): + # drop names what was skipped. An answered section renders normally + # and must not be removed by a stale drop entry. + with tempfile.TemporaryDirectory() as tmp: + path = self._existing(tmp) + init.write(path, {**self.ANSWERS, + "misp_url": "https://new.example.invalid", + "misp_api_key": "new"}, + force=True, drop=frozenset({"misp"})) + data = tomllib.loads(path.read_text()) + self.assertEqual(data["misp"]["url"], "https://new.example.invalid") + + def test_the_result_still_parses_and_loads(self): + with tempfile.TemporaryDirectory() as tmp: + path = self._existing(tmp) + init.write(path, self.ANSWERS, force=True, + drop=frozenset({"misp", "abusedb"})) + self.assertEqual(config.load(path).destinations, set()) + + +class AsksBeforeKeepingASkippedSection(unittest.TestCase): + """The prompt half, driven directly rather than through stdin.""" + + SKIPPED = {"misp_url": "", "misp_api_key": "", "abusedb_api_key": ""} + + def _answer(self, replies): + replies = iter(replies) + cli._ask = lambda question: next(replies) + + def setUp(self): + self._real_ask = cli._ask + + def tearDown(self): + cli._ask = self._real_ask + + def test_declining_drops_the_section(self): + with tempfile.TemporaryDirectory() as tmp: + path = pathlib.Path(tmp) / "config.toml" + init.write(path, {"trusted_relays": ["192.0.2.0/24"], + "misp_url": "https://old.example.invalid", + "misp_api_key": "old"}) + self._answer(["n"]) + with contextlib.redirect_stdout(io.StringIO()): + dropped = cli._confirm_dropped(path, self.SKIPPED) + self.assertEqual(dropped, frozenset({"misp"})) + + def test_the_default_keeps_it(self): + # Enter means keep: removal is the destructive answer and must be + # typed, not fallen into. + with tempfile.TemporaryDirectory() as tmp: + path = pathlib.Path(tmp) / "config.toml" + init.write(path, {"trusted_relays": ["192.0.2.0/24"], + "misp_url": "https://old.example.invalid", + "misp_api_key": "old"}) + self._answer([""]) + with contextlib.redirect_stdout(io.StringIO()): + dropped = cli._confirm_dropped(path, self.SKIPPED) + self.assertEqual(dropped, frozenset()) + + def test_an_answered_section_is_never_asked_about(self): + with tempfile.TemporaryDirectory() as tmp: + path = pathlib.Path(tmp) / "config.toml" + init.write(path, {"trusted_relays": ["192.0.2.0/24"], + "misp_url": "https://old.example.invalid", + "misp_api_key": "old"}) + # No replies queued: asking anything raises StopIteration. + self._answer([]) + with contextlib.redirect_stdout(io.StringIO()): + dropped = cli._confirm_dropped( + path, {"misp_url": "https://new.example.invalid", + "misp_api_key": "new"}) + self.assertEqual(dropped, frozenset()) + + def test_a_first_run_asks_nothing(self): + with tempfile.TemporaryDirectory() as tmp: + self._answer([]) + dropped = cli._confirm_dropped( + pathlib.Path(tmp) / "absent.toml", self.SKIPPED) + self.assertEqual(dropped, frozenset()) + + def test_an_unreadable_config_asks_nothing(self): + # Not a file to make removal decisions from, and write() backs it + # up regardless. + with tempfile.TemporaryDirectory() as tmp: + path = pathlib.Path(tmp) / "config.toml" + path.write_text("not [valid toml", encoding="utf-8") + self._answer([]) + dropped = cli._confirm_dropped(path, self.SKIPPED) + self.assertEqual(dropped, frozenset()) + + if __name__ == "__main__": unittest.main() |