diff --git a/documentation/expansion/url_dns.json b/documentation/expansion/url_dns.json new file mode 100644 index 000000000..eeb833ebe --- /dev/null +++ b/documentation/expansion/url_dns.json @@ -0,0 +1,60 @@ +{ + "name": "URL DNS and ASN Expansion", + "module": "url_dns", + "module-type": ["expansion"], + "description": "Resolve a URL into IPv4/IPv6 TCP endpoints and IPv4 origin ASNs, linked to the original URL attribute.", + "requirements": ["dnspython", "PyMISP", "DNS access to a recursive resolver, including Team Cymru TXT records"], + "references": ["https://www.team-cymru.com/ip-asn-mapping"], + "features": [ + "Resolve all A and AAAA records with the system resolver or an optional nameserver; deduplicate addresses.", + "Create one standard ip-port object per address with ip, dst-port, protocol=tcp and hostname when the URL uses a hostname.", + "An explicit port overrides scheme defaults. HTTP defaults to 80 and HTTPS to 443. Other schemes require an explicit port from 1 to 65535.", + "Handle bracketed IPv6 literals and IDNA hostnames. IP literals skip forward DNS.", + "Query Team Cymru origin.asn.cymru.com TXT records for globally routable unicast IPv4 addresses. Multi-origin prefixes produce multiple ASN objects.", + "Deduplicate ASN objects within the expansion and preserve all announced IPv4 subnets. Retrieve the ASN description when available.", + "Each ip-port object is derived-from the original URL UUID and belongs-to its ASN objects. Each ASN object is related-to the original URL UUID.", + "Return the original attribute with its UUID and metadata; all newly generated object attributes use to_ids=false.", + "Missing records for one address family are normal. Partial DNS failures and unavailable IPv4 ASN data are reported in endpoint comments; successful endpoint results are retained.", + "IPv6 ASN enrichment is outside this module's scope. Non-global and multicast IPv4 addresses retain their endpoints without ASN queries.", + "This module performs DNS lookups only. The TCP protocol and port are inferred from the URL; no service reachability or protocol negotiation is performed." + ], + "input": { + "format": "misp_standard", + "types": ["url"], + "required_attribute_fields": ["type", "value", "uuid"] + }, + "output": { + "format": "misp_standard", + "objects": ["ip-port", "asn"], + "errors": "Invalid requests, URLs or configuration, and failure to resolve any addresses return an error. ASN lookup failures retain endpoint results." + }, + "config": { + "nameserver": { + "required": false, + "default": "System resolver", + "description": "Optional IPv4 or IPv6 address of a recursive DNS server. Used for both forward resolution and ASN TXT queries. Timeout is two seconds per attempt; lifetime is five seconds per DNS query." + } + }, + "installation": [ + "The module is installed at misp_modules/modules/expansion/url_dns.py and is automatically discovered by misp-modules.", + "Install dnspython and PyMISP in the service's Python environment if missing: python -m pip install dnspython pymisp. Both are already in the repository's minimal dependency set.", + "Restart the misp-modules service using your normal service manager, then enable url_dns in MISP's Enrichment plugin settings.", + "Expand an existing URL attribute. Its original UUID must be provided in the request; do not generate a replacement UUID.", + "No custom MISP object templates or API keys are required. The resolver must be able to retrieve A, AAAA and Team Cymru TXT records." + ], + "verification": { + "command": "python -m pytest -q tests/test_url_dns.py tests/test_rdap.py", + "result": "63 module tests and 6 existing RDAP tests passed with Python 3.12, PyMISP 2.5.34.4 and dnspython 2.9.0.", + "checks": "Strict standard object-template validation, module discovery, Black, isort and flake8 passed.", + "limitations": "DNS responses were mocked. Live DNS and installation into a running MISP instance have not been tested." + }, + "example": { + "module": "url_dns", + "attribute": { + "type": "url", + "value": "https://example.com:8443/path", + "uuid": "5b582d80-7a7e-4b6a-9f22-77656e72bb3b" + }, + "config": {} + } +} diff --git a/misp_modules/modules/expansion/url_dns.py b/misp_modules/modules/expansion/url_dns.py new file mode 100644 index 000000000..c198e99bd --- /dev/null +++ b/misp_modules/modules/expansion/url_dns.py @@ -0,0 +1,217 @@ +"""Expand a URL into DNS endpoints and IPv4 origin ASNs.""" + +import ipaddress +import json +from urllib.parse import urlsplit +from uuid import UUID + +import dns.exception +import dns.resolver +from pymisp import MISPAttribute, MISPEvent, MISPObject +from pymisp.exceptions import PyMISPError + +from . import check_input_attribute, standard_error_message + +mispattributes = {"input": ["url"], "format": "misp_standard"} +moduleinfo = { + "version": "1.0", + "author": "MISP", + "description": "Resolve URL hosts into IPv4/IPv6 IP-port objects and IPv4 ASN objects.", + "module-type": ["expansion"], + "name": "URL DNS and ASN Expansion", + "logo": "", + "requirements": ["dnspython", "PyMISP"], + "features": ( + "Resolve all A and AAAA records, create one TCP ip-port object per address, and query Team Cymru DNS" + " for IPv4 origin ASNs. Explicit ports override HTTP/HTTPS defaults. Every object references the" + " original URL attribute. ASN failures preserve endpoint results with an explanatory comment." + ), + "references": ["https://www.team-cymru.com/ip-asn-mapping"], + "input": "A URL attribute with type, value and UUID.", + "output": "Linked ip-port objects for IPv4/IPv6 and asn objects for public IPv4 addresses.", +} +moduleconfig = ["nameserver"] + + +def _parse_url(value): + if not isinstance(value, str) or not value or any(c.isspace() or ord(c) < 32 for c in value): + raise ValueError("The URL must be a non-empty string without whitespace or control characters.") + parsed = urlsplit(value) + if not parsed.scheme or not parsed.netloc or not parsed.hostname: + raise ValueError("The URL must include a scheme and hostname, for example https://example.org/.") + host = parsed.hostname + if "%" in host: + raise ValueError("Percent-encoded hostnames and scoped IPv6 addresses are not supported.") + port = parsed.port + if parsed.netloc.endswith(":"): + raise ValueError("An explicit port must contain a number.") + if port is None: + port = {"http": 80, "https": 443}.get(parsed.scheme.lower()) + if port is None: + raise ValueError("URL schemes other than HTTP/HTTPS require an explicit TCP port.") + if not 1 <= port <= 65535: + raise ValueError("The TCP port must be between 1 and 65535.") + try: + address = ipaddress.ip_address(host) + except ValueError: + host = host.encode("idna").decode("ascii").rstrip(".") + # Validate before DNS to avoid accepting invalid URL authorities as DNS names. + labels = host.split(".") + if len(host) > 253 or any( + not label + or len(label) > 63 + or label.startswith("-") + or label.endswith("-") + or any(c not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-" for c in label) + for label in labels + ): + raise ValueError("Invalid URL hostname.") + address = None + return host, port, address + + +def _resolve_addresses(resolver, host): + addresses = set() + warnings = [] + for record_type, family in (("A", 4), ("AAAA", 6)): + try: + for record in resolver.resolve(f"{host}.", record_type): + address = ipaddress.ip_address(str(record)) + if address.version == family: + addresses.add(address) + except dns.resolver.NoAnswer: + continue + except dns.resolver.NXDOMAIN: + warnings.append("The URL hostname does not exist (NXDOMAIN).") + break + except (dns.exception.DNSException, ValueError) as error: + warnings.append(f"{record_type} lookup failed ({type(error).__name__}).") + if not addresses: + raise ValueError("No IPv4 or IPv6 addresses resolved. " + " ".join(warnings)) + return sorted(addresses, key=lambda address: (address.version, int(address))), warnings + + +def _txt_records(resolver, name): + # A TXT RR can consist of multiple quoted chunks; concatenate before parsing. + return [b"".join(record.strings).decode("utf-8") for record in resolver.resolve(name, "TXT")] + + +def _origin_asns(resolver, address): + name = ".".join(reversed(str(address).split("."))) + ".origin.asn.cymru.com." + records = set() + for text in _txt_records(resolver, name): + fields = [field.strip() for field in text.split("|")] + if len(fields) < 2: + continue + try: + network = ipaddress.ip_network(fields[1]) + if network.version != 4 or address not in network: + continue + except ValueError: + continue + # Multi-origin prefixes contain space-separated AS numbers. + for number in fields[0].split(): + if number.isascii() and number.isdigit() and 1 <= int(number) <= 4294967295: + records.add((int(number), str(network))) + if not records: + raise ValueError("No valid IPv4 origin ASN record returned.") + return sorted(records) + + +def _asn_object(resolver, number, source_uuid): + obj = MISPObject("asn", strict=True) + obj.add_attribute("asn", f"AS{number}", to_ids=False) + obj.add_reference(source_uuid, "related-to") + try: + for text in _txt_records(resolver, f"AS{number}.asn.cymru.com."): + fields = [field.strip() for field in text.split("|", 4)] + if len(fields) == 5 and fields[0] == str(number) and fields[4] and fields[4] != "NA": + obj.add_attribute("description", fields[4], to_ids=False) + break + except (dns.exception.DNSException, UnicodeError): + obj.comment = "ASN description lookup failed; the origin ASN is retained." + return obj + + +def handler(q=False): + if q is False: + return False + try: + request = json.loads(q) + except (TypeError, ValueError): + return {"error": "Invalid JSON request."} + if not isinstance(request, dict): + return {"error": "The request must be a JSON object."} + attribute = request.get("attribute") + if not isinstance(attribute, dict) or not check_input_attribute(attribute): + return {"error": f"{standard_error_message}, which should contain at least a type, a value and an uuid."} + if attribute["type"] != "url": + return {"error": "Unsupported attribute type: expected url."} + + try: + if not isinstance(attribute["uuid"], str): + raise ValueError("The URL attribute UUID must be a string.") + UUID(attribute["uuid"]) + host, port, literal = _parse_url(attribute["value"]) + config = request.get("config") or {} + if not isinstance(config, dict): + raise ValueError("The module config must be a JSON object.") + resolver = dns.resolver.Resolver() + resolver.timeout = 2 + resolver.lifetime = 5 + if config.get("nameserver"): + resolver.nameservers = [str(ipaddress.ip_address(config["nameserver"]))] + addresses, warnings = ([literal], []) if literal is not None else _resolve_addresses(resolver, host) + + event = MISPEvent() + source = MISPAttribute() + source.from_dict(**attribute) + event.add_attribute(**source) + asns = {} + subnets = {} + for address in addresses: + endpoint = MISPObject("ip-port", strict=True) + endpoint.add_attribute("ip", str(address), to_ids=False) + endpoint.add_attribute("dst-port", port, to_ids=False) + endpoint.add_attribute("protocol", "tcp", to_ids=False) + if literal is None: + endpoint.add_attribute("hostname", host, to_ids=False) + endpoint.add_reference(source.uuid, "derived-from") + notes = list(warnings) + if address.version == 4: + if not address.is_global or address.is_multicast: + notes.append("ASN lookup skipped: this IPv4 address is not globally routable unicast.") + else: + try: + origins = _origin_asns(resolver, address) + except (dns.exception.DNSException, UnicodeError, ValueError) as error: + notes.append(f"IPv4 ASN lookup unavailable ({type(error).__name__}).") + else: + linked_asns = set() + for number, subnet in origins: + if number not in asns: + asns[number] = _asn_object(resolver, number, source.uuid) + subnets[number] = set() + if subnet not in subnets[number]: + asns[number].add_attribute("subnet-announced", subnet, to_ids=False) + subnets[number].add(subnet) + if number not in linked_asns: + endpoint.add_reference(asns[number].uuid, "belongs-to") + linked_asns.add(number) + if notes: + endpoint.comment = " ".join(notes) + event.add_object(endpoint) + for obj in asns.values(): + event.add_object(obj) + data = json.loads(event.to_json()) + return {"results": {key: data[key] for key in ("Attribute", "Object")}} + except (ValueError, TypeError, dns.exception.DNSException, PyMISPError) as error: + return {"error": f"URL expansion failed: {error}"} + + +def introspection(): + return mispattributes + + +def version(): + return dict(moduleinfo, config=moduleconfig) diff --git a/tests/test_url_dns.py b/tests/test_url_dns.py new file mode 100644 index 000000000..c6182f0dd --- /dev/null +++ b/tests/test_url_dns.py @@ -0,0 +1,267 @@ +import json +from unittest.mock import MagicMock + +import dns.exception +import dns.resolver +import dns.rrset +import pytest +from pymisp import MISPObject + +from misp_modules.modules.expansion import url_dns + +SOURCE_UUID = "5b582d80-7a7e-4b6a-9f22-77656e72bb3b" +ORIGIN = "1.1.1.1.origin.asn.cymru.com." +ASN_NAME = "AS13335.asn.cymru.com." + + +@pytest.fixture +def resolver(monkeypatch): + resolver = MagicMock() + monkeypatch.setattr(url_dns.dns.resolver, "Resolver", lambda: resolver) + return resolver + + +def _answers(resolver, records): + def resolve(name, record_type): + answer = records.get((name, record_type), dns.resolver.NoAnswer()) + if isinstance(answer, Exception): + raise answer + return dns.rrset.from_text(name, 60, "IN", record_type, *answer) + + resolver.resolve.side_effect = resolve + + +def _run(value="https://example.com/path", **extra): + request = {"attribute": {"type": "url", "value": value, "uuid": SOURCE_UUID}} + request.update(extra) + return url_dns.handler(json.dumps(request)) + + +def _objects(result, name): + return [obj for obj in result["results"]["Object"] if obj["name"] == name] + + +def _values(obj, relation): + return [attr["value"] for attr in obj["Attribute"] if attr["object_relation"] == relation] + + +def _references(obj): + return {(ref["referenced_uuid"], ref["relationship_type"]) for ref in obj["ObjectReference"]} + + +def test_dual_stack_asn_and_original_url_links(resolver): + _answers( + resolver, + { + ("example.com.", "A"): ["1.1.1.1", "1.0.0.1"], + ("example.com.", "AAAA"): ["2606:4700:4700::1111"], + (ORIGIN, "TXT"): ['"13335 | 1.1.1.0/24 | AU | apnic | 2011-08-11"'], + ("1.0.0.1.origin.asn.cymru.com.", "TXT"): ['"13335 | 1.0.0.0/24 | AU | apnic | 2010-07-14"'], + (ASN_NAME, "TXT"): ['"13335 | US | arin | 2010-07-14 | CLOUDFLARENET, US"'], + }, + ) + result = _run() + endpoints = _objects(result, "ip-port") + asns = _objects(result, "asn") + assert len(endpoints) == 3 + assert len(asns) == 1 + assert _values(asns[0], "asn") == ["AS13335"] + assert _values(asns[0], "description") == ["CLOUDFLARENET, US"] + assert set(_values(asns[0], "subnet-announced")) == {"1.1.1.0/24", "1.0.0.0/24"} + assert (SOURCE_UUID, "related-to") in _references(asns[0]) + assert result["results"]["Attribute"][0]["uuid"] == SOURCE_UUID + assert result["results"]["Attribute"][0]["value"] == "https://example.com/path" + for endpoint in endpoints: + assert _values(endpoint, "dst-port") == ["443"] + assert _values(endpoint, "protocol") == ["tcp"] + assert _values(endpoint, "hostname") == ["example.com"] + assert (SOURCE_UUID, "derived-from") in _references(endpoint) + if ":" not in _values(endpoint, "ip")[0]: + assert (asns[0]["uuid"], "belongs-to") in _references(endpoint) + else: + assert len(_references(endpoint)) == 1 + for obj in result["results"]["Object"]: + strict_obj = MISPObject(obj["name"], strict=True) + strict_obj.from_dict(**obj) + strict_obj._validate() + assert all(attr["to_ids"] is False for attr in obj["Attribute"]) + assert resolver.resolve.call_args_list.count(((ASN_NAME, "TXT"),)) == 1 + + +@pytest.mark.parametrize( + "url,port", + [ + ("http://example.com", "80"), + ("https://example.com", "443"), + ("HTTP://example.com", "80"), + ("http://example.com:8080/path", "8080"), + ("https://user:password@example.com:8443/path?q=:9999", "8443"), + ("ftp://example.com:21/file", "21"), + ("custom://example.com:65535", "65535"), + ("https://example.com/path:8443", "443"), + ("https://example.com?port=:8443", "443"), + ], +) +def test_port_selection(resolver, url, port): + _answers(resolver, {("example.com.", "AAAA"): ["2001:db8::1"]}) + endpoint = _objects(_run(url), "ip-port")[0] + assert _values(endpoint, "dst-port") == [port] + + +@pytest.mark.parametrize( + "url", + [ + "example.com/path", + "ftp://example.com", + "http:///path", + "http://example.com:0", + "http://example.com:65536", + "http://example.com:abc", + "http://example.com:", + "https://[not-ipv6]/", + "http://bad host/", + "http://example.com/\n", + "http://%65xample.com/", + "http://[fe80::1%25eth0]/", + "https://example..com/", + "https://-example.com/", + "https://example.com./:bad port", + "", + None, + 123, + ], +) +def test_invalid_urls_do_not_query_dns(resolver, url): + assert "error" in _run(url) + resolver.resolve.assert_not_called() + + +@pytest.mark.parametrize( + "url,ip,port", [("http://192.168.1.1", "192.168.1.1", "80"), ("https://[2001:db8::1]:8443/", "2001:db8::1", "8443")] +) +def test_ip_literal_skips_forward_dns(resolver, url, ip, port): + result = _run(url) + endpoint = _objects(result, "ip-port")[0] + assert _values(endpoint, "ip") == [ip] + assert _values(endpoint, "dst-port") == [port] + assert not _objects(result, "asn") + resolver.resolve.assert_not_called() + + +def test_public_ipv4_literal_gets_asn(resolver): + _answers(resolver, {(ORIGIN, "TXT"): ['"13335 | 1.1.1.0/24"']}) + result = _run("https://1.1.1.1") + assert len(_objects(result, "asn")) == 1 + assert _objects(result, "asn")[0]["comment"] + assert all(call.args[1] == "TXT" for call in resolver.resolve.call_args_list) + + +def test_idna_hostname(resolver): + _answers(resolver, {("xn--bcher-kva.example.", "AAAA"): ["2001:db8::1"]}) + endpoint = _objects(_run("https://bücher.example./"), "ip-port")[0] + assert _values(endpoint, "hostname") == ["xn--bcher-kva.example"] + + +@pytest.mark.parametrize("failure", [dns.exception.Timeout(), dns.resolver.NoNameservers()]) +def test_one_family_failure_preserves_other_family(resolver, failure): + _answers(resolver, {("example.com.", "A"): failure, ("example.com.", "AAAA"): ["2001:db8::1"]}) + endpoint = _objects(_run(), "ip-port")[0] + assert "A lookup failed" in endpoint["comment"] + + +@pytest.mark.parametrize("failure", [dns.resolver.NXDOMAIN(), dns.resolver.NoAnswer(), dns.exception.Timeout()]) +def test_no_addresses_returns_error(resolver, failure): + _answers(resolver, {("example.com.", "A"): failure, ("example.com.", "AAAA"): failure}) + assert "No IPv4 or IPv6 addresses" in _run()["error"] + + +@pytest.mark.parametrize("failure", [dns.exception.Timeout(), dns.resolver.NXDOMAIN(), dns.resolver.NoAnswer()]) +def test_asn_failure_keeps_endpoint(resolver, failure): + _answers(resolver, {("example.com.", "A"): ["1.1.1.1"], (ORIGIN, "TXT"): failure}) + result = _run() + assert not _objects(result, "asn") + assert "ASN lookup unavailable" in _objects(result, "ip-port")[0]["comment"] + + +def test_multiple_origin_asns_and_split_txt_chunks(resolver): + _answers(resolver, {(ORIGIN, "TXT"): ['"13335 64500 | " "1.1.1.0/24 | AU | apnic | 2011-08-11"']}) + result = _run("http://1.1.1.1") + asns = _objects(result, "asn") + assert {_values(obj, "asn")[0] for obj in asns} == {"AS13335", "AS64500"} + endpoint = _objects(result, "ip-port")[0] + assert len(_references(endpoint)) == 3 + + +def test_overlapping_prefixes_do_not_duplicate_asn_references(resolver): + _answers(resolver, {(ORIGIN, "TXT"): ['"13335 | 1.1.1.0/24"', '"13335 | 1.1.0.0/16"']}) + result = _run("http://1.1.1.1") + endpoint = _objects(result, "ip-port")[0] + assert len(endpoint["ObjectReference"]) == 2 + assert len(_objects(result, "asn")) == 1 + assert set(_values(_objects(result, "asn")[0], "subnet-announced")) == {"1.1.1.0/24", "1.1.0.0/16"} + + +@pytest.mark.parametrize("txt", ["NA | 1.1.1.0/24", "13335 | 2.2.2.0/24", "13335 | not-a-subnet", "oops"]) +def test_malformed_asn_records_keep_endpoint(resolver, txt): + _answers(resolver, {(ORIGIN, "TXT"): [f'"{txt}"']}) + result = _run("http://1.1.1.1") + assert not _objects(result, "asn") + assert "ASN lookup unavailable" in _objects(result, "ip-port")[0]["comment"] + + +def test_private_ipv4_keeps_endpoint_without_asn_query(resolver): + _answers(resolver, {("example.com.", "A"): ["10.0.0.1"]}) + result = _run() + assert "not globally routable" in _objects(result, "ip-port")[0]["comment"] + assert all(call.args[1] != "TXT" for call in resolver.resolve.call_args_list) + + +@pytest.mark.parametrize("address", ["224.0.0.1", "239.255.255.250"]) +@pytest.mark.parametrize("literal", [True, False]) +def test_multicast_ipv4_keeps_endpoint_without_asn_query(resolver, address, literal): + _answers(resolver, {("example.com.", "A"): [address]}) + result = _run(f"https://{address}/" if literal else "https://example.com/") + endpoint = _objects(result, "ip-port")[0] + assert _values(endpoint, "ip") == [address] + assert _values(endpoint, "dst-port") == ["443"] + assert (SOURCE_UUID, "derived-from") in _references(endpoint) + assert not _objects(result, "asn") + assert "not globally routable" in endpoint["comment"] + assert "ASN lookup unavailable" not in endpoint["comment"] + assert all(call.args[1] != "TXT" for call in resolver.resolve.call_args_list) + + +def test_custom_nameserver_and_timeouts(resolver): + _answers(resolver, {("example.com.", "AAAA"): ["2001:db8::1"]}) + assert "results" in _run(config={"nameserver": "9.9.9.9"}) + assert resolver.nameservers == ["9.9.9.9"] + assert resolver.timeout == 2 + assert resolver.lifetime == 5 + + +@pytest.mark.parametrize("config", [{"nameserver": "bad"}, "bad", ["bad"]]) +def test_bad_config_returns_error(resolver, config): + assert "error" in _run(config=config) + resolver.resolve.assert_not_called() + + +@pytest.mark.parametrize("query", ["{", "null", "[]", "{}", '{"attribute": []}', '{"attribute": {"type": "url"}}']) +def test_bad_requests(query): + assert "error" in url_dns.handler(query) + + +def test_unsupported_type_and_invalid_uuid(resolver): + for attribute in ( + {"type": "domain", "value": "example.com", "uuid": SOURCE_UUID}, + {"type": "url", "value": "http://example.com", "uuid": "invalid"}, + {"type": "url", "value": "http://example.com", "uuid": None}, + ): + assert "error" in _run(attribute=attribute) + resolver.resolve.assert_not_called() + + +def test_metadata_and_no_query(): + assert url_dns.handler() is False + assert url_dns.introspection() == {"input": ["url"], "format": "misp_standard"} + assert url_dns.version()["config"] == ["nameserver"] + assert url_dns.version()["module-type"] == ["expansion"]