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 --- tests/test_contacts.py | 98 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 98 insertions(+) (limited to 'tests') 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