Stop resolver module faults from failing DNS queries

README.md

@@ -96,7 +96,8 @@ adds latency). A fully recursive DNS query can take tens to hundreds of millisec
 Tested on OpenBSD from 7.0 (Unbound 1.16, Python 3.8) to 7.9 (Unbound 1.26, Python 3.13).
 
 **Re-running an installer upgrades the code and leaves your configuration
-alone.** `/etc/pfui_firewall.yml`, `pfui_unbound.yml` and the resolver's own
+alone**, and the resolver installer offers to skip rebuilding Unbound, which is
+what a module-only upgrade wants. `/etc/pfui_firewall.yml`, `pfui_unbound.yml` and the resolver's own
 `pfui_unbound.conf` are kept as they are, with a timestamped backup taken and
 the shipped example named so you can diff it for keys added since. Only a first
 install, where no config exists yet, lays down the examples.

client-unbound/pfui_unbound.py

@@ -76,6 +76,9 @@ def logger(qstate):
 
     r = qstate.return_msg.rep
     q = qstate.return_msg.qinfo
+    if not r or not q:
+        log_info(f"Query: {qstate.qinfo.qname_str} carried no reply to log")
+        return
     log_info("-" * 100)
     log_info(
         f"Query: {qstate.qinfo.qname_str}, "
@@ -194,27 +197,34 @@ def read_rr(rep=None, qname_str="", from_cache=False):
         return False
 
 
-def udp_transmit(soc, data, ip, port, retry=1):
+def udp_transmit(soc, data, ip, port, retry=1, wait_for_ack=True):
+    """Send one datagram, optionally waiting for the firewall to acknowledge it.
+
+    Returns the acknowledgement received, or None. ACKUPDATE counts as one:
+    the datagrams can arrive in either order, and taking only ACKDATA meant a
+    firewall that had already updated PF was retransmitted to and then counted
+    a failure.
+    """
     tries = 0
-    msg = None  # retry may be 0, and the summary below reads this
+    msg = None
     while tries < retry:
+        tries += 1
         try:
             log_info(f"PFUIDNS: UDP Transmitting {len(data)} bytes")
             soc.sendto(data, (ip, port))
-        except TIMEOUT:
-            log_err(
-                f"PFUIDNS: UDP socket timeout {ip}:{port}"
-            )
         except Exception as e:
-            log_err(f"PFUIDNS: UDP socket exception {ip}:{port}, '{e}'")
-        msg = udp_receive(soc=soc, rcvbuf=40, retry=1)  # Wait pfui_firewall ack data
-        if msg == b"ACKDATA":  # 40
-            log_info(f"PFUIDNS: Received ACKDATA (transmit success): {msg}")
+            # No acknowledgement can follow a send that did not leave, so this
+            # does not pay the ACK timeout before retrying
+            log_err(f"PFUIDNS: UDP send failed {ip}:{port}, '{e}'")
+            continue
+        if not wait_for_ack:
+            return None
+        msg = udp_receive(soc=soc, rcvbuf=64, retry=1)
+        if msg in (b"ACKDATA", b"ACKUPDATE"):
+            log_info(f"PFUIDNS: Received {msg!r} (transmit success)")
             return msg
-        else:
-            log_info(f"PFUIDNS: Received message not ACKDATA: {msg}")
-        tries += 1
-    log_info(f"PFUIDNS: timeout udp_transmit: buff {msg}")
+        log_info(f"PFUIDNS: Received message not an acknowledgement: {msg!r}")
+    log_info(f"PFUIDNS: no acknowledgement from {ip}:{port}, last was {msg!r}")
 
 
 def udp_receive(soc, rcvbuf=1400, retry=1):
@@ -241,19 +251,21 @@ def udp_transmit_close(data, ip, port, blocking):
     # 3s each a down firewall blocked the resolver for ~2 minutes per query
     soc.settimeout(float(pfui_cfg["UDP_ACK_TIMEOUT"]))
 
-    # transmit pf firewall data
-    reply = udp_transmit(soc, data, ip, port, int(pfui_cfg["UDP_RETRY"]))
-    breaker_record(f"{ip}:{port}", ok=(reply == b"ACKDATA"))
-
-    # wait for pf firewall update
-    if blocking:  # Wait for secondary ACKUPDATE
-        msg = udp_receive(soc=soc, rcvbuf=42, retry=1)
-        if msg == b"ACKUPDATE":
-            log_info(
-                "PFUIDNS: Recv pfui_firewall Update ACK"
-            )
-        else:
-            log_info(f"PFUIDNS: Unexpected msg: {msg}")
+    # A cache report is fire-and-forget, so it waits for nothing: holding the
+    # answer for an acknowledgement is what BLOCKING asks for, and this is not it
+    reply = udp_transmit(
+        soc, data, ip, port, pfui_cfg["UDP_RETRY"], wait_for_ack=blocking
+    )
+    if blocking:
+        confirmed = reply == b"ACKUPDATE"
+        if not confirmed and reply == b"ACKDATA":
+            # ACKDATA only says the message decoded; the tables are updated by
+            # the time ACKUPDATE follows
+            msg = udp_receive(soc=soc, rcvbuf=64, retry=1)
+            confirmed = msg == b"ACKUPDATE"
+            if not confirmed:
+                log_err(f"PFUIDNS: {ip}:{port} did not confirm the update: {msg!r}")
+        breaker_record(f"{ip}:{port}", ok=confirmed)
 
     # close sender udp socket
     soc.close()
@@ -350,17 +362,25 @@ def stream_transmit_close(data, family, address, target, blocking):
     # keeps denying the traffic either way.
     try:
         if blocking and sent:  # Nothing to acknowledge if the send failed
-            reply = conn.recv(36)  # Wait for pfui_firewall to ACK
+            # Read until the firewall closes rather than taking one segment: a
+            # reply split in transit compared unequal to ACKUPDATE and was
+            # charged to the breaker as a refusal
+            reply = b""
+            while len(reply) < 36:
+                chunk = conn.recv(36 - len(reply))
+                if not chunk:
+                    break
+                reply += chunk
             breaker_record(target, ok=(reply == b"ACKUPDATE"))
             if reply != b"ACKUPDATE":
                 # The firewall replies with a reason when it refuses a message,
                 # e.g. a version skew that leaves the wire format mismatched
                 log_err(
                     f"PFUIDNS: {target} did not confirm the update: {reply!r}"
                 )
-        elif sent:
-            # Non-blocking: delivery is the only thing observable from here
-            breaker_record(target, ok=True)
+        # A non-blocking send records nothing: the firewall has not answered
+        # yet, and counting it a success cleared the failures the blocking path
+        # was accumulating, so the breaker could never open
     except TIMEOUT:
         breaker_record(target, ok=False)
         log_err(
@@ -490,19 +510,26 @@ def transmit_all(pfui_dict, blocking=True):
 def inplace_cache_callback(
     qinfo, qstate, rep, rcode, edns, opt_list_out, region, **kwargs
 ):
-    """pythonmod: Inplace callback function for cache responses."""
-    if pfui_cfg["LOGGING"] and pfui_cfg["LOG_LEVEL"] == "DEBUG":
-        log_info("pythonmod: cache_callback called - answering from cache.")
+    """pythonmod: Inplace callback function for cache responses.
 
-    if pfui_cfg["LOGGING"] and pfui_cfg["LOG_LEVEL"] == "DEBUG":
-        log_info(
-            f"Cache data - qinfo: {qinfo}, qstate: {qstate}, rep: {rep}, rcode: {rcode}, edns: {edns}, opt_list_out: {opt_list_out}, region: {region}"
-        )
+    Returns True, and never raises: an exception here discards the cache hit,
+    and the callback's return value is read as a boolean, so returning nothing
+    leaves an error pending for the next query to fail on.
+    """
+    try:
+        if pfui_cfg["LOGGING"] and pfui_cfg["LOG_LEVEL"] == "DEBUG":
+            log_info("pythonmod: cache_callback called - answering from cache.")
+            log_info(
+                f"Cache data - qinfo: {qinfo}, qstate: {qstate}, rep: {rep}, rcode: {rcode}, edns: {edns}, opt_list_out: {opt_list_out}, region: {region}"
+            )
 
-    if rep is not None:
-        pfui_msg = read_rr(rep, qinfo.qname_str, from_cache=True)
-        if pfui_msg:
-            transmit_all(pfui_msg, blocking=False)
+        if rep is not None:
+            pfui_msg = read_rr(rep, qinfo.qname_str, from_cache=True)
+            if pfui_msg:
+                transmit_all(pfui_msg, blocking=False)
+    except Exception as exc:
+        log_err(f"PFUIDNS: {describe_exception(exc)}")
+    return True
 
 
 def init(id, cfg):
@@ -532,6 +559,10 @@ def init_standard(id, env):
         )
 
     if not register_inplace_cb_reply_cache(inplace_cache_callback, env, id):
+        log_err(
+            "PFUIDNS: could not register the cache callback; cache hits will "
+            "whitelist nothing. Refusing to start."
+        )
         return False
     return True
 
@@ -562,6 +593,28 @@ def inform_super(id, qstate, superqstate, qdata):
     return True
 
 
+# pythonmod injects these into this module's globals. strmodulevent() is not
+# used to name an event: its binding rejects anything outside 'enum module_ev'
+# with an OverflowError, and a log line must not be able to fail.
+EVENT_NAMES = (
+    "MODULE_EVENT_NEW",
+    "MODULE_EVENT_PASS",
+    "MODULE_EVENT_REPLY",
+    "MODULE_EVENT_NOREPLY",
+    "MODULE_EVENT_CAPSFAIL",
+    "MODULE_EVENT_MODDONE",
+    "MODULE_EVENT_ERROR",
+)
+
+
+def event_name(event):
+    """Name for a pythonmod event value, or the value itself if unrecognised."""
+    for name in EVENT_NAMES:
+        if globals().get(name) == event:
+            return name
+    return f"event {event!r}"
+
+
 def describe_exception(exc):
     """One-line type, message and call site for an exception.
 
@@ -595,7 +648,13 @@ def operate(id, event, qstate, qdata):
     except Exception as exc:
         log_err(f"PFUIDNS: {describe_exception(exc)}")
         try:
-            qstate.ext_state[id] = MODULE_WAIT_MODULE
+            # A finished query must not be handed back for more work: the mesh
+            # reads MODULE_WAIT_MODULE as "advance to the next module", which
+            # re-enters one that has already run
+            if event == MODULE_EVENT_MODDONE:
+                qstate.ext_state[id] = MODULE_FINISHED
+            else:
+                qstate.ext_state[id] = MODULE_WAIT_MODULE
         except Exception:
             pass
         return True
@@ -604,11 +663,7 @@ def operate(id, event, qstate, qdata):
 def _operate(id, event, qstate, qdata):
 
     if pfui_cfg["LOGGING"]:
-        log_info(
-            "pythonmod: operate, id: {}, event {}".format(
-                str(id), str(strmodulevent(event))
-            )
-        )
+        log_info(f"pythonmod: operate, id: {id}, {event_name(event)}")
 
     if event == MODULE_EVENT_MODDONE:
         if pfui_cfg["LOGGING"]:
@@ -618,7 +673,7 @@ def _operate(id, event, qstate, qdata):
         pfui_msg = None
         if qstate.return_msg:
             if pfui_cfg["LOGGING"] and pfui_cfg["LOG_LEVEL"] == "DEBUG":
-                if qstate.return_msg.qinfo:
+                if qstate.return_msg.rep and qstate.return_msg.qinfo:
                     logger(qstate)
             if qstate.return_msg.rep:
                 pfui_msg = read_rr(qstate.return_msg.rep, qstate.qinfo.qname_str)
@@ -681,6 +736,28 @@ def load_config(location=CONFIG_LOCATION):
     cfg = safe_load(open(location)) or {}
     for key, value in CONFIG_DEFAULTS.items():
         cfg.setdefault(key, value)
+
+    # A key present but empty parses as None, which setdefault cannot replace
+    cfg["FIREWALLS"] = cfg["FIREWALLS"] or []
+    if not isinstance(cfg["FIREWALLS"], list):
+        raise ValueError(f"FIREWALLS must be a list, not {cfg['FIREWALLS']!r}")
+
+    # Coerced once here so no send path can be handed a string or a None: those
+    # raise from inside socket calls, where the failure is a resolver fault
+    # rather than the configuration error it really is
+    for key, cast in (
+        ("SOCKET_TIMEOUT", float),
+        ("UDP_ACK_TIMEOUT", float),
+        ("BREAKER_COOLOFF", float),
+        ("UDP_RETRY", int),
+        ("BREAKER_FAILURES", int),
+        ("DEFAULT_PORT", int),
+    ):
+        try:
+            cfg[key] = cast(cfg[key])
+        except (TypeError, ValueError):
+            raise ValueError(f"{key} must be a number, not {cfg[key]!r}") from None
+
     cfg["SOCKET_PROTO"] = str(cfg["SOCKET_PROTO"]).strip().upper()
     if cfg["SOCKET_PROTO"] not in ("TCP", "UDP"):
         raise ValueError(
@@ -698,6 +775,13 @@ def load_config(location=CONFIG_LOCATION):
                 f"FIREWALLS[{index}] sets both SOCKET ({socket_path}) and HOST "
                 f"({host}); one firewall is reached one way or the other"
             )
+        if host and fw.get("PORT") is not None:
+            try:
+                int(fw["PORT"])
+            except (TypeError, ValueError):
+                raise ValueError(
+                    f"FIREWALLS[{index}] PORT must be a number, not {fw['PORT']!r}"
+                ) from None
         if socket_path and not str(socket_path).startswith("/"):
             raise ValueError(
                 f"FIREWALLS[{index}] SOCKET must be an absolute path, "