diff options
| -rw-r--r-- | abusectl/cli.py | 135 |
1 files changed, 101 insertions, 34 deletions
diff --git a/abusectl/cli.py b/abusectl/cli.py index d43cccd..3872a85 100644 --- a/abusectl/cli.py +++ b/abusectl/cli.py @@ -22,6 +22,7 @@ plainly would collide. """ import argparse +import ipaddress import sys from pathlib import Path @@ -95,45 +96,105 @@ def _resolve_relays(args) -> list[str]: return deduped -def _prompt_answers(sample: Path | None) -> dict: - """Ask the interactive questions and return an answers dict for init.build. +def _ask(question: str) -> str: + """Ask once. Ctrl+C and EOF end the run rather than looping forever.""" + try: + return input(question).strip() + except EOFError: + raise KeyboardInterrupt from None - Deliberately simple: this is hand-tested, not unit-tested, so the value - here is asking the right questions, not the exact wording. - """ - trusted_relays: list[str] = [] - if sample is not None: - raw = sample.read_bytes() - hops = init_module.hops_from_sample(raw) - print("Received-chain hops in the sample, outermost first:") - for i, ip in enumerate(hops, start=1): - print(f" {i}. {ip}") - picks = input("Which are your own infrastructure? (comma-separated numbers): ") +def _pick_hops(hops: list[str]) -> list[str]: + """Ask which Received hops are the user's own, until the answer is usable. + + Re-asks rather than exiting. Getting this wrong is easy, and throwing + away the answers already given to punish a typo is worse than asking + again. + """ + print("\nReceived-chain hops in the sample, outermost first.") + print("The outermost ones are usually yours: your own mail server adds") + print("them last, so they sit at the top of the chain.\n") + for number, ip in enumerate(hops, start=1): + print(f" {number}. {ip}") + + while True: + picks = _ask( + "\nWhich of these are YOUR OWN servers? (comma-separated numbers): " + ) + chosen = [] for token in picks.split(","): token = token.strip() if not token: continue - index = int(token) - 1 - trusted_relays.append(f"{hops[index]}/32") - else: - answer = input( - "Trusted relay CIDRs (comma-separated), or leave blank to name a provider: " - ).strip() - if answer: - trusted_relays = [r.strip() for r in answer.split(",") if r.strip()] + if not token.isdigit() or not 1 <= int(token) <= len(hops): + # Naming the valid range is what makes the retry useful. + print(f" '{token}' is not one of 1 to {len(hops)}.") + chosen = [] + break + chosen.append(f"{hops[int(token) - 1]}/32") else: - provider = input( - f"Provider name ({', '.join(sorted(init_module.PROVIDERS))}): " - ).strip() - trusted_relays = init_module.provider_relays(provider) - if not trusted_relays: - raise ValueError(f"unknown provider {provider!r}") + if chosen: + return chosen + print(" Pick at least one, or Ctrl+C to stop.") + + +def _ask_relays() -> list[str]: + """Ask for the trust boundary, accepting CIDRs or a provider name. + + A provider name typed here rather than at the second prompt used to be + validated as a CIDR, so `Gmail` produced "does not appear to be an IPv4 + or IPv6 network", which names nothing the user can act on. Both answers + are accepted at the same prompt because the user does not know, and + should not have to know, which of the two questions they are answering. + """ + known = ", ".join(sorted(init_module.PROVIDERS)) + print("\nabusectl needs to know which mail servers are YOURS.") + print("Everything below your own servers in a message's Received chain") + print("was written by whoever was talking to them and can be forged, so") + print("this boundary decides which address gets reported for abuse.\n") + + while True: + answer = _ask( + f"Your relays as CIDRs, or a provider name ({known}): " + ) + if not answer: + print(" Needed: a CIDR such as 192.0.2.0/24, or a provider name.") + continue + + # A provider name first: it is the answer that cannot be mistaken for + # anything else, and trying it as a network produces a message about + # IPv4 that means nothing to someone who typed "Gmail". + relays = init_module.provider_relays(answer) + if relays: + print(f" {answer.strip().lower()}: {len(relays)} published ranges.") + return relays + + candidates = [r.strip() for r in answer.split(",") if r.strip()] + try: + for candidate in candidates: + ipaddress.ip_network(candidate, strict=False) + except ValueError as error: + # Validated HERE, at the question that produced it, rather than + # after the next one: erroring later throws away an answer the + # user had already given to a question they got right. + print(f" {error}") + print(f" Expected a CIDR such as 192.0.2.0/24, or one of: {known}") + continue + return candidates + + +def _prompt_answers(sample: Path | None) -> dict: + """Ask the interactive questions and return an answers dict for init.build. - cases = input( - f"Cases directory [{config.DEFAULT_CASES}]: " - ).strip() + Every answer is validated at the prompt that asked for it, so a mistake + costs one retry rather than the whole run. + """ + if sample is not None: + trusted_relays = _pick_hops(init_module.hops_from_sample(sample.read_bytes())) + else: + trusted_relays = _ask_relays() + cases = _ask(f"\nCases directory [{config.DEFAULT_CASES}]: ") return {"trusted_relays": trusted_relays, "cases": cases} @@ -206,10 +267,16 @@ def main(argv: list[str] | None = None) -> int: parser = _build_parser() args = parser.parse_args(argv) - if args.command == "init": - return _cmd_init(args) - if args.command == "parse": - return _cmd_parse(args) + try: + if args.command == "init": + return _cmd_init(args) + if args.command == "parse": + return _cmd_parse(args) + except KeyboardInterrupt: + # Ctrl+C, or EOF at a prompt. Setup writes nothing until every + # answer is in hand, so abandoning it leaves no half-written config. + print("\ncancelled, nothing written", file=sys.stderr) + return EXIT_ERROR parser.error(f"unknown command {args.command!r}") return EXIT_ERROR |