diff options
| -rw-r--r-- | abusectl/contacts.py | 58 | ||||
| -rw-r--r-- | tests/test_contacts.py | 65 |
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( |