From 4e700367d00ab59db42a0f3182199e257ee91c96 Mon Sep 17 00:00:00 2001 From: Alexandre Dulaunoy Date: Thu, 29 Oct 2026 18:09:42 +0100 Subject: [PATCH 1/2] feat: [dnstwist] add domain permutation enrichment --- documentation/expansion/dnstwist.md | 13 ++ misp_modules/modules/expansion/dnstwist.py | 207 +++++++++++++++++++++ tests/test_dnstwist.py | 107 +++++++++++ 3 files changed, 327 insertions(+) create mode 100644 documentation/expansion/dnstwist.md create mode 100644 misp_modules/modules/expansion/dnstwist.py create mode 100644 tests/test_dnstwist.py diff --git a/documentation/expansion/dnstwist.md b/documentation/expansion/dnstwist.md new file mode 100644 index 00000000..4b31a81e --- /dev/null +++ b/documentation/expansion/dnstwist.md @@ -0,0 +1,13 @@ +# DNSTwist Domain Enrichment + +Generate domain permutations locally with optional DNS resolution. Install dnstwist in the module server Python environment. resolve defaults to false and performs no DNS checks. Set resolve=true to return DNS-resolving candidates with available A, AAAA, NS and MX observations. timeout defaults to 30 seconds (1–120); threads defaults to 4 (1–16). An isolated subprocess prevents changes to the module server argument list and bounds runtime. Generated candidates and resolving domains are investigation leads, not maliciousness assertions. Each result preserves its mutation algorithm. + +Input: domain, hostname. + +Configuration: resolve, timeout, threads, max_results. Required: none. + +max_results defaults to 25 and accepts 1–100. New attributes use `to_ids=false`; the source UUID, detection flag and metadata are preserved. Objects reference the source attribute. Requests verify TLS, reject redirects and use connection/read timeouts of 5/30 seconds. API errors do not echo credentials. No automatic submission retries or background polling occur. + +Reference: https://github.com/elceef/dnstwist + +Install the optional engine with `python -m pip install dnstwist` in the Python environment running misp-modules. The module uses the same Python executable for its isolated worker. diff --git a/misp_modules/modules/expansion/dnstwist.py b/misp_modules/modules/expansion/dnstwist.py new file mode 100644 index 00000000..8faed8dd --- /dev/null +++ b/misp_modules/modules/expansion/dnstwist.py @@ -0,0 +1,207 @@ +import ipaddress +import json +import re +import zipfile +from uuid import UUID +from pymisp import MISPEvent, MISPObject +from pymisp.exceptions import PyMISPError + + +class ModuleError(ValueError): + """A diagnostic safe to show to the analyst.""" + + +def _integer(config, key, default, maximum=100): + value = config.get(key) + if value is None or value == "": + return default + if isinstance(value, bool) or not str(value).isdigit() or (not 1 <= int(value) <= maximum): + raise ModuleError(f"{key} must be an integer between 1 and {maximum}.") + return int(value) + + +def _enabled(config, key, default=False): + value = config.get(key) + if value is None or value == "": + return default + if isinstance(value, bool): + return value + if str(value).lower() not in ("true", "false", "1", "0"): + raise ModuleError(f"{key} must be true or false.") + return str(value).lower() in ("true", "1") + + +def _normalise(kind, value): + if not isinstance(value, str) or not value.strip() or any(ord(c) < 32 for c in value): + raise ModuleError("A nonempty string value is required.") + value = value.strip() + if kind in ("md5", "sha1", "sha256", "x509-fingerprint-sha256"): + length = {"md5": 32, "sha1": 40}.get(kind, 64) + if not re.fullmatch("[a-fA-F0-9]{%d}" % length, value): + raise ModuleError("Invalid hash.") + return value.lower() + if kind in ("ip-src", "ip-dst"): + return str(ipaddress.ip_address(value)) + if kind in ("domain", "hostname"): + value = value.rstrip(".").encode("idna").decode("ascii").lower() + if len(value) > 253 or not all( + re.fullmatch("[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?", p) for p in value.split(".") + ): + raise ModuleError("Invalid DNS name.") + if kind == "vulnerability": + value = value.upper() + if not re.fullmatch("CVE-[0-9]{4}-[0-9]{4,}", value): + raise ModuleError("A CVE identifier is required.") + return value + + +def _object(event, attribute, name): + obj = MISPObject(name) + obj.add_reference(attribute["uuid"], "describes") + event.add_object(obj) + return obj + + +def _annotation(event, attribute, title, data): + obj = _object(event, attribute, "annotation") + obj.comment = title + obj.add_attribute("text", value=json.dumps(data, sort_keys=True), to_ids=False, disable_correlation=True) + return obj + + +def handler(q=False): + if q is False: + return False + try: + request = json.loads(q) + if not isinstance(request, dict) or not isinstance(request.get("attribute"), dict): + raise ModuleError("An attribute containing type, value and uuid is required.") + attribute = request["attribute"] + UUID(attribute["uuid"]) + if attribute["type"] not in mispattributes["input"]: + raise ModuleError("Unsupported attribute type.") + value = _normalise(attribute["type"], attribute["value"]) + config = request.get("config") or {} + if not isinstance(config, dict): + raise ModuleError("The config must be an object.") + for key in REQUIRED: + if not isinstance(config.get(key), str) or not config[key].strip(): + raise ModuleError(f"Missing {key} configuration.") + limit = _integer(config, "max_results", 25) + event = MISPEvent() + event.add_attribute(**attribute) + _expand(event, attribute, value, config, limit) + result = json.loads(event.to_json()) + return {"results": {key: result.get(key, []) for key in ("Attribute", "Object")}} + except ModuleError as exc: + return {"error": str(exc)} + except ( + ValueError, + TypeError, + KeyError, + AttributeError, + IndexError, + OSError, + RuntimeError, + zipfile.BadZipFile, + PyMISPError, + ): + return {"error": "Invalid request or service response."} + + +def introspection(): + return mispattributes + + +def version(): + return dict(moduleinfo, config=moduleconfig) + + +moduleinfo = { + "version": "1.0", + "author": "MISP Project", + "name": "DNSTwist Domain Enrichment", + "description": ( + "Generate domain permutations locally with optional DNS resolution. Install dnstwist in the module server" + " Python environment. resolve defaults to false and performs no DNS checks. Set resolve=true to return" + " DNS-resolving candidates with available A, AAAA, NS and MX observations. timeout defaults to 30 seconds" + " (1–120); threads defaults to 4 (1–16). An isolated subprocess prevents changes to the module server argument" + " list and bounds runtime. Generated candidates and resolving domains are investigation leads, not" + " maliciousness assertions. Each result preserves its mutation algorithm." + ), + "module-type": ["expansion"], + "data-source": "external", + "logo": "", + "requirements": ["requests", "pymisp", "dnstwist"], + "features": ( + "Generate domain permutations locally with optional DNS resolution. Install dnstwist in the module server" + " Python environment. resolve defaults to false and performs no DNS checks. Set resolve=true to return" + " DNS-resolving candidates with available A, AAAA, NS and MX observations. timeout defaults to 30 seconds" + " (1–120); threads defaults to 4 (1–16). An isolated subprocess prevents changes to the module server argument" + " list and bounds runtime. Generated candidates and resolving domains are investigation leads, not" + " maliciousness assertions. Each result preserves its mutation algorithm." + ), + "references": ["https://github.com/elceef/dnstwist"], + "input": "domain, hostname", + "output": "MISP objects linked to the original attribute.", +} +mispattributes = {"input": ["domain", "hostname"], "format": "misp_standard"} +moduleconfig = ["resolve", "timeout", "threads", "max_results"] +REQUIRED = [] + + +def _expand(event, attribute, value, config, limit): + import subprocess + import sys + import tempfile + + timeout = _integer(config, "timeout", 30, 120) + threads = _integer(config, "threads", 4, 16) + resolve = _enabled(config, "resolve") + if resolve: + command = [ + sys.executable, + "-m", + "dnstwist", + "--format", + "json", + "--threads", + str(threads), + "--registered", + value, + ] + else: + program = ( + "import dnstwist,json,sys; f=dnstwist.Fuzzer(sys.argv[1]); f.generate(); print(json.dumps([dict(row) for" + " row in f.permutations()]))" + ) + command = [sys.executable, "-c", program, value] + with tempfile.TemporaryFile() as output: + try: + completed = subprocess.run(command, stdout=output, stderr=subprocess.DEVNULL, timeout=timeout, check=False) + except subprocess.TimeoutExpired: + raise ModuleError("DNSTwist exceeded timeout; narrow the domain or increase timeout.") from None + if completed.returncode: + raise ModuleError("DNSTwist failed; install the dnstwist package in the module server environment.") + output.seek(0, 2) + if output.tell() > 16 * 1024 * 1024: + raise ModuleError("DNSTwist output exceeded the size limit.") + output.seek(0) + raw = output.read().decode("utf-8") + rows = json.loads(raw) + if not isinstance(rows, list): + raise ModuleError("DNSTwist returned invalid results.") + rows = [row for row in rows if row.get("domain") != value] + for row in rows[:limit]: + candidate = _normalise("domain", row["domain"]) + obj = _object(event, attribute, "domain-ip") + obj.add_attribute("domain", value=candidate, to_ids=False) + obj.add_attribute("text", value=json.dumps({"fuzzer": row.get("fuzzer"), "dns_checked": resolve}), to_ids=False) + for key in ("dns_a", "dns_aaaa"): + for address in row.get(key, [])[:limit]: + obj.add_attribute("ip", value=str(ipaddress.ip_address(address)), to_ids=False) + for key in ("dns_ns", "dns_mx"): + if row.get(key): + _annotation(event, attribute, "DNSTwist DNS observation", {"domain": candidate, key: row[key]}) + if len(rows) > limit: + _annotation(event, attribute, "Result limit", {"truncated": True, "total": len(rows), "max_results": limit}) diff --git a/tests/test_dnstwist.py b/tests/test_dnstwist.py new file mode 100644 index 00000000..3643e927 --- /dev/null +++ b/tests/test_dnstwist.py @@ -0,0 +1,107 @@ +import json +from unittest.mock import Mock, patch + +import pytest +import requests + +from misp_modules.modules.expansion import dnstwist as module + +UUID = "32876e4a-916d-4935-9e51-7b69a76718f8" +SECRET = "test-service-secret" + + +def query(kind=None, value="example.com", config=None, **attribute_fields): + attribute = { + "type": kind or module.mispattributes["input"][0], + "value": value, + "uuid": UUID, + "to_ids": True, + "comment": "source context", + **attribute_fields, + } + return json.dumps( + {"attribute": attribute, "config": config if config is not None else {k: SECRET for k in module.REQUIRED}} + ) + + +def reply(payload=None, status=200): + return Mock(status_code=status, headers={}, json=Mock(return_value=payload)) + + +def assert_source(result): + assert "results" in result, result + source = next(a for a in result["results"]["Attribute"] if a["uuid"] == UUID) + assert source["to_ids"] is True + assert source["comment"] == "source context" + for obj in result["results"]["Object"]: + assert any(r["referenced_uuid"] == UUID for r in obj["ObjectReference"]) + assert all(a["to_ids"] is False for a in obj["Attribute"]) + + +def objects(result, name): + return [obj for obj in result["results"]["Object"] if obj["name"] == name] + + +def values(obj, relation): + return [a["value"] for a in obj["Attribute"] if a["object_relation"] == relation] + + +def test_discovery(): + assert module.handler(False) is False + assert module.introspection()["format"] == "misp_standard" + assert module.version()["config"] == module.moduleconfig + + +@pytest.mark.parametrize("raw", ["{", "[]", "null", "{}", '{"attribute": []}']) +def test_invalid_envelope(raw): + with patch("subprocess.run") as request: + assert "error" in module.handler(raw) + request.assert_not_called() + + +@pytest.mark.parametrize("raw", [0, -1, True, "bad", 101]) +def test_invalid_result_limit(raw): + with pytest.raises(module.ModuleError): + module._integer({"max_results": raw}, "max_results", 25) + + +def test_local_generation_has_no_resolution_and_caps_results(): + import subprocess + + def run(command, **kwargs): + kwargs["stdout"].write( + json.dumps( + [{"domain": "example.com"}, {"domain": "a.example", "fuzzer": "addition"}, {"domain": "b.example"}] + ).encode() + ) + return subprocess.CompletedProcess(command, 0) + + with patch("subprocess.run", side_effect=run) as run: + result = module.handler(query(config={"max_results": 1})) + assert_source(result) + assert len(objects(result, "domain-ip")) == 1 + assert "-c" in run.call_args.args[0] + assert "--registered" not in run.call_args.args[0] + assert "shell" not in run.call_args.kwargs + + +def test_resolution_keeps_fuzzer_and_dns_evidence(): + import subprocess + + def run(command, **kwargs): + kwargs["stdout"].write( + json.dumps([{"domain": "a.example", "fuzzer": "addition", "dns_a": ["192.0.2.1"]}]).encode() + ) + return subprocess.CompletedProcess(command, 0) + + with patch("subprocess.run", side_effect=run): + result = module.handler(query(config={"resolve": True})) + assert_source(result) + assert values(objects(result, "domain-ip")[0], "ip") == ["192.0.2.1"] + + +def test_timeout_becomes_an_actionable_error(): + import subprocess + + with patch("subprocess.run", side_effect=subprocess.TimeoutExpired("dnstwist", 30)): + assert "timeout" in module.handler(query())["error"] From 52c053574baa8c1773a1148c6ce118b420c71f70 Mon Sep 17 00:00:00 2001 From: Alexandre Dulaunoy Date: Thu, 29 Oct 2026 23:25:29 +0100 Subject: [PATCH 2/2] fix: [dnstwist] address review findings --- misp_modules/modules/expansion/dnstwist.py | 15 +++++++++++++-- tests/test_dnstwist.py | 17 +++++++++++++++++ 2 files changed, 30 insertions(+), 2 deletions(-) diff --git a/misp_modules/modules/expansion/dnstwist.py b/misp_modules/modules/expansion/dnstwist.py index 8faed8dd..cf743939 100644 --- a/misp_modules/modules/expansion/dnstwist.py +++ b/misp_modules/modules/expansion/dnstwist.py @@ -1,3 +1,4 @@ +import idna import ipaddress import json import re @@ -43,7 +44,7 @@ def _normalise(kind, value): if kind in ("ip-src", "ip-dst"): return str(ipaddress.ip_address(value)) if kind in ("domain", "hostname"): - value = value.rstrip(".").encode("idna").decode("ascii").lower() + value = idna.encode(value.rstrip("."), uts46=True, transitional=False).decode("ascii").lower() if len(value) > 253 or not all( re.fullmatch("[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?", p) for p in value.split(".") ): @@ -199,7 +200,17 @@ def _expand(event, attribute, value, config, limit): obj.add_attribute("text", value=json.dumps({"fuzzer": row.get("fuzzer"), "dns_checked": resolve}), to_ids=False) for key in ("dns_a", "dns_aaaa"): for address in row.get(key, [])[:limit]: - obj.add_attribute("ip", value=str(ipaddress.ip_address(address)), to_ids=False) + try: + parsed_address = str(ipaddress.ip_address(address)) + except ValueError: + _annotation( + event, + attribute, + "DNSTwist DNS lookup failure", + {"domain": candidate, "record": key, "status": str(address)}, + ) + continue + obj.add_attribute("ip", value=parsed_address, to_ids=False) for key in ("dns_ns", "dns_mx"): if row.get(key): _annotation(event, attribute, "DNSTwist DNS observation", {"domain": candidate, key: row[key]}) diff --git a/tests/test_dnstwist.py b/tests/test_dnstwist.py index 3643e927..f6ab68ed 100644 --- a/tests/test_dnstwist.py +++ b/tests/test_dnstwist.py @@ -105,3 +105,20 @@ def test_timeout_becomes_an_actionable_error(): with patch("subprocess.run", side_effect=subprocess.TimeoutExpired("dnstwist", 30)): assert "timeout" in module.handler(query())["error"] + + +def test_idna2008_domain_identity(): + assert module._normalise("domain", "faß.de") == "xn--fa-hia.de" + + +def test_dns_failure_sentinels_preserve_other_results(): + import subprocess + + def run(command, **kwargs): + kwargs["stdout"].write(json.dumps([{"domain": "a.example", "dns_a": ["!ServFail", "192.0.2.1"]}]).encode()) + return subprocess.CompletedProcess(command, 0) + + with patch("subprocess.run", side_effect=run): + result = module.handler(query(config={"resolve": True})) + assert_source(result) + assert values(objects(result, "domain-ip")[0], "ip") == ["192.0.2.1"]