fix: refuse an ipv6 scope id, and say why a value was refused

Four holes of one class, all in the path that admits a query.

ipaddress.ip_address accepts a scope id since Python 3.9, so
"fe80::1%victim@example.org" passed _is_ip, is_queryable returned on
that branch without inspecting the string further, and rdap.query_ip
interpolated the whole thing into the query URL unquoted. A recipient
identifier reached a registry, which is exactly the disclosure the
fourth property exists to prevent and exactly what the domain-branch
fix closed. Reject a scope id at the admission point. fe80::/10 is
link-local and never a legitimate RDAP target anyway.

The ip branch of worklist() dropped an unparseable value silently
rather than funnelling it through add() like the other two branches,
so a mangled indicator vanished from the manifest instead of showing
up as unusable. It now goes through add(), and is_queryable grew the
check that makes that classification correct: an all-digit last label
is never a TLD, so "999.999.999.999" is a malformed IP rather than a
host to ask a registry about.

Two behaviours had no test and both mutations survived. _host_of's
root-dot strip is now covered through the url branch, since failing to
fold discloses one host to a registry twice; the .lower() half was
dead work because urlsplit already lowercases, and is dropped with a
comment saying so. The two-label requirement is asserted on
is_queryable directly, because a single-label name matches no tld in
the bootstrap and a resolve test could never fail whatever the
validator decides.

The refusal message named no reason, leaving a reviewer unable to tell
whether the parser mangled a legitimate host or the attacker planted
something. Two cases now: a non-ASCII name is plausibly a real IDN
indicator to chase by hand, an illegal character is hostile.

Also let KeyboardInterrupt and SystemExit through resolve()'s handler,
so stopping a long run is not recorded as a failed query on whichever
indicator was in flight.

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

2 files changed, 112 insertions, 11 deletions

diff --git a/abusectl/contacts.py b/abusectl/contacts.py
index 251a3d4..f38f888 100644
--- a/abusectl/contacts.py
+++ b/abusectl/contacts.py
@@ -43,6 +43,10 @@ from dataclasses import dataclass, field
 
 from . import rdap
 
+_MAX_NAME = 253
+_MAX_LABEL = 63
+_LABEL_CHARS = set("abcdefghijklmnopqrstuvwxyz0123456789-")
+
 
 @dataclass
 class WorkItem:
@@ -66,12 +70,10 @@ def _host_of(url: str) -> str | None:
     host = parts.hostname
     if not host:
         return None
-    return host.strip(".").lower() or None
-
-
-_MAX_NAME = 253
-_MAX_LABEL = 63
-_LABEL_CHARS = set("abcdefghijklmnopqrstuvwxyz0123456789-")
+    # urlsplit().hostname already lowercases, so only the root dot is left
+    # to strip. "a.example.invalid." and "a.example.invalid" are the same
+    # host, and failing to fold them discloses it to a registry twice.
+    return host.strip(".") or None
 
 
 def is_queryable(value: str) -> bool:
@@ -104,10 +106,25 @@ def is_queryable(value: str) -> bool:
             return False
         if not set(label) <= _LABEL_CHARS:
             return False
-    return len(labels) >= 2
+    if len(labels) < 2:
+        return False
+    # An all-digit last label is never a TLD, so a value shaped like this
+    # is a malformed IP, not a host. Without this, "999.999.999.999" is
+    # refused by _is_ip and then accepted as a domain, and the tool asks a
+    # registry about garbage instead of showing the user the mangled
+    # indicator.
+    return not labels[-1].isdigit()
 
 
 def _is_ip(value: str) -> bool:
+    # ipaddress accepts a scope id ("fe80::1%eth0") since Python 3.9, and
+    # everything after the % is free text the attacker chose, which
+    # rdap.query_ip interpolates into the query URL unquoted. A scope id
+    # means nothing to a registry, and carrying one would send whatever
+    # was written there, the recipient's own address included, to a third
+    # party. Refuse it before is_queryable admits the value.
+    if "%" in value:
+        return False
     try:
         ipaddress.ip_address(value)
         return True
@@ -115,6 +132,21 @@ def _is_ip(value: str) -> bool:
         return False
 
 
