feat(dns): emit events on malformed/headerless/question-parse-error packets

Silent drops on <12B packets, qdcount=0, and question-section ValueError gave
fuzzers and scanners a completely dark target.  New events malformed_packet,
empty_question_section, and question_parse_error fire at severity=5 so these
probes are visible without counting toward recon_burst.

decnet/templates/dns/server.py

@@ -4,13 +4,17 @@ DNS server (UDP+TCP/53) — BIND 9.x persona.
 
 event_type values emitted:
   query                  — standard resolution attempt
-  fingerprint_probe  — version.bind / hostname.bind / id.server CHAOS queries
+  fingerprint_probe      — version.bind / hostname.bind / id.server / opcode / flag / qclass probes
   zone_transfer          — AXFR or IXFR (always REFUSED)
   amp_probe              — qtype=ANY or EDNS requestor udp_size > 1232
   tunneling_suspect      — long high-entropy labels or rapid TXT burst from same src
   flood_suspect          — source exceeding QPS threshold within rolling window
   tracking_evicted       — LRU state evicted (signals IP-rotation evasion)
   recon_burst            — same source hit ≥2 distinct high-signal event types within 60s
+  malformed_packet       — wire bytes shorter than 12 (no DNS header possible)
+  empty_question_section — qdcount=0 (headerless keepalive / scanner probe)
+  question_parse_error   — question section could not be decoded
+  multi_question         — qdcount>1; only question 0 is answered
 """
 
 import asyncio
@@ -646,15 +650,21 @@ async def _dispatch(data: bytes, src_ip: str, src_port: int, transport: str) ->
 def _handle(data: bytes, src_ip: str, src_port: int, transport: str) -> bytes | None:
     """Parse one DNS request and return the response wire bytes, emitting events."""
     if len(data) < 12:
+        _log("malformed_packet", severity=5, src=src_ip, src_port=src_port,
+             transport=transport, length=len(data))
         return None
     qid, flags_in, qdcount, ancount, nscount, arcount = struct.unpack_from(">HHHHHH", data, 0)
     if qdcount == 0:
+        _log("empty_question_section", severity=5, src=src_ip, src_port=src_port,
+             transport=transport, qid=qid)
         return None
     rd = bool(flags_in & 0x0100)
 
     try:
         qname, qtype, qclass, _ = _parse_question(data, 12)
-    except ValueError:
+    except ValueError as exc:
+        _log("question_parse_error", severity=5, src=src_ip, src_port=src_port,
+             transport=transport, reason=str(exc)[:64])
         return None
 
     edns_size = _parse_edns_size(data, qdcount, ancount, nscount, arcount)

tests/service_testing/test_dns.py

@@ -712,3 +712,54 @@ class TestServiceRegistration:
         assert ctx is not None
         assert ctx.name == "dns"
         assert (ctx / "Dockerfile").exists()
+
+
+# ── Parse hygiene events ───────────────────────────────────────────────────────
+
+class TestParseHygiene:
+    def test_malformed_packet_too_short(self):
+        mod, events = _load_dns()
+        resp = mod._handle(b"\x00\x01\x00\x00", "1.2.3.4", 1234, "udp")
+        assert resp is None
+        ev = _events_of(events, "malformed_packet")
+        assert len(ev) == 1
+        assert ev[0]["src"] == "1.2.3.4"
+        assert ev[0]["length"] == 4
+        assert ev[0]["transport"] == "udp"
+
+    def test_malformed_packet_empty(self):
+        mod, events = _load_dns()
+        resp = mod._handle(b"", "10.0.0.1", 5353, "tcp")
+        assert resp is None
+        ev = _events_of(events, "malformed_packet")
+        assert len(ev) == 1
+        assert ev[0]["length"] == 0
+
+    def test_empty_question_section(self):
+        mod, events = _load_dns()
+        # 12-byte header with qdcount=0
+        pkt = struct.pack(">HHHHHH", 0xBEEF, 0x0100, 0, 0, 0, 0)
+        resp = mod._handle(pkt, "2.2.2.2", 53, "udp")
+        assert resp is None
+        ev = _events_of(events, "empty_question_section")
+        assert len(ev) == 1
+        assert ev[0]["qid"] == 0xBEEF
+        assert ev[0]["src"] == "2.2.2.2"
+
+    def test_question_parse_error_truncated(self):
+        mod, events = _load_dns()
+        # Header claims qdcount=1 but question section is empty
+        pkt = struct.pack(">HHHHHH", 0x0001, 0x0100, 1, 0, 0, 0)
+        resp = mod._handle(pkt, "3.3.3.3", 1053, "udp")
+        assert resp is None
+        ev = _events_of(events, "question_parse_error")
+        assert len(ev) == 1
+        assert ev[0]["src"] == "3.3.3.3"
+        assert "reason" in ev[0]
+
+    def test_question_parse_error_no_malformed_event(self):
+        """question_parse_error must not also emit malformed_packet."""
+        mod, events = _load_dns()
+        pkt = struct.pack(">HHHHHH", 0x0001, 0x0100, 1, 0, 0, 0)
+        mod._handle(pkt, "3.3.3.3", 1053, "udp")
+        assert len(_events_of(events, "malformed_packet")) == 0