diff options
| -rw-r--r-- | abusectl/init.py | 295 | ||||
| -rw-r--r-- | tests/test_init.py | 185 |
2 files changed, 480 insertions, 0 deletions
diff --git a/abusectl/init.py b/abusectl/init.py new file mode 100644 index 0000000..d10f6ed --- /dev/null +++ b/abusectl/init.py @@ -0,0 +1,295 @@ +# 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. +"""Build and write the first-run config. + +The parser in config.py refuses to guess which mail relays belong to the +user, because guessing wrong means reporting an innocent third party for +abuse they did not commit. Refusing is only defensible if there is a route +out, and this module is that route: a pure builder that turns a mapping of +answers into config TOML, plus the write path that puts it on disk safely. + +The interactive prompts and command-line flags are a thin shell added on top +of this in a later task. Nothing here reads from a terminal or argv, which is +what lets the config-writing logic be tested with no terminal at all. +""" + +import ipaddress +import os +import pathlib +import time +import tomllib + +from abusectl import parse + +# Sending ranges the providers publish in their own SPF records, transcribed +# on 2026-09-08 from: +# +# gmail dig TXT _spf.google.com +# fastmail dig TXT spf.messagingengine.com +# proton dig TXT _spf.protonmail.ch + _spf2.protonmail.ch +# outlook dig TXT spf.protection.outlook.com +# zoho dig TXT spf.zoho.eu +# privateemail spf.privateemail.com is INCLUDES rather than addresses: +# ips1 + ips2 + fbrelay + spf-pe.jellyfish.systems +# + spf-ep-nc.jellyfish.systems, flattened here +# +# Static rather than read from SPF at runtime: SPF is a DNS lookup, and while +# the never-resolve rule is about parsing hostile mail rather than setup, a +# static table keeps the boundary unambiguous. +# +# IPv4 only. An IPv6 hop from one of these providers is simply not matched by +# the table, which is the safe direction: the user is asked instead of a hop +# being wrongly trusted. +# +# THESE GO STALE. A range that has been reassigned means a hop is treated as +# the user's own and the real sender is never reported, so re-check the SPF +# records before a release rather than trusting the date above. +PROVIDERS: dict[str, list[str]] = { + "gmail": [ + "74.125.0.0/16", + "209.85.128.0/17", + ], + "fastmail": [ + "103.168.172.128/27", + "202.12.124.128/27", + "204.75.18.128/27", + ], + "proton": [ + "185.70.40.0/24", + "185.70.41.0/24", + "185.70.43.0/24", + "79.135.106.0/24", + "79.135.107.0/24", + "109.224.244.0/24", + "85.9.206.169/32", + "85.9.210.45/32", + "37.187.220.204/32", + "51.83.17.38/32", + "57.129.93.249/32", + ], + "outlook": [ + "40.92.0.0/15", + "40.107.0.0/16", + "52.100.0.0/15", + "52.102.0.0/16", + "52.103.0.0/17", + "104.47.0.0/17", + ], + "privateemail": [ + "63.250.43.64/26", + "66.29.159.48/28", + "66.29.159.80/28", + "104.207.68.0/24", + "162.0.218.228/32", + "162.0.218.229/32", + "162.0.218.230/32", + "162.0.218.231/32", + "198.54.118.192/27", + "198.54.122.64/27", + "198.54.122.96/27", + "198.54.122.128/27", + "198.54.127.32/27", + "198.54.127.64/27", + "198.54.127.96/27", + "198.54.127.128/26", + "198.177.127.176/28", + "198.177.127.192/27", + ], + "zoho": [ + "185.20.209.0/24", + "31.186.226.0/24", + "31.186.243.0/24", + "89.36.170.0/24", + "185.20.211.0/24", + "185.172.199.0/24", + "91.135.68.104/29", + "185.230.214.0/23", + "136.143.168.0/22", + "34.241.242.183/32", + ], +} + + +def provider_relays(name: str) -> list[str]: + """Return the shipped ranges for a known provider, or [] if unknown.""" + return PROVIDERS.get(name.strip().lower(), []) + + +def hops_from_sample(raw: bytes) -> list[str]: + """Return the Received-chain IPs of a known-good sample, outermost first. + + Setup shows this list and asks which hops are the user's own, turning an + abstract question ("which relays do you own?") into picking from a real + chain the user can recognise. + """ + return [hop.ip for hop in parse.received_hops(raw)] + + +def _validate_relays(trusted_relays) -> list[str]: + relays = [r for r in trusted_relays if isinstance(r, str) and r.strip()] + if not relays: + raise ValueError("at least one trusted relay is required") + for relay in relays: + # A relay must be a str: an int silently parses as a packed address, + # e.g. ip_network(42) == "0.0.0.42/32", which would trust the wrong + # thing without ever raising. + ipaddress.ip_network(relay, strict=False) + return relays + + +def _quoted(value: str, field: str) -> str: + # This is a config file the user hand-edits, not a wire format, so a + # value needing escaping is a value they should not be setting: reject + # rather than try to escape a quote or newline out of it. + if '"' in value or "\n" in value or "\r" in value: + raise ValueError(f"{field} must not contain a quote or newline") + return f'"{value}"' + + +def build(answers: dict) -> str: + """Render answers as config TOML for the [general] table.""" + relays = _validate_relays(answers.get("trusted_relays", [])) + + lines = [ + "# abusectl configuration.", + "#", + "# trusted_relays names the mail infrastructure that is yours: the", + "# Received chain is walked outermost-first and the first hop NOT in", + "# this list is treated as the reportable sender. Guessing this wrong", + "# means reporting an innocent third party, so it is never inferred.", + "", + "[general]", + "trusted_relays = [", + ] + lines.extend(f' "{relay}",' for relay in relays) + lines.append("]") + + cases = answers.get("cases", "") + if isinstance(cases, str) and cases.strip(): + lines.append(f"cases = {_quoted(cases, 'cases')}") + + lines.extend([ + "", + "# Later parts of abusectl add further sections here:", + "# [misp] - MISP instance URL and API key", + "# [vendors] - per-vendor threat-intel API keys", + "# [reporting] - abuse-desk reporting defaults", + ]) + + return "\n".join(lines) + "\n" + + +def existing_summary(path: pathlib.Path) -> str: + """Describe an existing config by section/key NAMES only. + + Never values: the file holds API keys, and echoing a secret to the + terminal to ask about overwriting it is a poor trade. + """ + try: + with open(path, "rb") as f: + data = tomllib.load(f) + except (OSError, tomllib.TOMLDecodeError): + return "unreadable" + + if not data: + return "empty" + + parts = [] + for section in sorted(data): + values = data[section] + if isinstance(values, dict): + keys = ", ".join(sorted(values)) + else: + keys = "" + parts.append(f"[{section}]: {keys}") + return "; ".join(parts) + + +def back_up(path: pathlib.Path) -> pathlib.Path | None: + """Copy an existing config aside with a timestamped .bak name. + + Returns the backup path, or None if there was nothing to copy. Created + with mode 0o600 from the start: it holds the same secrets the config + does, so it must never be world-readable even for an instant. + """ + if not path.exists(): + return None + + stamp = time.strftime("%Y%m%d-%H%M%S") + backup_path = path.with_name(f"{path.name}.bak-{stamp}") + data = path.read_bytes() + fd = os.open(backup_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + try: + os.write(fd, data) + finally: + os.close(fd) + return backup_path + + +def _preserved_sections(path: pathlib.Path) -> str: + """Return the raw text of every top-level table except [general]. + + Carrying raw lines rather than re-rendering means a comment, or a field + this build does not understand, also survives an init that only sets + the relays. + """ + if not path.exists(): + return "" + + text = path.read_text(encoding="utf-8") + lines = text.splitlines(keepends=True) + + kept = [] + skipping = False + for line in lines: + stripped = line.strip() + if stripped.startswith("[") and stripped.endswith("]"): + skipping = stripped.lstrip("[").rstrip("]").strip() == "general" + if skipping: + continue + if not skipping: + kept.append(line) + + result = "".join(kept).strip("\n") + return result + + +def write(path: pathlib.Path, answers: dict, force: bool = False) -> pathlib.Path: + """Write a config built from answers to path. + + Refuses an existing file unless force=True: the refusal is what makes an + interactive confirmation meaningful, since the caller has to have + decided rather than the write silently clobbering something. Any + section other than [general] in an existing file is preserved verbatim, + and the existing file is backed up before being replaced. + """ + if path.exists() and not force: + raise FileExistsError(f"{path} already exists; pass --force to overwrite") + + preserved = _preserved_sections(path) + text = build(answers) + if preserved: + text = text.rstrip("\n") + "\n\n" + preserved + "\n" + + back_up(path) + + path.parent.mkdir(parents=True, exist_ok=True) + fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + try: + os.write(fd, text.encode("utf-8")) + finally: + os.close(fd) + + return path diff --git a/tests/test_init.py b/tests/test_init.py new file mode 100644 index 0000000..619d6ba --- /dev/null +++ b/tests/test_init.py @@ -0,0 +1,185 @@ +# 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. +"""Tests for the first-run config builder in abusectl.init.""" + +import ipaddress +import pathlib +import tempfile +import tomllib +import unittest + +from abusectl import config, init + + +class TestBuildConfig(unittest.TestCase): + def test_the_answers_become_readable_toml(self): + text = init.build({"trusted_relays": ["192.0.2.0/24"], "cases": "~/c"}) + parsed = tomllib.loads(text) + self.assertEqual(parsed["general"]["trusted_relays"], ["192.0.2.0/24"]) + self.assertEqual(parsed["general"]["cases"], "~/c") + + def test_a_skipped_answer_is_absent_not_empty(self): + # An empty string reads as configured-and-broken later on. + text = init.build({"trusted_relays": ["192.0.2.0/24"], "cases": ""}) + self.assertNotIn("cases", tomllib.loads(text)["general"]) + + def test_a_malformed_relay_is_rejected(self): + with self.assertRaises(ValueError): + init.build({"trusted_relays": ["nonsense"]}) + + def test_no_relays_at_all_is_rejected(self): + with self.assertRaises(ValueError): + init.build({"trusted_relays": []}) + + def test_the_result_loads_back_through_config(self): + with tempfile.TemporaryDirectory() as tmp: + path = pathlib.Path(tmp) / "config.toml" + path.write_text( + init.build({"trusted_relays": ["192.0.2.0/24"]}), encoding="utf-8" + ) + self.assertEqual(config.load(path).trusted_relays, ["192.0.2.0/24"]) + + def test_a_quote_in_an_answer_cannot_break_out_of_the_toml(self): + # The cases path reaches the file as a quoted string, so a value + # containing a quote must not be able to close it early. + with self.assertRaises(ValueError): + init.build({"trusted_relays": ["192.0.2.0/24"], + "cases": 'x" \ntrusted_relays = ["0.0.0.0/0"]\n#'}) + + +class TestProviderTable(unittest.TestCase): + def test_a_known_provider_resolves_to_ranges(self): + self.assertTrue(init.provider_relays("gmail")) + + def test_lookup_is_case_insensitive(self): + self.assertEqual(init.provider_relays("Gmail"), init.provider_relays("gmail")) + + def test_an_unknown_provider_returns_nothing(self): + self.assertEqual(init.provider_relays("nosuchprovider"), []) + + def test_every_shipped_range_is_a_valid_network(self): + for name, ranges in init.PROVIDERS.items(): + self.assertTrue(ranges, f"{name} has no ranges") + for entry in ranges: + ipaddress.ip_network(entry, strict=False) + + def test_a_provider_result_survives_the_builder(self): + # The table feeds the builder directly, so its entries must satisfy + # the same validation a typed answer does. + text = init.build({"trusted_relays": init.provider_relays("gmail")}) + self.assertIn("74.125.0.0/16", text) + + +class TestSampleChain(unittest.TestCase): + def test_hops_are_offered_for_picking(self): + raw = ( + pathlib.Path(__file__).parent / "fixtures" / "simple.eml" + ).read_bytes() + hops = init.hops_from_sample(raw) + self.assertEqual(hops, ["192.0.2.11", "203.0.113.42"]) + + +class TestWriteGuard(unittest.TestCase): + def test_writing_over_an_existing_config_refuses_without_force(self): + with tempfile.TemporaryDirectory() as tmp: + path = pathlib.Path(tmp) / "config.toml" + path.write_text("[general]\n", encoding="utf-8") + with self.assertRaises(FileExistsError): + init.write(path, {"trusted_relays": ["192.0.2.0/24"]}) + + def test_force_overwrites_and_leaves_a_backup(self): + with tempfile.TemporaryDirectory() as tmp: + path = pathlib.Path(tmp) / "config.toml" + path.write_text('[general]\ntrusted_relays = ["10.0.0.0/8"]\n', + encoding="utf-8") + init.write(path, {"trusted_relays": ["192.0.2.0/24"]}, force=True) + + self.assertIn("192.0.2.0/24", path.read_text()) + backups = list(pathlib.Path(tmp).glob("config.toml.bak-*")) + self.assertEqual(len(backups), 1) + self.assertIn("10.0.0.0/8", backups[0].read_text()) + + def test_a_backup_is_not_world_readable_either(self): + # It holds the same secrets the config does. + with tempfile.TemporaryDirectory() as tmp: + path = pathlib.Path(tmp) / "config.toml" + path.write_text("[general]\n", encoding="utf-8") + init.write(path, {"trusted_relays": ["192.0.2.0/24"]}, force=True) + backup = next(pathlib.Path(tmp).glob("config.toml.bak-*")) + self.assertEqual(backup.stat().st_mode & 0o077, 0) + + def test_an_unsupplied_section_survives_a_rewrite(self): + # Once the config holds a MISP key, an init that only sets the relays + # must not silently discard it. The backup makes that recoverable; + # not losing it is better. + with tempfile.TemporaryDirectory() as tmp: + path = pathlib.Path(tmp) / "config.toml" + path.write_text( + '[general]\ntrusted_relays = ["10.0.0.0/8"]\n\n' + '[misp]\nurl = "https://misp.example.invalid"\n' + 'api_key = "kept"\n', + encoding="utf-8", + ) + init.write(path, {"trusted_relays": ["192.0.2.0/24"]}, force=True) + + rewritten = path.read_text() + self.assertIn("192.0.2.0/24", rewritten) + self.assertIn("[misp]", rewritten) + self.assertIn("kept", rewritten) + + def test_a_rewrite_still_loads(self): + # Preserving sections verbatim must not produce a file the reader + # then chokes on. + with tempfile.TemporaryDirectory() as tmp: + path = pathlib.Path(tmp) / "config.toml" + path.write_text( + '[general]\ntrusted_relays = ["10.0.0.0/8"]\n\n' + '[misp]\napi_key = "kept"\n', + encoding="utf-8", + ) + init.write(path, {"trusted_relays": ["192.0.2.0/24"]}, force=True) + self.assertEqual(config.load(path).trusted_relays, ["192.0.2.0/24"]) + + def test_a_written_config_is_not_world_readable(self): + # It will hold API keys as later parts land. + with tempfile.TemporaryDirectory() as tmp: + path = pathlib.Path(tmp) / "config.toml" + init.write(path, {"trusted_relays": ["192.0.2.0/24"]}) + self.assertEqual(path.stat().st_mode & 0o077, 0) + + def test_a_first_write_creates_no_backup(self): + with tempfile.TemporaryDirectory() as tmp: + path = pathlib.Path(tmp) / "config.toml" + init.write(path, {"trusted_relays": ["192.0.2.0/24"]}) + self.assertEqual(list(pathlib.Path(tmp).glob("*.bak-*")), []) + + def test_existing_summary_names_sections_without_showing_values(self): + # The file holds API keys. Echoing a secret to the terminal to ask + # about overwriting it is a poor trade. + with tempfile.TemporaryDirectory() as tmp: + path = pathlib.Path(tmp) / "config.toml" + path.write_text( + '[general]\ntrusted_relays = ["10.0.0.0/8"]\n\n' + '[misp]\napi_key = "topsecret"\n', + encoding="utf-8", + ) + summary = init.existing_summary(path) + self.assertIn("misp", summary) + self.assertIn("api_key", summary) + self.assertNotIn("topsecret", summary) + + +if __name__ == "__main__": + unittest.main() |