diff --git a/misp_modules/modules/expansion/otilabs.py b/misp_modules/modules/expansion/otilabs.py new file mode 100644 index 00000000..5b527809 --- /dev/null +++ b/misp_modules/modules/expansion/otilabs.py @@ -0,0 +1,309 @@ +import json +import re +from datetime import datetime +from ipaddress import ip_address + +import requests +from pymisp import MISPAttribute, MISPEvent, MISPObject + +from . import check_input_attribute, checking_error, standard_error_message + +misperrors = {"error": "Error"} +mispattributes = {"input": ["domain", "hostname"], "format": "misp_standard"} +moduleinfo = { + "version": "1", + "author": "OTI Labs", + "description": ( + "An expansion module for the OTI Labs Domain Intelligence API (https://oti-labs.com/domain-intelligence-api)." + " One request returns WHOIS/RDAP registration, DNS records, the certificate the domain serves, live" + " subdomains with their IP addresses and the SPF, DMARC and DKIM records." + ), + "module-type": ["expansion", "hover"], + "name": "OTI Labs Domain Intelligence", + "logo": "", + "requirements": [ + "A RapidAPI key subscribed to the OTI Labs Domain Intelligence API (the free plan includes 1,000 requests a" + " month)." + ], + "features": ( + "The module takes a domain or a hostname and queries the /lookup endpoint of the OTI Labs Domain Intelligence" + " API.\n\nThe response is mapped to a whois object (registrar, creation, update and expiry dates," + " nameservers), an x509 object (issuer, subject, serial number, validity dates, SANs), a dns-record object" + " (MX, NS, TXT, SOA), a domain-ip object for the domain (A/AAAA records plus the SPF, DMARC and DKIM" + " results as text) and one domain-ip object per live subdomain with its IP address. The number of" + " subdomain objects is capped by the subdomain_limit setting (default 50)." + ), + "references": [ + "https://oti-labs.com/domain-intelligence-api", + "https://github.com/osiris-technical-institute/domain-intelligence-api", + ], + "input": "A domain or a hostname.", + "output": "whois, x509, dns-record and domain-ip objects describing the domain and its live subdomains.", +} + +# config fields that your code expects from the site admin +moduleconfig = ["apikey", "subdomain_limit"] + +API_HOST = "domain-intelligence-api.p.rapidapi.com" +API_URL = f"https://{API_HOST}/lookup/" +API_TIMEOUT = 30 +DEFAULT_SUBDOMAIN_LIMIT = 50 +SAN_LIMIT = 50 + + +def _datetime(value): + """Return the value if it is an ISO 8601 date, else None (some ccTLD registries return free text).""" + if not isinstance(value, str) or not value: + return None + try: + datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + return value + + +def _is_ip(value): + try: + ip_address(value) + except ValueError: + return False + return True + + +def _hostname(value): + """Lower-case a DNS name and drop the trailing dot; MX values also drop their priority.""" + if not isinstance(value, str) or not value.strip(): + return None + return value.split()[-1].rstrip(".").lower() or None + + +def _txt(value): + """Join the quoted strings of a TXT record into one value.""" + parts = re.findall(r'"((?:[^"\\]|\\.)*)"', value) + return "".join(parts) if parts else value + + +def _subdomain_limit(value): + try: + limit = int(value) + except (TypeError, ValueError): + return DEFAULT_SUBDOMAIN_LIMIT + return max(0, limit) + + +def _lookup(apikey, domain): + headers = {"X-RapidAPI-Key": apikey, "X-RapidAPI-Host": API_HOST, "Accept": "application/json"} + try: + response = requests.get(API_URL + domain, headers=headers, timeout=API_TIMEOUT) + except requests.exceptions.RequestException as e: + return None, "Error while querying the OTI Labs API: %s" % e + if response.status_code in (401, 403): + return None, ( + "The OTI Labs API rejected the key (HTTP %s). Use a RapidAPI key subscribed to the Domain Intelligence API." + % response.status_code + ) + if response.status_code == 429: + return None, "OTI Labs API quota or rate limit reached (HTTP 429)." + if response.status_code != 200: + return None, "Error while querying the OTI Labs API: HTTP %s" % response.status_code + try: + data = response.json() + except ValueError: + return None, "Invalid JSON received from the OTI Labs API" + if not isinstance(data, dict): + return None, "Unexpected response from the OTI Labs API" + return data, None + + +class OTILabsParser: + def __init__(self, attribute, subdomain_limit): + self.misp_event = MISPEvent() + self.attribute = MISPAttribute() + self.attribute.from_dict(**attribute) + self.misp_event.add_attribute(**self.attribute) + self.domain = attribute["value"].strip().lower().rstrip(".") + self.subdomain_limit = subdomain_limit + self.objects = 0 + + def parse(self, data): + sections = {} + for name in ("whois", "ssl", "dns", "email_security", "subdomains"): + section = data.get(name) + sections[name] = section if isinstance(section, dict) and "error" not in section else {} + self._parse_whois(sections["whois"]) + self._parse_ssl(sections["ssl"]) + self._parse_dns(sections["dns"], sections["email_security"]) + self._parse_subdomains(sections["subdomains"]) + + def results(self): + if not self.objects: + return {"error": "The OTI Labs API returned no data for %s." % self.domain} + event = json.loads(self.misp_event.to_json()) + return {"results": {key: event[key] for key in ("Attribute", "Object") if key in event}} + + def _add(self, misp_object, relationship="characterizes"): + misp_object.add_reference(self.attribute.uuid, relationship) + self.misp_event.add_object(misp_object) + self.objects += 1 + + def _parse_whois(self, whois): + if not whois: + return + whois_object = MISPObject("whois") + whois_object.add_attribute("domain", self.domain) + if whois.get("registrar"): + whois_object.add_attribute("registrar", whois["registrar"]) + for relation, key in ( + ("creation-date", "created"), + ("modification-date", "updated"), + ("expiration-date", "expires"), + ): + value = _datetime(whois.get(key)) + if value: + whois_object.add_attribute(relation, value) + for nameserver in whois.get("nameservers") or []: + nameserver = _hostname(nameserver) + if nameserver: + whois_object.add_attribute("nameserver", nameserver) + if whois.get("status"): + whois_object.add_attribute("text", "Status: %s" % ", ".join(whois["status"]), disable_correlation=True) + if len(whois_object.attributes) > 1: + self._add(whois_object) + + def _parse_ssl(self, ssl): + if not ssl: + return + x509_object = MISPObject("x509") + for relation, key in ( + ("issuer", "issuer"), + ("subject", "subject"), + ("serial-number", "serial_number"), + ("signature_algorithm", "signature_algorithm"), + ): + if ssl.get(key): + x509_object.add_attribute(relation, str(ssl[key])) + for relation, key in (("validity-not-before", "valid_from"), ("validity-not-after", "valid_to")): + value = _datetime(ssl.get(key)) + if value: + x509_object.add_attribute(relation, value) + for name in (ssl.get("sans") or [])[:SAN_LIMIT]: + if isinstance(name, str) and name and not name.startswith("*"): + x509_object.add_attribute("dns_names", name.lower()) + if ssl.get("issuer") or ssl.get("serial_number"): + self._add(x509_object) + + def _parse_dns(self, dns, email_security): + domain_object = MISPObject("domain-ip") + domain_object.add_attribute("domain", self.domain) + for address in (dns.get("A") or []) + (dns.get("AAAA") or []): + if _is_ip(address): + domain_object.add_attribute("ip", address) + for text in self._email_security_texts(email_security): + domain_object.add_attribute("text", text, disable_correlation=True) + if len(domain_object.attributes) > 1: + self._add(domain_object) + + record_object = MISPObject("dns-record") + record_object.add_attribute("queried-domain", self.domain) + for relation, key in (("mx-record", "MX"), ("ns-record", "NS"), ("soa-record", "SOA")): + for value in dns.get(key) or []: + name = _hostname(value.split()[0]) if key == "SOA" else _hostname(value) + if name: + record_object.add_attribute(relation, name) + for value in dns.get("TXT") or []: + record_object.add_attribute("txt-record", _txt(value), disable_correlation=True) + if dns.get("CAA"): + record_object.add_attribute("text", "CAA: %s" % "; ".join(dns["CAA"]), disable_correlation=True) + if len(record_object.attributes) > 1: + self._add(record_object) + + @staticmethod + def _email_security_texts(email_security): + texts = [] + spf = email_security.get("spf") + if isinstance(spf, dict): + records = spf.get("records") or [] + texts.append("SPF: %s" % (" | ".join(records) if records else "MISSING")) + dmarc = email_security.get("dmarc") + if isinstance(dmarc, dict): + records = dmarc.get("records") or [] + texts.append("DMARC: %s" % (" | ".join(records) if records else "MISSING")) + dkim = email_security.get("dkim") + if isinstance(dkim, dict): + found = dkim.get("found") or [] + if found: + texts.append("DKIM: found on selector(s) %s" % ", ".join(found)) + else: + texts.append("DKIM: not found on the common selectors tested") + return texts + + def _parse_subdomains(self, subdomains): + if not subdomains: + return + live = subdomains.get("live") or [] + added = 0 + for item in live: + if added >= self.subdomain_limit: + break + if not isinstance(item, dict): + continue + host = _hostname(item.get("host")) + if not host or host == self.domain: + continue + subdomain_object = MISPObject("domain-ip") + subdomain_object.add_attribute("hostname", host) + target = item.get("ip") + if isinstance(target, str) and _is_ip(target): + subdomain_object.add_attribute("ip", target) + elif isinstance(target, str) and target: + subdomain_object.add_attribute("text", "CNAME target: %s" % target, disable_correlation=True) + self._add(subdomain_object, "related-to") + added += 1 + count = subdomains.get("count") + if count is None: + return + live_count = subdomains.get("live_count") + if live_count is None: + # The first lookup of a domain returns before the API has resolved every name. + summary = ( + "OTI Labs found %s subdomains of %s; the live check is still running, query again shortly for" + " the live hosts and their IP addresses" % (count, self.domain) + ) + else: + summary = "OTI Labs found %s subdomains of %s, %s of them resolving now" % (count, self.domain, live_count) + if added < len(live): + summary += "; the first %s live ones were added" % added + self.misp_event.add_attribute(type="text", value=summary, category="Other", disable_correlation=True) + + +def handler(q=False): + if q is False: + return False + request = json.loads(q) + config = request.get("config") or {} + apikey = config.get("apikey") + if not apikey: + misperrors["error"] = "An OTI Labs API key (a RapidAPI key) is required." + return misperrors + attribute = request.get("attribute") + if not attribute or not check_input_attribute(attribute): + return {"error": "%s, %s." % (standard_error_message, checking_error)} + if attribute["type"] not in mispattributes["input"]: + return {"error": "Unsupported attribute type (expected domain or hostname)."} + domain = attribute["value"].strip().lower().rstrip(".") + data, error = _lookup(apikey, domain) + if error: + misperrors["error"] = error + return misperrors + parser = OTILabsParser(attribute, _subdomain_limit(config.get("subdomain_limit"))) + parser.parse(data) + return parser.results() + + +def introspection(): + return mispattributes + + +def version(): + moduleinfo["config"] = moduleconfig + return moduleinfo diff --git a/tests/test_otilabs.py b/tests/test_otilabs.py new file mode 100644 index 00000000..d9fda167 --- /dev/null +++ b/tests/test_otilabs.py @@ -0,0 +1,259 @@ +"""Unit tests for the OTI Labs Domain Intelligence MISP expansion module. + +The API is mocked with unittest.mock, so the tests run offline without an API key. The sample +response is a trimmed copy of a real /lookup response for stripe.com. +""" + +import copy +import json +from unittest.mock import MagicMock, patch + +from requests.exceptions import ConnectionError as RequestsConnectionError + +from misp_modules.modules.expansion import otilabs + +MOCK_ATTRIBUTE = {"type": "domain", "value": "stripe.com", "uuid": "4f1d2b6e-2a51-4c3d-9a2f-6a7d3c9e1b20"} +MOCK_CONFIG = {"apikey": "test_rapidapi_key"} + +SAMPLE_RESPONSE = { + "domain": "stripe.com", + "dns": { + "A": ["198.137.150.41", "198.202.176.41"], + "AAAA": [], + "MX": ["10 aspmx.l.google.com.", "20 alt1.aspmx.l.google.com."], + "TXT": [ + '"vercel-domain-verification-n462w8=JRePwTQbccpon6VAqYhidisHw"', + ( + '"v=spf1 ip4:198.2.180.60/32 ip4:13.111.2.227/32 include:spf1.stripe.com' + ' include:greenhouse-outbound-mail.stripe.com include:_spf.qualtrics.com ~all"' + ), + ], + "NS": ["ns-1087.awsdns-07.org.", "ns-705.awsdns-24.net."], + "CAA": ['0 issue "amazon.com"'], + "SOA": ["ns-1882.awsdns-43.co.uk. awsdns-hostmaster.amazon.com. 1 900 900 1209600 3600"], + }, + "ssl": { + "issuer": "CN=DigiCert G5 TLS ECC SHA384 2021 CA1,O=DigiCert\\, Inc.,C=US", + "subject": ( + "CN=stripe.com,O=Stripe\\, LLC,L=South San Francisco,ST=California,C=US,2.5.4.5=4675506,2.5.4.15=Private" + " Organization,1.3.6.1.4.1.311.60.2.1.2=Delaware,1.3.6.1.4.1.311.60.2.1.3=US" + ), + "valid_from": "2026-09-24T00:00:00+00:00", + "valid_to": "2026-12-10T23:59:59+00:00", + "days_until_expiry": 67, + "serial_number": "1939952089730671370756115214599670534", + "sans": ["stripe.com", "www.stripe.com"], + "signature_algorithm": "ecdsa-with-SHA384", + }, + "whois": { + "registrar": "SafeNames Ltd.", + "created": "1995-09-12T04:00:00Z", + "updated": "2025-10-01T01:39:51Z", + "expires": "2027-09-11T04:00:00Z", + "nameservers": ["NS-1087.AWSDNS-07.ORG", "NS-1882.AWSDNS-43.CO.UK"], + "status": ["client delete prohibited", "client transfer prohibited"], + "_source": "rdap.org", + }, + "subdomains": { + "count": 493, + "live_count": 283, + "returned": 3, + "subdomains": ["58.email.stripe.com", "59.email.stripe.com", "access.stripe.com"], + "live": [ + {"host": "58.email.stripe.com", "ip": "13.224.202.101"}, + {"host": "59.email.stripe.com", "ip": "13.35.107.110"}, + {"host": "access.stripe.com", "ip": "198.137.150.231"}, + ], + "pools": [], + }, + "email_security": { + "spf": { + "present": True, + "records": [ + "v=spf1 ip4:198.2.180.60/32 ip4:13.111.2.227/32 include:spf1.stripe.com" + " include:greenhouse-outbound-mail.stripe.com include:_spf.qualtrics.com ~all" + ], + }, + "dmarc": { + "present": True, + "records": [ + "v=DMARC1; p=reject; pct=100; fo=1; rua=mailto:dmarc-reports@stripe.com;" + " ruf=mailto:dmarc-forensics@stripe.com;" + ], + }, + "dkim": { + "found": ["google", "s1"], + "records": [ + {"selector": "google", "records": ["v=DKIM1; k=rsa; p=MIGfMA0GCSq..."]}, + {"selector": "s1", "records": ["v=DKIM1; k=rsa; p=MIGfMA0GCSq..."]}, + ], + "note": ( + "Probed common selectors (Google, Microsoft 365, Mailchimp, SendGrid, etc.). Custom or hash-based" + " selectors aren't auto-discoverable." + ), + }, + }, +} + + +def _query(attribute=None, config=None): + return json.dumps( + { + "attribute": attribute if attribute is not None else MOCK_ATTRIBUTE, + "config": config if config is not None else MOCK_CONFIG, + } + ) + + +def _response(status_code=200, payload=None): + response = MagicMock() + response.status_code = status_code + response.json.return_value = payload if payload is not None else SAMPLE_RESPONSE + return response + + +def _objects(result, name): + return [o for o in result["results"]["Object"] if o["name"] == name] + + +def _values(misp_object, relation): + return [a["value"] for a in misp_object["Attribute"] if a["object_relation"] == relation] + + +def test_introspection(): + info = otilabs.introspection() + assert info["input"] == ["domain", "hostname"] + assert info["format"] == "misp_standard" + + +def test_version(): + meta = otilabs.version() + assert meta["name"] == "OTI Labs Domain Intelligence" + assert meta["config"] == ["apikey", "subdomain_limit"] + + +def test_missing_apikey(): + result = otilabs.handler(_query(config={})) + assert "API key" in result["error"] + + +def test_unsupported_attribute_type(): + attribute = {"type": "ip-src", "value": "198.51.100.1", "uuid": MOCK_ATTRIBUTE["uuid"]} + result = otilabs.handler(_query(attribute=attribute)) + assert "Unsupported attribute type" in result["error"] + + +@patch("misp_modules.modules.expansion.otilabs.requests.get") +def test_domain_lookup(mock_get): + mock_get.return_value = _response() + result = otilabs.handler(_query()) + + args, kwargs = mock_get.call_args + assert args[0] == "https://domain-intelligence-api.p.rapidapi.com/lookup/stripe.com" + assert kwargs["headers"]["X-RapidAPI-Key"] == "test_rapidapi_key" + assert kwargs["headers"]["X-RapidAPI-Host"] == "domain-intelligence-api.p.rapidapi.com" + + (whois,) = _objects(result, "whois") + assert _values(whois, "registrar") == ["SafeNames Ltd."] + assert _values(whois, "nameserver") == ["ns-1087.awsdns-07.org", "ns-1882.awsdns-43.co.uk"] + assert _values(whois, "creation-date") + + (x509,) = _objects(result, "x509") + assert _values(x509, "serial-number") == ["1939952089730671370756115214599670534"] + assert _values(x509, "dns_names") == ["stripe.com", "www.stripe.com"] + + (records,) = _objects(result, "dns-record") + assert _values(records, "mx-record") == ["aspmx.l.google.com", "alt1.aspmx.l.google.com"] + assert _values(records, "soa-record") == ["ns-1882.awsdns-43.co.uk"] + assert any(v.startswith("v=spf1 ") for v in _values(records, "txt-record")) + + domain_objects = _objects(result, "domain-ip") + (apex,) = [o for o in domain_objects if _values(o, "domain") == ["stripe.com"]] + assert _values(apex, "ip") == ["198.137.150.41", "198.202.176.41"] + texts = _values(apex, "text") + assert any(t.startswith("SPF: v=spf1") for t in texts) + assert any("p=reject" in t for t in texts if t.startswith("DMARC:")) + assert "DKIM: found on selector(s) google, s1" in texts + + subdomains = [o for o in domain_objects if _values(o, "hostname")] + assert [_values(o, "hostname")[0] for o in subdomains] == [ + "58.email.stripe.com", + "59.email.stripe.com", + "access.stripe.com", + ] + assert _values(subdomains[2], "ip") == ["198.137.150.231"] + + summary = [a["value"] for a in result["results"]["Attribute"] if a["type"] == "text"] + assert summary == ["OTI Labs found 493 subdomains of stripe.com, 283 of them resolving now"] + + +@patch("misp_modules.modules.expansion.otilabs.requests.get") +def test_subdomain_limit(mock_get): + mock_get.return_value = _response() + result = otilabs.handler(_query(config={"apikey": "test_rapidapi_key", "subdomain_limit": "1"})) + subdomains = [o for o in _objects(result, "domain-ip") if _values(o, "hostname")] + assert len(subdomains) == 1 + summary = [a["value"] for a in result["results"]["Attribute"] if a["type"] == "text"] + assert summary[0].endswith("the first 1 live ones were added") + + +@patch("misp_modules.modules.expansion.otilabs.requests.get") +def test_live_check_still_running(mock_get): + payload = copy.deepcopy(SAMPLE_RESPONSE) + payload["subdomains"]["live_count"] = None + payload["subdomains"]["live"] = [] + mock_get.return_value = _response(payload=payload) + result = otilabs.handler(_query()) + assert not [o for o in _objects(result, "domain-ip") if _values(o, "hostname")] + summary = [a["value"] for a in result["results"]["Attribute"] if a["type"] == "text"] + assert "the live check is still running" in summary[0] + + +@patch("misp_modules.modules.expansion.otilabs.requests.get") +def test_failed_section_is_skipped(mock_get): + payload = copy.deepcopy(SAMPLE_RESPONSE) + payload["whois"] = {"error": "RDAP lookup failed"} + mock_get.return_value = _response(payload=payload) + result = otilabs.handler(_query()) + assert not _objects(result, "whois") + assert _objects(result, "x509") + + +@patch("misp_modules.modules.expansion.otilabs.requests.get") +def test_free_text_whois_date_is_dropped(mock_get): + payload = copy.deepcopy(SAMPLE_RESPONSE) + payload["whois"]["created"] = "before Aug-1996" + mock_get.return_value = _response(payload=payload) + result = otilabs.handler(_query()) + (whois,) = _objects(result, "whois") + assert not _values(whois, "creation-date") + assert _values(whois, "registrar") == ["SafeNames Ltd."] + + +@patch("misp_modules.modules.expansion.otilabs.requests.get") +def test_rejected_key(mock_get): + mock_get.return_value = _response(status_code=403) + result = otilabs.handler(_query()) + assert "rejected the key" in result["error"] + + +@patch("misp_modules.modules.expansion.otilabs.requests.get") +def test_quota_reached(mock_get): + mock_get.return_value = _response(status_code=429) + result = otilabs.handler(_query()) + assert "429" in result["error"] + + +@patch("misp_modules.modules.expansion.otilabs.requests.get") +def test_network_error(mock_get): + mock_get.side_effect = RequestsConnectionError("connection refused") + result = otilabs.handler(_query()) + assert "Error while querying the OTI Labs API" in result["error"] + + +@patch("misp_modules.modules.expansion.otilabs.requests.get") +def test_no_usable_data(mock_get): + payload = {name: {"error": "unavailable"} for name in ("whois", "ssl", "dns", "subdomains", "email_security")} + mock_get.return_value = _response(payload=payload) + result = otilabs.handler(_query()) + assert "no data" in result["error"]