chg: [rdap] look hostnames up at their registered domain

misp_modules/modules/expansion/rdap.py

@@ -3,6 +3,7 @@
 from urllib.parse import urlparse
 
 import requests
+from publicsuffixlist import PublicSuffixList
 from pymisp import MISPAttribute, MISPEvent, MISPObject
 
 from . import check_input_attribute, standard_error_message
@@ -12,7 +13,7 @@
     "format": "misp_standard",
 }
 moduleinfo = {
-    "version": 1,
+    "version": 2,
     "author": "Ali Bhutto",
     "description": (
         "An expansion module to query the public RDAP bootstrap (rdap.org) for"
@@ -29,7 +30,10 @@
         " input, resolves a URL to its host, and queries the rdap.org bootstrap"
         " which redirects to the authoritative RDAP server for the object. The"
         " registrar, registration and expiration dates, name servers, status and"
-        " registrant information are parsed into a MISP whois object."
+        " registrant information are parsed into a MISP whois object. A"
+        " hostname, subdomain or URL host is looked up at its registered domain"
+        " (ICANN section of the Public Suffix List), and the object's comment"
+        " says so."
     ),
     "references": ["https://about.rdap.org/", "https://rdap.org/"],
     "input": "A domain, hostname, IP address or URL attribute.",
@@ -39,6 +43,11 @@
 
 RDAP_URL = "https://rdap.org"
 
+# Registries delegate under ICANN suffixes only: a name under a private suffix
+# such as duckdns.org is registered under .org, so its registry answers for
+# duckdns.org and not for foo.duckdns.org.
+_PSL = PublicSuffixList(only_icann=True)
+
 # RDAP event actions (RFC 9083) mapped to whois object relations.
 _EVENT_MAPPING = {
     "registration": "creation-date",
@@ -55,6 +64,12 @@ def _is_ip(value):
     return True
 
 
+def _registered_domain(host):
+    """The registered domain of ``host``, or ``host`` itself when it is a
+    public suffix."""
+    return _PSL.privatesuffix(host) or host
+
+
 def _vcard_value(entity, field):
     """Pull a single field (e.g. ``fn``, ``org``, ``email``) out of an RDAP
     entity's jCard ``vcardArray`` (RFC 7095), or ``None``."""
@@ -83,10 +98,19 @@ def _add_if(misp_object, relation, value):
         misp_object.add_attribute(relation, value)
 
 
-def _parse_rdap(rdap, queried_value, is_ip):
-    """Build a MISP whois object from an RDAP response."""
+def _parse_rdap(rdap, queried_value, is_ip, host=None):
+    """Build a MISP whois object from an RDAP response.
+
+    ``queried_value`` is the name asked for, used rather than RDAP's
+    ``ldhName`` which some registries return uppercased. ``host`` is the input
+    host when it differs from the registered domain asked for. It is only named
+    in the comment, never added as a second ``domain``, so that the record does
+    not read as the host's own registration.
+    """
     whois = MISPObject("whois")
     whois.add_attribute("ip-address" if is_ip else "domain", queried_value)
+    if host:
+        whois.add_attribute("comment", f"Registration data for {queried_value}, the registered domain of {host}.")
 
     registrar = _entity_by_role(rdap.get("entities"), "registrar")
     if registrar:
@@ -128,8 +152,14 @@ def handler(q=False):
         if not host:
             return {"error": f"Could not extract a host from URL {value}."}
         value = host
+    value = value.strip().lower().rstrip(".")
 
     is_ip = _is_ip(value)
+    host = None
+    if not is_ip:
+        registered = _registered_domain(value)
+        if registered != value:
+            host, value = value, registered
     path = f"ip/{value}" if is_ip else f"domain/{value}"
     try:
         response = requests.get(
@@ -140,7 +170,11 @@ def handler(q=False):
     except requests.RequestException as e:
         return {"error": f"Error while querying rdap.org: {e}"}
 
-    if response.status_code == 404:
+    # Not every registry answers 404 for an unknown name (PIR answers 400). Do
+    # not call it unregistered: a TLD without an RDAP service also answers 404.
+    if response.status_code == 404 or (response.status_code == 400 and not is_ip):
+        if host:
+            return {"error": f"No RDAP record found for {value} (the registered domain of {host})."}
         return {"error": f"No RDAP record found for {value}."}
     if response.status_code != 200:
         return {"error": f"Error while querying rdap.org - {response.status_code}: {response.reason}"}
@@ -155,7 +189,7 @@ def handler(q=False):
     input_attribute.from_dict(**attribute)
     misp_event.add_attribute(**input_attribute)
 
-    whois = _parse_rdap(rdap, value, is_ip)
+    whois = _parse_rdap(rdap, value, is_ip, host)
     whois.add_reference(input_attribute.uuid, "related-to")
     misp_event.add_object(whois)
 

tests/test_rdap.py

@@ -84,14 +84,111 @@ def test_ip_uses_ip_endpoint_and_relation():
     assert ("ip-address", "1.1.1.1") in {(a["object_relation"], a["value"]) for a in whois["Attribute"]}
 
 
-def test_url_resolves_to_host():
+_EMPTY_DOMAIN_RDAP = {"objectClassName": "domain", "events": [], "entities": []}
+
+
+def _relations(whois, relation):
+    return [a["value"] for a in whois["Attribute"] if a["object_relation"] == relation]
+
+
+def _asked(mock_get):
+    assert mock_get.call_count == 1
+    return mock_get.call_args.args[0]
+
+
+def test_url_resolves_to_registered_domain_of_host():
     attribute = {
         "type": "url",
         "value": "https://sub.example.com/path?q=1",
         "uuid": "5b582d80-7a7e-4b6a-9f22-77656e72bb3b",
     }
-    _, mock_get = _run(attribute, {"objectClassName": "domain", "events": [], "entities": []})
-    assert "rdap.org/domain/sub.example.com" in mock_get.call_args.args[0]
+    result, mock_get = _run(attribute, _EMPTY_DOMAIN_RDAP)
+    assert _asked(mock_get).endswith("rdap.org/domain/example.com")
+    whois = _whois_object(result)
+    assert _relations(whois, "domain") == ["example.com"]
+    assert _relations(whois, "comment") == [
+        "Registration data for example.com, the registered domain of sub.example.com."
+    ]
+
+
+def test_subdomain_is_looked_up_at_registered_domain():
+    attribute = {"type": "hostname", "value": "login.example.com", "uuid": "5b582d80-7a7e-4b6a-9f22-77656e72bb3b"}
+    result, mock_get = _run(attribute, _DOMAIN_RDAP)
+    assert _asked(mock_get).endswith("rdap.org/domain/example.com")
+    whois = _whois_object(result)
+    assert _relations(whois, "domain") == ["example.com"]
+    assert _relations(whois, "comment") == [
+        "Registration data for example.com, the registered domain of login.example.com."
+    ]
+
+
+def test_registered_domain_has_no_comment():
+    attribute = {"type": "domain", "value": "example.com", "uuid": "5b582d80-7a7e-4b6a-9f22-77656e72bb3b"}
+    result, mock_get = _run(attribute, _DOMAIN_RDAP)
+    assert _asked(mock_get).endswith("rdap.org/domain/example.com")
+    whois = _whois_object(result)
+    assert _relations(whois, "domain") == ["example.com"]
+    assert _relations(whois, "comment") == []
+
+
+def test_multi_label_public_suffix():
+    attribute = {"type": "hostname", "value": "news.bbc.co.uk", "uuid": "5b582d80-7a7e-4b6a-9f22-77656e72bb3b"}
+    result, mock_get = _run(attribute, _EMPTY_DOMAIN_RDAP)
+    assert _asked(mock_get).endswith("rdap.org/domain/bbc.co.uk")
+    assert _relations(_whois_object(result), "domain") == ["bbc.co.uk"]
+
+
+def test_private_suffix_is_ignored():
+    attribute = {"type": "hostname", "value": "foo.duckdns.org", "uuid": "5b582d80-7a7e-4b6a-9f22-77656e72bb3b"}
+    result, mock_get = _run(attribute, _EMPTY_DOMAIN_RDAP)
+    assert _asked(mock_get).endswith("rdap.org/domain/duckdns.org")
+    assert _relations(_whois_object(result), "domain") == ["duckdns.org"]
+
+
+def test_input_is_normalised():
+    attribute = {"type": "domain", "value": " Login.Example.COM. ", "uuid": "5b582d80-7a7e-4b6a-9f22-77656e72bb3b"}
+    result, mock_get = _run(attribute, _DOMAIN_RDAP)
+    assert _asked(mock_get).endswith("rdap.org/domain/example.com")
+    assert _relations(_whois_object(result), "comment") == [
+        "Registration data for example.com, the registered domain of login.example.com."
+    ]
+
+
+def test_public_suffix_is_asked_as_given():
+    attribute = {"type": "domain", "value": "co.uk", "uuid": "5b582d80-7a7e-4b6a-9f22-77656e72bb3b"}
+    result, mock_get = _run(attribute, _EMPTY_DOMAIN_RDAP)
+    assert _asked(mock_get).endswith("rdap.org/domain/co.uk")
+    whois = _whois_object(result)
+    assert _relations(whois, "domain") == ["co.uk"]
+    assert _relations(whois, "comment") == []
+
+
+def test_unknown_tld_uses_default_rule():
+    attribute = {"type": "domain", "value": "foo.notatld", "uuid": "5b582d80-7a7e-4b6a-9f22-77656e72bb3b"}
+    result, mock_get = _run(attribute, _EMPTY_DOMAIN_RDAP)
+    assert _asked(mock_get).endswith("rdap.org/domain/foo.notatld")
+    assert _relations(_whois_object(result), "comment") == []
+
+
+def test_domain_relation_ignores_ldhname_case():
+    attribute = {"type": "hostname", "value": "www.example.com", "uuid": "5b582d80-7a7e-4b6a-9f22-77656e72bb3b"}
+    result, mock_get = _run(attribute, dict(_DOMAIN_RDAP, ldhName="EXAMPLE.COM"))
+    assert _asked(mock_get).endswith("rdap.org/domain/example.com")
+    assert _relations(_whois_object(result), "domain") == ["example.com"]
+
+
+def test_bad_request_on_registered_domain_names_both():
+    attribute = {"type": "hostname", "value": "login.example.org", "uuid": "5b582d80-7a7e-4b6a-9f22-77656e72bb3b"}
+    result, mock_get = _run(attribute, None, status_code=400)
+    assert _asked(mock_get).endswith("rdap.org/domain/example.org")
+    assert result == {"error": "No RDAP record found for example.org (the registered domain of login.example.org)."}
+
+
+def test_not_found_on_registered_domain_names_both():
+    attribute = {"type": "hostname", "value": "x.github.io", "uuid": "5b582d80-7a7e-4b6a-9f22-77656e72bb3b"}
+    result, mock_get = _run(attribute, None, status_code=404)
+    assert _asked(mock_get).endswith("rdap.org/domain/github.io")
+    assert result == {"error": "No RDAP record found for github.io (the registered domain of x.github.io)."}
 
 
 def test_not_found_returns_error():