Merge branch 'main' into pre-commit

boefjes/boefjes/plugins/kat_kat_finding_types/kat_finding_types.json

@@ -509,5 +509,12 @@
     "source": "Check the version of the host manually.",
     "impact": "Unknown. The server may or may not be vulnerable. OpenKAT is not able to determine the version.",
     "recommendation": "Verify manually if the software is up to date as OpenKAT is not able to determine the software version ."
+  },
+  "KAT-LAME-DELEGATION": {
+    "description": "The nameservers for this object are configured but cannot be reached or are non existent.",
+    "risk": "recommendation",
+    "source": "Check the nameservers of the host or iprange manually.",
+    "impact": "No resolving can be done for this IP or host. This might cause problems for mailservers or slow down connections.",
+    "recommendation": "Verify that the listed nameservers are reachable and have valid hostnames."
   }
 }

boefjes/boefjes/plugins/kat_rdns/main.py

@@ -1,4 +1,5 @@
 import dns
+from dns.edns import EDEOption
 from dns.resolver import Answer
 
 from boefjes.config import settings
@@ -9,14 +10,29 @@ def run(boefje_meta: BoefjeMeta) -> list[tuple[set, bytes | str]]:
     """return results to normalizer."""
     ip = boefje_meta.arguments["input"]["address"]
 
+    resolver = dns.resolver.Resolver()
+    # https://dnspython.readthedocs.io/en/stable/_modules/dns/edns.html
+    # enable EDE to get the ServFail return values if the server supports it # codespell-ignore
+    resolver.use_edns(options=[EDEOption(15)])
+    resolver.nameservers = [str(settings.remote_ns)]
+    reverse_ip = dns.reversename.from_address(ip)
     try:
-        resolver = dns.resolver.Resolver()
-        resolver.nameservers = [str(settings.remote_ns)]
-        reverse_ip = dns.reversename.from_address(ip)
         answer: Answer = resolver.resolve(reverse_ip, "PTR")
         result = f"RESOLVER: {answer.nameserver}\n{answer.response}"
         return [(set(), result)]
     except dns.resolver.NXDOMAIN:
         return [(set(), "NXDOMAIN")]
+    except dns.resolver.NoNameservers as error:
+        # no servers responded happily, we'll check the response from the first
+        # https://dnspython.readthedocs.io/en/latest/_modules/dns/rcode.html
+        # https://www.rfc-editor.org/rfc/rfc8914#name-extended-dns-error-code-6-d
+        first_error = error.kwargs["errors"][0]
+        if first_error[3] == "SERVFAIL":
+            for ede_error in first_error[4].options:
+                if int(ede_error.code) == 22:
+                    # Auth nameserver for ip could not be reached, error codes defined in RFC 8914
+                    return [(set(), "NoAuthServersReachable:" + ede_error.text)]
+                    # returned when the resolver indicates a Lame delegation.
+        return [(set(), "NoAnswer")]
     except (dns.resolver.Timeout, dns.resolver.NoAnswer):
         return [(set(), "NoAnswer")]

boefjes/boefjes/plugins/kat_rdns/normalize.py

@@ -7,22 +7,32 @@
 from octopoes.models import Reference
 from octopoes.models.ooi.dns.records import DNSPTRRecord
 from octopoes.models.ooi.dns.zone import Hostname
+from octopoes.models.ooi.findings import Finding, KATFindingType
 from octopoes.models.ooi.network import Network
 
 
 def run(input_ooi: dict, raw: bytes) -> Iterable[NormalizerOutput]:
-    ooi = Reference.from_str(input_ooi["primary_key"])
+    reference = Reference.from_str(input_ooi["primary_key"])
+
     answers = raw.decode()
     if answers == "NXDOMAIN" or answers == "NoAnswer":
         return
-    lines = [line for line in answers.split("\n") if not line.startswith("option")]
-    for rrset in from_text("\n".join(lines[1:])).answer:
-        for rr in rrset:
-            if isinstance(rr, PTR):
-                value = rrset.to_text()
-                hostname = Hostname(
-                    name=rr.to_text().rstrip("."), network=Network(name=input_ooi["network"]["name"]).reference
-                )
-                yield hostname
-                ptr_record = DNSPTRRecord(address=ooi, hostname=hostname.reference, value=value, ttl=rrset.ttl)
-                yield ptr_record
+    if answers.startswith("NoAuthServersReachable:"):
+        ft = KATFindingType(id="KAT-LAME-DELEGATION")
+        f = Finding(finding_type=ft.reference, ooi=reference, description=answers.split(":", 1)[1])
+        yield ft
+        yield f
+    else:
+        lines = [line for line in answers.split("\n") if not line.startswith("option")]
+        for rrset in from_text("\n".join(lines[1:])).answer:
+            for rr in rrset:
+                if isinstance(rr, PTR):
+                    value = rrset.to_text()
+                    hostname = Hostname(
+                        name=rr.to_text().rstrip("."), network=Network(name=input_ooi["network"]["name"]).reference
+                    )
+                    yield hostname
+                    ptr_record = DNSPTRRecord(
+                        address=reference, hostname=hostname.reference, value=value, ttl=rrset.ttl
+                    )
+                    yield ptr_record

rocky/reports/views/aggregate_report.py

@@ -121,7 +121,7 @@ def post(self, request, *args, **kwargs):
             messages.error(request, _("You do not have the required permissions to enable plugins."))
             return PostRedirect(self.get_previous())
 
-        client = get_katalogus(self.organization.code)
+        client = get_katalogus(self.organization_member)
         for selected_plugin in selected_plugins:
             try:
                 client.enable_boefje_by_id(selected_plugin)