+def _refusal_reason(value: str) -> str:
+    """Say WHY a value was refused, in terms the reviewer can act on.
+
+    A generic sentence leaves a consultant unable to tell whether the
+    parser mangled a legitimate host or the attacker planted something.
+    Two cases, deliberately: a non-ASCII name is plausibly a real IDN
+    indicator to chase by hand, an illegal character is hostile.
+    """
+    if not value.isascii():
+        return ("refused as a query: not ASCII, and we do not guess at an "
+                "IDN encoding, so it was never sent to a registry")
+    return ("refused as a query: not a bare host or IP address, "
+            "so it was never sent to a registry")
+
+
 def worklist(iocs: list[dict]) -> list[WorkItem]:
     """Build the deduplicated list of queries for a set of indicators.
 
@@ -142,7 +174,11 @@ def worklist(iocs: list[dict]) -> list[WorkItem]:
         ioc_id = ioc.get("id")
 
         if ioc_type in ("ipv4", "ipv6"):
-            if _is_ip(value):
+            # Unconditionally, like the other two branches: a value
+            # is_queryable refuses becomes "unusable" and stays visible.
+            # Testing _is_ip here instead dropped it without a trace, which
+            # is the silent failure the add() comment above warns about.
+            if value:
                 add("ip", value, ioc_id)
         elif ioc_type == "domain":
             host = value.strip(".").lower()
@@ -178,7 +214,7 @@ def resolve(iocs: list[dict], bootstraps: dict, fetch=rdap.http_fetch) -> list[d
         }
 
         if item.kind == "unusable":
-            entry["error"] = "not a usable hostname, so it was never queried"
+            entry["error"] = _refusal_reason(item.query)
             results.append(entry)
             continue
 
@@ -202,6 +238,10 @@ def resolve(iocs: list[dict], bootstraps: dict, fetch=rdap.http_fetch) -> list[d
                     continue
                 if queried and queried != item.query:
                     entry["queried"] = queried
+        except (KeyboardInterrupt, SystemExit):
+            # A user stopping a long run must not be recorded as a failed
+            # query on whichever indicator happened to be in flight.
+            raise
         except Exception as exc:
             entry["error"] = f"{type(exc).__name__}: {exc}"
             results.append(entry)
diff --git a/tests/test_contacts.py b/tests/test_contacts.py
index 0451fe6..d3267cf 100644
--- a/tests/test_contacts.py
+++ b/tests/test_contacts.py
@@ -108,6 +108,26 @@ class Worklist(unittest.TestCase):
         iocs = [{"id": "ioc-1", "type": "url", "value": "not a url"}]
         self.assertEqual(contacts.worklist(iocs), [])
 
+    def test_a_trailing_dot_url_folds_with_the_bare_host(self):
+        """_host_of must strip the root dot, not just the domain branch.
+        Failing to fold means disclosing the same host to a registry
+        twice, which doubles what the user leaks per campaign."""
+        iocs = [
+            {"id": "ioc-1", "type": "url", "value": "http://a.example.invalid./x"},
+            {"id": "ioc-2", "type": "url", "value": "http://a.example.invalid/x"},
+        ]
+        work = contacts.worklist(iocs)
+        self.assertEqual(len(work), 1)
+        self.assertEqual(work[0].query, "a.example.invalid")
+
+    def test_an_unparseable_ip_indicator_is_flagged_rather_than_dropped(self):
+        """Silent was the bug: an ip indicator the parser mangled must
+        still show up in the manifest for the user to see."""
+        iocs = [{"id": "ioc-1", "type": "ipv4", "value": "999.999.999.999"}]
+        work = contacts.worklist(iocs)
+        self.assertEqual([(i.kind, i.query) for i in work],
+                         [("unusable", "999.999.999.999")])
+
 
 class Resolve(unittest.TestCase):
     IPV4 = {"services": [[["198.51.100.0/24"], ["https://rir.example.invalid/"]]]}
@@ -290,6 +310,26 @@ class HostileDomainIndicator(unittest.TestCase):
         calls, _ = self._calls_for("a/../../x.invalid")
         self.assertEqual(calls, [])
 
+    def test_an_ipv6_scope_id_is_never_queried(self):
+        """ipaddress.ip_address accepts a scope id since Python 3.9, and
+        everything after the % is free text the attacker chose. rdap
+        interpolates the query unquoted, so a scope id carrying the
+        recipient's own address would reach a registry, which is the
+        disclosure the fourth property exists to prevent."""
+        for value in ("fe80::1%eth0", "fe80::1%victim@example.org"):
+            with self.subTest(value=value):
+                calls, results = self._calls_for(value, ioc_type="ipv6")
+                self.assertEqual(calls, [])
+                self.assertEqual(results[0]["iocs"], ["ioc-1"])
+                self.assertIn("refused as a query", results[0]["error"])
+
+    def test_an_ipv6_scope_id_in_a_url_is_never_queried(self):
+        calls, results = self._calls_for(
+            "http://[fe80::1%25eth0]/x", ioc_type="url"
+        )
+        self.assertEqual(calls, [])
+        self.assertIn("refused as a query", results[0]["error"])
+
     def test_query_fragment_and_space_values_are_never_queried(self):
         for value in ("a?e=secret.invalid", "a#frag.invalid", "a b.invalid",
                       "a@b.invalid", "a:80.invalid", "a\x00b.invalid"):
@@ -304,12 +344,23 @@ class HostileDomainIndicator(unittest.TestCase):
         self.assertEqual(len(results), 1)
         self.assertEqual(results[0]["iocs"], ["ioc-1"])
         self.assertEqual(results[0]["abuse"], [])
-        self.assertIn("hostname", results[0]["error"])
+        self.assertEqual(
+            results[0]["error"],
+            "refused as a query: not a bare host or IP address, "
+            "so it was never sent to a registry",
+        )
 
     def test_a_non_ascii_host_is_rejected_rather_than_guessed_at(self):
+        """A non-ASCII name is plausibly a legitimate IDN indicator the
+        user may want to handle by hand, and an illegal character is
+        hostile. Those imply different actions, so the reasons differ."""
         calls, results = self._calls_for("exämple.invalid")
         self.assertEqual(calls, [])
-        self.assertIn("hostname", results[0]["error"])
+        self.assertEqual(
+            results[0]["error"],
+            "refused as a query: not ASCII, and we do not guess at an IDN "
+            "encoding, so it was never sent to a registry",
+        )
 
     def test_a_bad_label_is_rejected(self):
         for value in ("a..invalid", "-a.invalid", "a-.invalid",
@@ -318,6 +369,16 @@ class HostileDomainIndicator(unittest.TestCase):
                 calls, _ = self._calls_for(value)
                 self.assertEqual(calls, [])
 
+    def test_a_single_label_host_is_refused(self):
+        """Asserted on is_queryable rather than through resolve: a
+        single-label name matches no tld in the bootstrap, so a resolve
+        test issues no query whatever the validator decides and could
+        never fail. A bare label is not a registrable name and asking a
+        registry about one discloses the investigation for nothing."""
+        for value in ("localhost", "invalid", "a"):
+            with self.subTest(value=value):
+                self.assertFalse(contacts.is_queryable(value))
+
     def test_a_normal_host_still_resolves(self):
         calls, results = self._calls_for("mail.example.invalid")
         self.assertEqual(