Kat dns serverversion (#3291)

Co-authored-by: ammar92 <[email protected]>

boefjes/boefjes/plugins/kat_dns_version/boefje.json

@@ -0,0 +1,9 @@
+{
+  "id": "dns-bind-version",
+  "name": "DNS software version",
+  "description": "Uses the DNS VERSION.BIND command to attempt to learn the servers software.",
+  "consumes": [
+    "IPService"
+  ],
+  "scan_level": 2
+}

boefjes/boefjes/plugins/kat_dns_version/description.md

@@ -0,0 +1,3 @@
+# Fetch DNS Server software version
+
+This boefje tries to detect the DNS Server version by doing a VERSION.BIND call.

boefjes/boefjes/plugins/kat_dns_version/main.py

@@ -0,0 +1,42 @@
+"""Boefje script for getting namserver version"""
+
+import json
+from os import getenv
+
+import dns
+import dns.message
+import dns.query
+
+from boefjes.job_models import BoefjeMeta
+
+
+def run(boefje_meta: BoefjeMeta) -> list[tuple[set, str | bytes]]:
+    input_ = boefje_meta.arguments["input"]  # input is IPService
+    ip_port = input_["ip_port"]
+    if input_["service"]["name"] != "domain":
+        return [({"boefje/error"}, "Not a DNS service")]
+
+    ip = ip_port["address"]["address"]
+    port = int(ip_port["port"])
+    protocol = ip_port["protocol"]
+
+    timeout = float(getenv("TIMEOUT", 30))
+
+    method = dns.query.udp if protocol == "udp" else dns.query.tcp
+
+    queries = [
+        dns.message.make_query("VERSION.BIND", dns.rdatatype.TXT, dns.rdataclass.CHAOS),
+        dns.message.make_query("VERSION.SERVER", dns.rdatatype.TXT, dns.rdataclass.CHAOS),
+    ]
+
+    results = []
+    for query in queries:
+        response = method(query, where=ip, timeout=timeout, port=port)
+
+        try:
+            answer = response.answer[0]
+            results.append(answer.to_rdataset().pop().strings[0].decode())
+        except IndexError:
+            pass
+
+    return [(set(), json.dumps(results))]

boefjes/boefjes/plugins/kat_dns_version/normalize.py

@@ -0,0 +1,36 @@
+import json
+from collections.abc import Iterable
+
+from boefjes.job_models import NormalizerOutput
+from octopoes.models import Reference
+from octopoes.models.ooi.software import Software, SoftwareInstance
+
+
+def run(input_ooi: dict, raw: bytes) -> Iterable[NormalizerOutput]:
+    input_ooi_reference = Reference.from_str(input_ooi["primary_key"])
+
+    results = json.loads(raw)
+    for version in results:
+        if version.startswith("bind"):
+            name = "bind"
+            version_number = version.split("-")[1]
+        elif version.startswith("9."):
+            name = "bind"
+            version_number = version
+        elif version.startswith("Microsoft DNS"):
+            name = "Microsoft DNS"
+            version_number = version.replace("Microsoft DNS ", "").split(" ")[0]
+        elif version.startswith("dnsmasq"):
+            name = "dnsmasq"
+            version_number = version.split("-")[1]
+        elif version.startswith("PowerDNS"):
+            name = "PowerDNS"
+            version_number = version.replace("PowerDNS Authoritative Server ", "").split(" ")[0]
+        else:
+            name = None
+            version_number = None
+
+        if name and version_number:
+            software = Software(name=name, version=version_number)
+            software_instance = SoftwareInstance(ooi=input_ooi_reference, software=software.reference)
+            yield from [software, software_instance]

boefjes/boefjes/plugins/kat_dns_version/normalizer.json

@@ -0,0 +1,10 @@
+{
+  "id": "dns-bind-version-normalize",
+  "consumes": [
+    "boefje/dns-bind-version"
+  ],
+  "produces": [
+    "Software",
+    "SoftwareInstance"
+  ]
+}

boefjes/boefjes/plugins/kat_dns_version/schema.json

@@ -0,0 +1,13 @@
+{
+  "title": "Arguments",
+  "type": "object",
+  "properties": {
+    "TIMEOUT": {
+      "title": "TIMEOUT",
+      "type": "integer",
+      "description": "Timeout for requests to the targeted dns servers",
+      "default": 30,
+      "minimum": 0
+    }
+  }
+}

boefjes/tests/integration/test_api.py

@@ -47,9 +47,9 @@ def test_get_local_plugin(self):
 
     def test_filter_plugins(self):
         response = self.client.get(f"/v1/organisations/{self.org.id}/plugins/")
-        self.assertEqual(len(response.json()), 97)
+        self.assertEqual(len(response.json()), 99)
         response = self.client.get(f"/v1/organisations/{self.org.id}/plugins?plugin_type=boefje")
-        self.assertEqual(len(response.json()), 43)
+        self.assertEqual(len(response.json()), 44)
 
         response = self.client.get(f"/v1/organisations/{self.org.id}/plugins?limit=10")
         self.assertEqual(len(response.json()), 10)
@@ -74,7 +74,7 @@ def test_add_boefje(self):
         self.assertEqual(response.status_code, 422)
 
         response = self.client.get(f"/v1/organisations/{self.org.id}/plugins/?plugin_type=boefje")
-        self.assertEqual(len(response.json()), 44)
+        self.assertEqual(len(response.json()), 45)
 
         boefje_dict = boefje.dict()
         boefje_dict["consumes"] = list(boefje_dict["consumes"])
@@ -99,7 +99,7 @@ def test_add_normalizer(self):
         self.assertEqual(response.status_code, 201)
 
         response = self.client.get(f"/v1/organisations/{self.org.id}/plugins/?plugin_type=normalizer")
-        self.assertEqual(len(response.json()), 55)
+        self.assertEqual(len(response.json()), 56)
 
         response = self.client.get(f"/v1/organisations/{self.org.id}/plugins/test_normalizer")
         self.assertEqual(response.json(), normalizer.dict())