feat: first-run config builder

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KphFXTc2QajxXsHWyvGJ4R

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