fix: validate a query host at the one point a query is admitted

THE FOURTH PROPERTY was breached through the domain branch of worklist().
The url branch is cleaned by _host_of, which uses urlsplit().hostname,
and the domain branch did only value.strip(".").lower(). rdap.query_domain
then interpolates that value into the fetch URL with no quoting.

parse._domain_of takes everything after the @ of a From, Sender or
Reply-To addr-spec, and email.utils.parseaddr permits /, ?, # and %
there, so the whole shape is attacker-controlled through a header they
own. A From of `Bank <phish@victim%40example.org.invalid>` sent the
recipient's own address to a registry, which is precisely the identity
disclosure the property exists to prevent, reaching a third party.
`a/../../x.invalid` escaped the /domain/ endpoint altogether, and query,
fragment and space values all reached the wire.

Fixed at the SINGLE admission point rather than in the offending branch,
because per-branch validation is what failed here: one branch was
cleaned, the next was written without it. is_queryable() now guards
worklist()'s add(), so domain, url and any future branch pass through it.

A rejected value is not dropped. It keeps an entry with an empty abuse
list and an error saying it was never queried, following the rule
suspect_path_segments already sets: flagged and visible to the user
during review, because silent was the bug.

A non-ASCII host is refused rather than encoded to punycode. Guessing the
encoding of an attacker-supplied name is a query that cannot be
justified, and the ceiling is noted in a ponytail comment.

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

2 files changed, 152 insertions, 1 deletions

diff --git a/abusectl/contacts.py b/abusectl/contacts.py
index a0857c4..251a3d4 100644
--- a/abusectl/contacts.py
+++ b/abusectl/contacts.py
@@ -31,6 +31,10 @@ nothing else. Path, query and fragment never leave the machine.
 This is a trap rather than a theoretical concern. The obvious
 implementation resolves "a contact for each indicator" by reading each
 indicator's value, and for a url indicator that value is an entire URL.
