diff options
| -rw-r--r-- | abusectl/rdap.py | 87 | ||||
| -rw-r--r-- | tests/test_rdap.py | 85 |
2 files changed, 172 insertions, 0 deletions
diff --git a/abusectl/rdap.py b/abusectl/rdap.py index fc9cb12..94a5d02 100644 --- a/abusectl/rdap.py +++ b/abusectl/rdap.py @@ -181,3 +181,90 @@ def server_for_tld(tld: str, bootstrap_data: dict) -> str | None: if any(name.lower() == wanted for name in names): return urls[0] return None + + +import email.utils + +_MAX_ENTITY_DEPTH = 4 + + +def _valid_address(raw: str) -> str | None: + """Return a usable address, or None. + + This value becomes a mail recipient in report and submit, so it is + validated where it enters rather than where it is used: a control + character here is header injection into mail this tool sends. + """ + if not isinstance(raw, str) or not raw.strip(): + return None + if any(character in raw for character in "\r\n\t"): + return None + if any(ord(character) < 32 for character in raw): + return None + + name, address = email.utils.parseaddr(raw) + if not address or address.count("@") != 1: + return None + local, _, domain = address.partition("@") + if not local or not domain or "." not in domain: + return None + return address + + +def _emails_from_vcard(entity: dict) -> list[str]: + """Pull every email property value out of a jCard. + + jCard encodes vCard as nested arrays: vcardArray[1] is a list of + [name, params, type, value] properties, so the value is at index 3. + """ + found = [] + vcard = entity.get("vcardArray") + if not isinstance(vcard, list) or len(vcard) < 2: + return found + for prop in vcard[1]: + if not isinstance(prop, list) or len(prop) < 4: + continue + if prop[0] != "email": + continue + address = _valid_address(prop[3]) + if address: + found.append(address) + return found + + +def abuse_addresses(response: dict) -> list[str]: + """Return every published abuse address in an RDAP response. + + STRICT: only an entity whose roles contain "abuse" counts. There is no + fallback to a technical or registrant contact, because that is a named + human who never volunteered for abuse mail, and no fallback to + abuse@<domain> by convention, because for a phishing domain that + mailbox belongs to the ATTACKER and mailing it would confirm both the + catch and that the user's address is live. + + A links referral is never followed. If the address is not in this + response, there is no address: following a URL the response chose for + us is an outbound fetch under remote control. + """ + found: list[str] = [] + + def walk(entities, depth): + if depth > _MAX_ENTITY_DEPTH or not isinstance(entities, list): + return + for entity in entities: + if not isinstance(entity, dict): + continue + roles = entity.get("roles") or [] + if isinstance(roles, list) and "abuse" in roles: + found.extend(_emails_from_vcard(entity)) + walk(entity.get("entities"), depth + 1) + + walk(response.get("entities"), 1) + + seen = set() + unique = [] + for address in found: + if address not in seen: + seen.add(address) + unique.append(address) + return unique diff --git a/tests/test_rdap.py b/tests/test_rdap.py index 14a1fd4..b7baa21 100644 --- a/tests/test_rdap.py +++ b/tests/test_rdap.py @@ -176,5 +176,90 @@ class ServerSelection(unittest.TestCase): self.assertIsNone(rdap.server_for_tld("example", self.DNS)) +def _entity(roles, emails, entities=None): + """Build an RDAP entity in real jCard shape.""" + properties = [["version", {}, "text", "4.0"]] + for address in emails: + properties.append(["email", {}, "text", address]) + entity = {"roles": roles, "vcardArray": ["vcard", properties]} + if entities: + entity["entities"] = entities + return entity + + +class AbuseExtraction(unittest.TestCase): + def test_an_abuse_entity_yields_its_address(self): + response = {"entities": [_entity(["abuse"], ["abuse@example.invalid"])]} + self.assertEqual( + rdap.abuse_addresses(response), ["abuse@example.invalid"] + ) + + def test_a_nested_abuse_entity_is_found(self): + """The abuse entity is usually a child of the organisation entity.""" + response = { + "entities": [ + _entity( + ["registrant"], [], + entities=[_entity(["abuse"], ["abuse@example.invalid"])], + ) + ] + } + self.assertEqual( + rdap.abuse_addresses(response), ["abuse@example.invalid"] + ) + + def test_a_technical_only_response_yields_nothing(self): + """A technical contact is a named human who never volunteered to + receive abuse mail. Mailing them is useless and is a small privacy + harm to an uninvolved third party.""" + response = {"entities": [_entity(["technical"], ["someone@example.invalid"])]} + self.assertEqual(rdap.abuse_addresses(response), []) + + def test_every_abuse_address_is_kept(self): + """Some netblocks publish two desks, and picking one arbitrarily + can drop the one that would have answered.""" + response = { + "entities": [ + _entity(["abuse"], ["one@example.invalid", "two@example.invalid"]) + ] + } + self.assertEqual( + rdap.abuse_addresses(response), + ["one@example.invalid", "two@example.invalid"], + ) + + def test_a_newline_in_an_address_is_rejected(self): + """The address becomes a mail recipient in report and submit, so a + CRLF here is header injection into mail this tool sends.""" + response = { + "entities": [ + _entity(["abuse"], ["abuse@example.invalid\r\nBcc: victim@example.org"]) + ] + } + self.assertEqual(rdap.abuse_addresses(response), []) + + def test_a_non_address_is_rejected(self): + response = {"entities": [_entity(["abuse"], ["not an address"])]} + self.assertEqual(rdap.abuse_addresses(response), []) + + def test_recursion_is_depth_capped(self): + """Remote JSON must not be able to hang the tool.""" + deep = _entity(["abuse"], ["deep@example.invalid"]) + for _ in range(10): + deep = _entity(["registrant"], [], entities=[deep]) + self.assertEqual(rdap.abuse_addresses({"entities": [deep]}), []) + + def test_duplicate_addresses_collapse(self): + response = { + "entities": [ + _entity(["abuse"], ["abuse@example.invalid"]), + _entity(["abuse"], ["abuse@example.invalid"]), + ] + } + self.assertEqual( + rdap.abuse_addresses(response), ["abuse@example.invalid"] + ) + + if __name__ == "__main__": unittest.main() |