aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorDanilo M. <danix@danix.xyz>2026-09-10 17:26:17 +0200
committerDanilo M. <danix@danix.xyz>2026-09-10 17:26:17 +0200
commita6ead5cb8904cd46d3f3c6f6e4c4b3bc9bc2bb0b (patch)
treedab9115f1c4ee9113690094a459c4e4d668276ad
parentfa9d51612259c484448815162bba7a199bca60c7 (diff)
downloadabusectl-a6ead5cb8904cd46d3f3c6f6e4c4b3bc9bc2bb0b.tar.gz
abusectl-a6ead5cb8904cd46d3f3c6f6e4c4b3bc9bc2bb0b.zip
fix: ask before keeping a destination the user skippedHEADmaster
Skipping a destination at init left the previous one in the file. Sections a run does not render are preserved verbatim, which is what stops an init setting only the relays from discarding a key set earlier, but a skip is a deliberate "do not use this" and the two rules collided. Declining MISP re-enabled the instance from an earlier run, and MISP is the gate for everything irreversible, so report would have submitted to an instance the user had just said no to. Both vendors behaved the same way. Found by hand-testing the prompts, which is why they are hand-tested: no unit test covers a sequence of runs, and the suite was green throughout. write() gains `drop`, the sections the user was ASKED about and skipped, which join the set _preserved_sections already drops. An unasked section is still preserved, so the non-interactive path and the existing carry-across tests are unchanged. The prompt only fires when a section BOTH exists and was skipped, and Enter keeps it: removal is the destructive answer and has to be typed. DESTINATION_ANSWER_KEYS is derived from _DESTINATION_FIELDS rather than written out again, so a destination added later cannot be forgotten here. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0176FYdVfpzUq8S9jecqQqL6
-rw-r--r--abusectl/cli.py59
-rw-r--r--abusectl/init.py21
-rw-r--r--tests/test_init.py139
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()