diff options
| author | Danilo M. <danix@danix.xyz> | 2026-09-10 16:45:54 +0200 |
|---|---|---|
| committer | Danilo M. <danix@danix.xyz> | 2026-09-10 16:45:54 +0200 |
| commit | 88d07d352e84b0c32012c6989e8402980e0a3b94 (patch) | |
| tree | b45cdf88fdc6d65ed06f53725dfc3753f0ad27e8 | |
| parent | 44ebca17e307a6c86308f59789f52f98c683e1c2 (diff) | |
| download | abusectl-88d07d352e84b0c32012c6989e8402980e0a3b94.tar.gz abusectl-88d07d352e84b0c32012c6989e8402980e0a3b94.zip | |
feat: read which reporting destinations are configured
Adds DESTINATION_KEYS and two Config fields, destinations and
incomplete, so config.load() reports which of misp/abusedb/urlhaus
have every required key filled with a non-empty string. A half-filled
MISP pair is reported as data via incomplete, never raised: parse and
contacts load this same file and read neither key, and report (task 6)
is the one that will refuse it.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0176FYdVfpzUq8S9jecqQqL6
| -rw-r--r-- | abusectl/config.py | 57 | ||||
| -rw-r--r-- | tests/test_config.py | 58 |
2 files changed, 114 insertions, 1 deletions
diff --git a/abusectl/config.py b/abusectl/config.py index 26c45f1..15b0abe 100644 --- a/abusectl/config.py +++ b/abusectl/config.py @@ -68,6 +68,18 @@ def path() -> pathlib.Path: # third party by way of a config the user hand-edited. REPORTER_KEYS = ("name", "org", "email") +# What each reporting destination's section must carry to count as +# configured. MISP takes a url because it is self-hosted and has no default +# endpoint; the vendors' endpoints are fixed and belong to their own +# destination modules. Everything ELSE inside these tables is carried +# through untouched: submit will want timeouts and category codes, and this +# reader deliberately does not design that. +DESTINATION_KEYS = { + "misp": ("url", "api_key"), + "abusedb": ("api_key",), + "urlhaus": ("api_key",), +} + @dataclass(frozen=True) class Config: @@ -78,6 +90,11 @@ class Config: # only reads it, and a defensive copy here would buy nothing while # suggesting a guarantee this class does not make. reporter: dict = field(default_factory=dict) + # Which destinations are fully configured, and which are half-filled. + # Reported as data rather than raised: every subcommand loads this file + # and most of them read neither key. + destinations: set = field(default_factory=set) + incomplete: dict = field(default_factory=dict) def load(from_path: pathlib.Path | None = None) -> Config: @@ -153,4 +170,42 @@ def load(from_path: pathlib.Path | None = None) -> Config: if value.strip(): reporter[key] = value.strip() - return Config(trusted_relays=trusted_relays, cases=cases, reporter=reporter) + destinations = set() + incomplete = {} + for name, required in DESTINATION_KEYS.items(): + section = data.get(name) + if section is None: + continue + if not isinstance(section, dict): + raise ValueError( + f"{name} in {source} must be a table, got " + f"{type(section).__name__}" + ) + + present = [] + for key in required: + if key not in section: + continue + value = section[key] + # Rejected, not dropped, for the reason [reporter] records: a + # wrong value that reads as plausible is worse than an error. + if not isinstance(value, str): + raise ValueError( + f"{name}.{key} in {source} must be a string, got " + f"{type(value).__name__}: {value!r}" + ) + if value.strip(): + present.append(key) + + if len(present) == len(required): + destinations.add(name) + elif present: + # Some but not all: configured-and-broken rather than skipped. + # Absence of the WHOLE section is a skip; absence of one half of + # a pair is a mistake, and the command that acts on it says so. + missing = [key for key in required if key not in present] + incomplete[name] = missing[0] + + return Config(trusted_relays=trusted_relays, cases=cases, + reporter=reporter, destinations=destinations, + incomplete=incomplete) diff --git a/tests/test_config.py b/tests/test_config.py index 1815993..f62960b 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -281,5 +281,63 @@ class TestConfig(unittest.TestCase): ) +class ConfiguredDestinations(unittest.TestCase): + def _load(self, text): + with tempfile.TemporaryDirectory() as tmp: + path = pathlib.Path(tmp) / "config.toml" + general = '[general]\ntrusted_relays = ["192.0.2.0/24"]\n' + # A bare top-level assignment (no leading "[") is only valid TOML + # before the first table header, so it must precede [general] + # rather than follow it. + combined = text + general if text and not text.startswith("[") \ + else general + text + path.write_text(combined) + return config.load(path) + + def test_no_sections_means_nothing_configured(self): + self.assertEqual(self._load("").destinations, set()) + + def test_a_vendor_with_a_key_is_configured(self): + settings = self._load('[abusedb]\napi_key = "k"\n') + self.assertEqual(settings.destinations, {"abusedb"}) + + def test_a_vendor_section_with_no_key_is_not_configured(self): + # Absent reads as not-configured, which is the whole convention. + self.assertEqual(self._load("[urlhaus]\n").destinations, set()) + + def test_an_empty_key_is_not_configured(self): + # api_key = "" is the trap this file exists to refuse: it reads as + # configured-and-broken and produces an auth error much later. + self.assertEqual(self._load('[urlhaus]\napi_key = ""\n').destinations, + set()) + + def test_misp_needs_both_halves(self): + both = '[misp]\nurl = "https://misp.example.invalid"\napi_key = "k"\n' + self.assertEqual(self._load(both).destinations, {"misp"}) + + def test_half_a_misp_pair_is_reported_separately_not_configured(self): + # Not an exception here: parse and contacts load this same file and + # read neither key. report refuses instead, with a sentence. + settings = self._load('[misp]\nurl = "https://misp.example.invalid"\n') + self.assertEqual(settings.destinations, set()) + self.assertEqual(settings.incomplete, {"misp": "api_key"}) + + def test_the_other_half_is_reported_too(self): + settings = self._load('[misp]\napi_key = "k"\n') + self.assertEqual(settings.destinations, set()) + self.assertEqual(settings.incomplete, {"misp": "url"}) + + def test_a_non_string_key_is_rejected_naming_the_file(self): + with self.assertRaises(ValueError) as caught: + self._load("[abusedb]\napi_key = 42\n") + self.assertIn("abusedb.api_key", str(caught.exception)) + self.assertIn("config.toml", str(caught.exception)) + + def test_a_non_table_section_is_rejected(self): + with self.assertRaises(ValueError) as caught: + self._load('abusedb = "nope"\n') + self.assertIn("abusedb", str(caught.exception)) + + if __name__ == "__main__": unittest.main() |
