From 67cd02eb7747977d5dac984f2e36b198852c267d Mon Sep 17 00:00:00 2001 From: "Danilo M." Date: Wed, 9 Sep 2026 09:33:08 +0200 Subject: 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 ` 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 Claude-Session: https://claude.ai/code/session_01Wrfqr2xqQfhtXCscU7zrdz --- abusectl/contacts.py | 55 +++++++++++++++++++++++++++- tests/test_contacts.py | 98 ++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 152 insertions(+), 1 deletion(-) 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 \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() -- cgit v1.2.3