+
+The url branch was cleaned and the domain branch was not, which is why
+is_queryable() sits at the SINGLE point where a query is admitted rather
+than in either branch. See its docstring for what got through.
 """
 
 import ipaddress
@@ -44,7 +48,7 @@ from . import rdap
 class WorkItem:
     """One thing to ask a registry about, and every indicator behind it."""
 
-    kind: str  # "ip" or "domain"
+    kind: str  # "ip", "domain", or "unusable"
     query: str
     iocs: list[str] = field(default_factory=list)
 
@@ -65,6 +69,44 @@ def _host_of(url: str) -> str | None:
     return host.strip(".").lower() or None
 
 
+_MAX_NAME = 253
+_MAX_LABEL = 63
+_LABEL_CHARS = set("abcdefghijklmnopqrstuvwxyz0123456789-")
+
+
+def is_queryable(value: str) -> bool:
+    """Is this a plausible bare host or IP, safe to put in a query URL?
+
+    THE SINGLE ADMISSION POINT for a query string. It is here rather than
+    in each branch of worklist() because per-branch validation is exactly
+    what failed: the url branch was cleaned by _host_of and the domain
+    branch was not, so anything the attacker wrote after the @ of a From
+    header went straight into the fetch URL unquoted. That reached a
+    registry carrying the recipient's own address (`phish@victim%40...`),
+    which is the disclosure the fourth property exists to prevent, and
+    `a/../../x.invalid` escaped the /domain/ endpoint altogether. A future
+    branch gets this for free.
+
+    ponytail: ASCII only. A non-ASCII host is REFUSED, not encoded to
+    punycode, because guessing the encoding of an attacker-supplied name
+    is a query we cannot justify. The ceiling is IDN indicators, which
+    resolve to nothing and are flagged for the user instead.
+    """
+    if not isinstance(value, str) or not value or len(value) > _MAX_NAME:
+        return False
+    if _is_ip(value):
+        return True
+    labels = value.split(".")
+    for label in labels:
+        if not label or len(label) > _MAX_LABEL:
+            return False
+        if label[0] == "-" or label[-1] == "-":
+            return False
+        if not set(label) <= _LABEL_CHARS:
+            return False
+    return len(labels) >= 2
+
+
 def _is_ip(value: str) -> bool:
     try:
         ipaddress.ip_address(value)
@@ -82,6 +124,12 @@ def worklist(iocs: list[dict]) -> list[WorkItem]:
     items: dict[tuple[str, str], WorkItem] = {}
 
     def add(kind, query, ioc_id):
+        # Every branch funnels through here, so nothing reaches a query URL
+        # without passing is_queryable. A rejected value is kept as
+        # "unusable" rather than dropped: silent was the bug in
+        # suspect_path_segments too, and the user must see it during review.
+        if not is_queryable(query):
+            kind = "unusable"
         key = (kind, query)
         if key not in items:
             items[key] = WorkItem(kind=kind, query=query)
@@ -129,6 +177,11 @@ def resolve(iocs: list[dict], bootstraps: dict, fetch=rdap.http_fetch) -> list[d
             "source": "rdap",
         }
 
+        if item.kind == "unusable":
+            entry["error"] = "not a usable hostname, so it was never queried"
+            results.append(entry)
+            continue
+
         try:
             if item.kind == "ip":
                 family = "ipv6" if ":" in item.query else "ipv4"
diff --git a/tests/test_contacts.py b/tests/test_contacts.py
index 0ac197d..0451fe6 100644
--- a/tests/test_contacts.py
+++ b/tests/test_contacts.py
@@ -235,5 +235,103 @@ class Resolve(unittest.TestCase):
             self.assertNotIn("?", url)
 
 
+class HostileDomainIndicator(unittest.TestCase):
+    """THE FOURTH PROPERTY through the domain branch.
+
+    The url branch is cleaned by _host_of. The domain branch took its value
+    from parse._domain_of, which is everything after the @ of a From,
+    Sender or Reply-To addr-spec, a header the attacker owns completely,
+    and email.utils.parseaddr permits /, ?, # and % there.
+    """
+
+    IPV4 = {"services": [[["198.51.100.0/24"], ["https://rir.example.invalid/"]]]}
+    IPV6 = {"services": []}
+    DNS = {"services": [[["invalid"], ["https://registry.example.invalid/"]]]}
+
+    def _bootstraps(self):
+        return {"ipv4": self.IPV4, "ipv6": self.IPV6, "dns": self.DNS}
+
+    def _calls_for(self, value, ioc_type="domain"):
+        calls = []
+
+        def fetch(url):
+            calls.append(url)
+            return {"handle": "DOM-1", "entities": []}
+
+        iocs = [{"id": "ioc-1", "type": ioc_type, "value": value}]
+        results = contacts.resolve(
+            iocs, bootstraps=self._bootstraps(), fetch=fetch
+        )
+        return calls, results
+
+    def test_a_sender_domain_carrying_the_victim_address_is_never_queried(self):
+        """The attacker writes the recipient's own address into the domain
+        of the From header, and this tool would send it to a registry. That
+        is the disclosure the fourth property exists to prevent, reaching a
+        third party."""
+        raw = (b"Received: from relay.example.invalid ([192.0.2.10])\r\n"
+               b"From: Bank <phish@victim%40example.org.invalid>\r\n"
+               b"Subject: test\r\n\r\nbody\r\n")
+        from abusectl import parse
+        iocs = parse.iocs(raw, trusted=["192.0.2.0/24"])
+        calls = []
+
+        def fetch(url):
+            calls.append(url)
+            return {"handle": "DOM-1", "entities": []}
+
+        contacts.resolve(iocs, bootstraps=self._bootstraps(), fetch=fetch)
+        for url in calls:
+            self.assertNotIn("victim", url)
+            self.assertNotIn("%40", url)
+
+    def test_a_traversal_value_is_never_queried(self):
+        """a/../../x.invalid escapes the /domain/ endpoint altogether."""
+        calls, _ = self._calls_for("a/../../x.invalid")
+        self.assertEqual(calls, [])
+
+    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"):
+            with self.subTest(value=value):
+                calls, _ = self._calls_for(value)
+                self.assertEqual(calls, [])
+
+    def test_a_rejected_value_is_flagged_rather_than_dropped(self):
+        """Silent was the bug elsewhere too: the user must see it during
+        review rather than wonder why an indicator vanished."""
+        calls, results = self._calls_for("a?e=secret.invalid")
+        self.assertEqual(len(results), 1)
+        self.assertEqual(results[0]["iocs"], ["ioc-1"])
+        self.assertEqual(results[0]["abuse"], [])
+        self.assertIn("hostname", results[0]["error"])
+
+    def test_a_non_ascii_host_is_rejected_rather_than_guessed_at(self):
+        calls, results = self._calls_for("exämple.invalid")
+        self.assertEqual(calls, [])
+        self.assertIn("hostname", results[0]["error"])
+
+    def test_a_bad_label_is_rejected(self):
+        for value in ("a..invalid", "-a.invalid", "a-.invalid",
+                      "x" * 64 + ".invalid", ("a." * 130) + "invalid"):
+            with self.subTest(value=value):
+                calls, _ = self._calls_for(value)
+                self.assertEqual(calls, [])
+
+    def test_a_normal_host_still_resolves(self):
+        calls, results = self._calls_for("mail.example.invalid")
+        self.assertEqual(
+            calls, ["https://registry.example.invalid/domain/mail.example.invalid"]
+        )
+        self.assertEqual(results[0]["handle"], "DOM-1")
+
+    def test_an_ip_valued_domain_indicator_still_takes_the_ip_branch(self):
+        work = contacts.worklist(
+            [{"id": "ioc-1", "type": "domain", "value": "198.51.100.7"}]
+        )
+        self.assertEqual([(i.kind, i.query) for i in work],
+                         [("ip", "198.51.100.7")])
+
+
 if __name__ == "__main__":
     unittest.main()