version 5.4.7

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

CHANGELOG.md

@@ -2,6 +2,22 @@
 
 All notable changes to this project are documented here, organized by release.
 
+## [5.4.7]
+
+### Fixed
+- clear all mypy errors
+- honour the legacy thresholds: config; repair the stale test suite
+- ignore datagrams that are not named HTB/PLG messages
+
+---
+
+## [5.4.6]
+
+### Added
+- re-resolve the server hostnames at the DNS TTL
+
+---
+
 ## [5.4.5]
 
 ### Added

CLAUDE.md

@@ -33,7 +33,7 @@ hbd passwd <username>
 
 Tests live in `tests/` (pytest, imported as a package). A second directory `test/` contains only TLS fixtures (`.pem` files), not test code.
 
-Line length: 111 characters (`black`, `flake8`, both configured).
+Line length: `black` and `flake8` both use 165 characters (set in `pyproject.toml`, `tox.ini` and `.flake8`).
 
 ## Architecture
 

README.md

@@ -20,7 +20,7 @@ A lightweight UDP-based host monitoring system. Monitored hosts run a client (`h
   └────────────────────┘              └────────────────────────────┘
 ```
 
-**Package:** `hbd` v5.4.5
+**Package:** `hbd` v5.4.7
 **Python:** 3.11+
 
 ### Subpackages
@@ -242,6 +242,7 @@ plugins:
 ### Connection behaviour
 
 - The client sends heartbeats over UDP to each server address resolved from the hostname (IPv4 and IPv6).
+- The server hostname is re-resolved every TTL seconds (the smallest A/AAAA TTL, at least 30s; 300s when the TTL cannot be determined, e.g. an IP literal), so an address change is followed without restarting the client. A failed lookup keeps the current addresses.
 - If a connection fails to open at startup, IPv6 connections are dropped after 3 consecutive failures. IPv4 connections retry indefinitely.
 - In daemon mode (`-d`), all log output goes to syslog (`LOG_DAEMON` facility).
 

hbd/__init__.py

@@ -14,4 +14,4 @@ Install options:
 """
 
 __all__ = ["__version__"]
-__version__ = "5.4.5"
+__version__ = "5.4.7"

hbd/client/__init__.py

@@ -1,3 +1,3 @@
 """HeartBeat Client (hbc) - System monitoring client."""
 
-from hbd import __version__
+from hbd import __version__  # noqa: F401

hbd/client/config.py

@@ -2,16 +2,16 @@
 
 import logging
 import os
-import logging
+from typing import Any, Dict
 
 logger = logging.getLogger(__name__)
 
 try:
     import yaml
 except Exception:
-    yaml = None
+    yaml = None  # type: ignore[assignment]
 
-CLIENT_DEFAULTS = {
+CLIENT_DEFAULTS: Dict[str, Any] = {
     # Network settings
     "hb_port": 50003,          # Port where hbd servers listen
     "interval": 10,             # Heartbeat interval in seconds

hbd/client/dnsttl.py

@@ -0,0 +1,170 @@
+"""Re-resolution of the server hostnames at the TTL of their A/AAAA records.
+
+getaddrinfo() hides the TTL, so a minimal stub-resolver query (resolv.conf nameservers, search
+list) is made only to learn it. The addresses themselves still come from getaddrinfo(). When
+the TTL cannot be learned (IP literal, /etc/hosts name, no resolv.conf, resolver timeout)
+DEFAULT_TTL is used.
+"""
+
+import ipaddress
+import logging
+import os
+import socket
+import struct
+from typing import List, Optional, Tuple
+
+logger = logging.getLogger("hbc.dns")
+
+DEFAULT_TTL = 300       # used when the real TTL is unknown
+MIN_TTL = 30            # a TTL of 0 must not turn into a resolution loop
+FAILED_RETRY = 60       # re-try delay after a failed resolution
+QUERY_TIMEOUT = 2.0
+
+TYPE_A, TYPE_CNAME, TYPE_AAAA = 1, 5, 28
+
+
+def parse_resolv_conf(path: str = "/etc/resolv.conf") -> Tuple[List[str], List[str]]:
+    """Return (nameservers, search domains)."""
+    servers: List[str] = []
+    search: List[str] = []
+    try:
+        with open(path) as f:
+            for line in f:
+                parts = line.split()
+                if len(parts) < 2:
+                    continue
+                if parts[0] == "nameserver":
+                    servers.append(parts[1].split("%")[0])
+                elif parts[0] in ("search", "domain"):
+                    search = parts[1:]
+    except OSError:
+        pass
+    return servers, search
+
+
+def build_query(name: str, qtype: int, qid: int) -> bytes:
+    out = struct.pack("!HHHHHH", qid, 0x0100, 1, 0, 0, 0)
+    for label in name.rstrip(".").split("."):
+        out += bytes([len(label)]) + label.encode("idna")
+    return out + b"\x00" + struct.pack("!HH", qtype, 1)
+
+
+def _skip_name(buf: bytes, off: int) -> int:
+    while True:
+        n = buf[off]
+        if n == 0:
+            return off + 1
+        if n & 0xC0:
+            return off + 2
+        off += n + 1
+
+
+def min_answer_ttl(resp: bytes, qid: int, qtype: int) -> Optional[int]:
+    """Smallest TTL among the answer RRs (CNAMEs included), or None if there is no qtype answer."""
+    try:
+        rid, flags, qd, an = struct.unpack("!HHHH", resp[:8])
+        if rid != qid or flags & 0x000F:
+            return None
+        off = 12
+        for _ in range(qd):
+            off = _skip_name(resp, off) + 4
+        ttls: List[int] = []
+        found = False
+        for _ in range(an):
+            off = _skip_name(resp, off)
+            rtype, _cls, ttl, rdlen = struct.unpack("!HHIH", resp[off:off + 10])
+            off += 10 + rdlen
+            if rtype in (qtype, TYPE_CNAME):
+                ttls.append(ttl)
+            found = found or rtype == qtype
+        return min(ttls) if found else None
+    except (struct.error, IndexError):
+        return None
+
+
+def _query_ttl(server: str, name: str, qtype: int) -> Optional[int]:
+    qid = struct.unpack("!H", os.urandom(2))[0]
+    try:
+        af = socket.AF_INET6 if ":" in server else socket.AF_INET
+        with socket.socket(af, socket.SOCK_DGRAM) as s:
+            s.settimeout(QUERY_TIMEOUT)
+            s.sendto(build_query(name, qtype, qid), (server, 53))
+            return min_answer_ttl(s.recv(4096), qid, qtype)
+    except (OSError, UnicodeError):
+        return None
+
+
+def lookup_ttl(host: str, af_filter: int = 0) -> Optional[int]:
+    """Blocking. Smallest A/AAAA TTL for host, or None when it cannot be determined."""
+    try:
+        ipaddress.ip_address(host)
+        return None
+    except ValueError:
+        pass
+    servers, search = parse_resolv_conf()
+    if host.endswith("."):
+        names = [host]
+    elif "." in host:
+        names = [host] + [f"{host}.{d}" for d in search]
+    else:
+        names = [f"{host}.{d}" for d in search] + [host]
+    qtypes = [t for t, af in ((TYPE_A, socket.AF_INET), (TYPE_AAAA, socket.AF_INET6))
+              if af_filter in (0, af)]
+    for name in names:
+        ttls = []
+        for qtype in qtypes:
+            for server in servers:
+                ttl = _query_ttl(server, name, qtype)
+                if ttl is not None:
+                    ttls.append(ttl)
+                    break
+        if ttls:
+            return min(ttls)
+    return None
+
+
+def refresh_delay(ttls: List[Optional[int]]) -> int:
+    """Seconds until the next resolution: the smallest known TTL, floored at MIN_TTL."""
+    known = [t for t in ttls if t is not None]
+    return max(min(known), MIN_TTL) if known else DEFAULT_TTL
+
+
+def resolve(hosts: List[str], port: int, af_filter: int) -> Optional[List[Tuple[int, str]]]:
+    """Blocking. Every (family, address) the hosts resolve to, deduplicated, or None if any lookup fails."""
+    found: List[Tuple[int, str]] = []
+    for host in hosts:
+        try:
+            infos = socket.getaddrinfo(host, port, af_filter, 0, socket.SOL_UDP)
+        except socket.gaierror as e:
+            logger.warning(f"Cannot resolve {host}: {e}")
+            return None
+        for info in infos:
+            entry = (info[0], str(info[4][0]))
+            if entry not in found:
+                found.append(entry)
+    return found or None
+
+
+def retarget(connections: list, resolved: List[Tuple[int, str]]) -> Tuple[List[Tuple[int, str]], list]:
+    """Point the connections at the resolved addresses, in place.
+
+    A connection whose address is gone is re-pointed at an address that is new, in list order, so
+    the first connection (the one plugin data uses) survives as long as anything resolves.
+    Returns (new addresses still needing a connection, connections to drop).
+    """
+    current = {c.addr for c in connections}
+    wanted = {addr for _, addr in resolved}
+    new = [(af, addr) for af, addr in resolved if addr not in current]
+    gone = [c for c in connections if c.addr not in wanted]
+    paired = min(len(gone), len(new))
+    for conn, (af, addr) in zip(gone, new):
+        logger.info(f"Server address {conn.addr} -> {addr}")
+        conn.set_target(addr, af)
+    drop = gone[paired:]
+    if connections and connections[0] in drop:
+        # plugin data goes out on the first connection: hand it a surviving address and drop that one instead
+        keep = next(c for c in connections if c not in gone)
+        connections[0].set_target(keep.addr, keep.af)
+        drop.remove(connections[0])
+        drop.append(keep)
+    return new[paired:], drop

hbd/client/main.py

@@ -16,7 +16,7 @@ import sys
 import time
 from logging.handlers import SysLogHandler
 from pathlib import Path
-from typing import Dict, List, Optional
+from typing import List, Optional
 
 # Import protocol and config
 from .config import load_config
@@ -25,6 +25,7 @@ from .. import __version__
 
 # Import plugin system
 from .plugin import PluginRegistry, PluginLoader, InfoPlugin, MonitorPlugin
+from . import dnsttl
 from .connectivity import Connectivity, set_shared
 
 # Constants
@@ -35,6 +36,7 @@ MAXRECV = 32767
 # Global state
 running = True
 dorestart = False
+send_shutdown = False
 shutdown_event: Optional[asyncio.Event] = None
 active_tasks: List[asyncio.Task] = []
 
@@ -96,6 +98,14 @@ class AsyncConnection:
             self.transport = None
             self.protocol = None
 
+    def set_target(self, addr: str, af: int):
+        """Re-point at a new server address; the socket is reopened on the next send if the family changed."""
+        if af != self.af:
+            self.close()
+            self.af = af
+        self.addr = addr
+        self.logger = logging.getLogger(f"hbc.conn.{addr}")
+
     async def sendto(self, msg: dict, msg_id: str = "HTB"):
         """Send a message to the server.
 
@@ -500,6 +510,48 @@ def stop():
             task.cancel()
 
 
+async def dns_refresher(connections: List[AsyncConnection], hosts: List[str], port: int, af_filter: int,
+                        name: str, interval: int, connectivity: Optional[Connectivity], next_id: int):
+    """Re-resolve the server hosts every TTL seconds and follow address changes.
+
+    A failed lookup leaves the connections untouched. Addresses that disappear are re-pointed at
+    new ones where possible; the rest are retired, and surplus new ones get a connection.
+    """
+    logger = logging.getLogger("hbc.dns")
+    loop = asyncio.get_running_loop()
+    delay = await loop.run_in_executor(None, _ttl_delay, hosts, af_filter)
+    while running:
+        try:
+            if shutdown_event:
+                await asyncio.wait_for(shutdown_event.wait(), timeout=delay)
+                break
+            await asyncio.sleep(delay)
+        except asyncio.TimeoutError:
+            pass
+        resolved = await loop.run_in_executor(None, dnsttl.resolve, hosts, port, af_filter)
+        if resolved is None:
+            delay = dnsttl.FAILED_RETRY
+            continue
+        new, gone = dnsttl.retarget(connections, resolved)
+        for conn in gone:
+            if len(connections) > 1:
+                logger.info(f"Dropping server address {conn.addr}")
+                conn._dead = True
+                conn.close()
+                connections.remove(conn)
+        for af, addr in new:
+            conn = AsyncConnection(next_id, addr, port, af, name)
+            next_id += 1
+            logger.info(f"Adding server address {addr}")
+            connections.append(conn)
+            active_tasks.append(asyncio.create_task(heartbeat_sender(conn, interval, connectivity)))
+        delay = await loop.run_in_executor(None, _ttl_delay, hosts, af_filter)
+
+
+def _ttl_delay(hosts: List[str], af_filter: int) -> int:
+    return dnsttl.refresh_delay([dnsttl.lookup_ttl(h, af_filter) for h in hosts])
+
+
 async def cleanup(connections: List[AsyncConnection]):
     """Cleanup connections on shutdown."""
     logger = logging.getLogger("hbc.cleanup")
@@ -521,7 +573,7 @@ async def cleanup(connections: List[AsyncConnection]):
 
 async def async_main(args, config):
     """Async main function."""
-    global running, shutdown_event, active_tasks, send_shutdown, allow_remote_command
+    global shutdown_event, active_tasks, send_shutdown, allow_remote_command
 
     # Create shutdown event
     shutdown_event = asyncio.Event()
@@ -640,6 +692,9 @@ async def async_main(args, config):
         task = asyncio.create_task(heartbeat_sender(conn, interval, connectivity))
         active_tasks.append(task)
 
+    active_tasks.append(asyncio.create_task(
+        dns_refresher(connections, hb_hosts, hb_port, af_filter, iam, interval, connectivity, conn_id)))
+
     # Plugin collector (uses all connections, but we'll use first one)
     if connections and registry.get_enabled():
         task = asyncio.create_task(plugin_collector(connections[0], registry))
@@ -784,7 +839,6 @@ def build_parser():
 
 def main(argv=None):
     """Main entry point."""
-    global running, dorestart
 
     parser = build_parser()
     args = parser.parse_args(argv)

hbd/client/plugin.py

@@ -17,7 +17,7 @@ import logging
 import sys
 from abc import ABC, abstractmethod
 from pathlib import Path
-from typing import Any, Dict, List, Optional, Type
+from typing import Any, Dict, List, Optional
 
 
 class Plugin(ABC):
@@ -256,7 +256,7 @@ class PluginRegistry:
         """Get all enabled plugins."""
         return [p for p in self._plugins.values() if p.enabled]
 
-    def get_by_type(self, plugin_type: Type[Plugin]) -> List[Plugin]:
+    def get_by_type(self, plugin_type: type) -> List[Plugin]:
         """Get all plugins of a specific type.
 
         Args:

hbd/client/plugins/cpu_monitor.py

@@ -5,8 +5,6 @@ load average, and process counts.
 """
 
 from typing import Any, Dict, Optional
-import sys
-from pathlib import Path
 
 # Import from parent package
 from hbd.client.plugin import MonitorPlugin
@@ -30,7 +28,7 @@ class CPUMonitorPlugin(MonitorPlugin):
 
     def __init__(self, config: Optional[Dict[str, Any]] = None):
         super().__init__(config)
-        self.psutil = None
+        self.psutil: Any = None
         self.per_core = config.get("per_core", False) if config else False
         self.interval = config.get("interval", 300) if config else 300
 

hbd/client/plugins/disk_monitor.py

@@ -5,12 +5,12 @@ Collects disk usage and I/O statistics using psutil.
 """
 
 import logging
-from typing import Dict, Any, Optional, List
+from typing import Dict, Any, Optional
 
 try:
     import psutil
 except ImportError:
-    psutil = None
+    psutil = None  # type: ignore[assignment]
 
 from hbd.client.plugin import MonitorPlugin
 
@@ -58,7 +58,7 @@ class DiskMonitorPlugin(MonitorPlugin):
             raise ImportError("psutil library is required for disk_monitor plugin")
 
         # Store previous I/O counters for delta calculation
-        self._prev_io = {}
+        self._prev_io: Dict[str, Any] = {}
 
     async def initialize(self):
         """Initialize the plugin (check psutil availability)."""
@@ -114,7 +114,7 @@ class DiskMonitorPlugin(MonitorPlugin):
 
     async def _collect_metrics(self) -> Dict[str, Any]:
         """Collect disk metrics from psutil."""
-        metrics = {}
+        metrics: Dict[str, Any] = {}
 
         # Collect partition usage
         partitions_data = {}

hbd/client/plugins/filesystem_info.py

@@ -10,7 +10,7 @@ from typing import Dict, Any, Optional
 try:
     import psutil
 except ImportError:
-    psutil = None
+    psutil = None  # type: ignore[assignment]
 
 from hbd.client.plugin import InfoPlugin
 
@@ -98,8 +98,8 @@ class FilesystemInfoPlugin(InfoPlugin):
 
     async def _collect_info(self) -> Dict[str, Any]:
         """Collect filesystem information from psutil."""
-        info = {}
-        filesystems = []
+        info: Dict[str, Any] = {}
+        filesystems: list = []
         filesystem_types = set()
 
         # Get mounted disk partitions
@@ -112,7 +112,7 @@ class FilesystemInfoPlugin(InfoPlugin):
             if partition.fstype in self.exclude_types:
                 continue
 
-            fs_info = {
+            fs_info: Dict[str, Any] = {
                 'device': partition.device,
                 'mountpoint': partition.mountpoint,
                 'fstype': partition.fstype,

hbd/client/plugins/memory_monitor.py

@@ -10,7 +10,7 @@ from typing import Dict, Any, Optional
 try:
     import psutil
 except ImportError:
-    psutil = None
+    psutil = None  # type: ignore[assignment]
 
 from hbd.client.plugin import MonitorPlugin
 
@@ -32,6 +32,7 @@ def _zfs_arc_bytes() -> int:
         pass
     return 0
 
+
 logger = logging.getLogger(__name__)
 
 
@@ -115,7 +116,7 @@ class MemoryMonitorPlugin(MonitorPlugin):
 
     async def _collect_metrics(self) -> Dict[str, Any]:
         """Collect memory metrics from psutil."""
-        metrics = {}
+        metrics: Dict[str, Any] = {}
 
         # Virtual (physical) memory statistics
         vmem = psutil.virtual_memory()

hbd/client/plugins/nagios_runner.py

@@ -124,7 +124,7 @@ class NagiosRunnerPlugin(MonitorPlugin):
         Returns:
             Dictionary with results from all plugins
         """
-        results = {}
+        results: Dict[str, Any] = {}
 
         for cmd_config in self.commands:
             name = cmd_config.get("name")
@@ -181,7 +181,8 @@ class NagiosRunnerPlugin(MonitorPlugin):
                 self.logger.error(f"Command timed out: {command}")
                 return NAGIOS_UNKNOWN, f"Command timed out after {self.timeout}s", {}
 
-            status_code = proc.returncode
+            # communicate() has finished, so the process has exited and returncode is set
+            status_code = proc.returncode if proc.returncode is not None else NAGIOS_UNKNOWN
 
             if status_code < 0:
                 return NAGIOS_UNKNOWN, f"Process killed by signal {-status_code}", {}
@@ -225,7 +226,7 @@ class NagiosRunnerPlugin(MonitorPlugin):
         Returns:
             Dictionary of metric_name: value
         """
-        perfdata = {}
+        perfdata: Dict[str, Any] = {}
 
         # Performance data comes after the pipe character
         if '|' not in output:

hbd/client/plugins/network_monitor.py

@@ -5,12 +5,12 @@ Collects network interface statistics and connection information using psutil.
 """
 
 import logging
-from typing import Dict, Any, Optional, List
+from typing import Dict, Any, Optional
 
 try:
     import psutil
 except ImportError:
-    psutil = None
+    psutil = None  # type: ignore[assignment]
 
 from hbd.client.plugin import MonitorPlugin
 from hbd.client.connectivity import get_shared
@@ -63,7 +63,7 @@ class NetworkMonitorPlugin(MonitorPlugin):
             raise ImportError("psutil library is required for network_monitor plugin")
 
         # Store previous I/O counters for delta calculation
-        self._prev_io = {}
+        self._prev_io: Dict[str, Any] = {}
 
     async def initialize(self):
         """Initialize the plugin (check psutil availability)."""
@@ -124,7 +124,7 @@ class NetworkMonitorPlugin(MonitorPlugin):
 
     async def _collect_metrics(self) -> Dict[str, Any]:
         """Collect network metrics from psutil."""
-        metrics = {}
+        metrics: Dict[str, Any] = {}
 
         # Collect per-interface I/O counters
         try:
@@ -169,7 +169,7 @@ class NetworkMonitorPlugin(MonitorPlugin):
         if self.include_connections:
             try:
                 connections = psutil.net_connections(kind='inet')
-                conn_stats = {}
+                conn_stats: Dict[str, int] = {}
 
                 # Count connections by state
                 for conn in connections:

hbd/client/plugins/os_info.py

@@ -5,7 +5,6 @@ kernel, architecture, and distribution details.
 """
 
 import platform
-import sys
 from pathlib import Path
 from typing import Any, Dict, Optional
 

hbd/client/plugins/ping_monitor.py

@@ -67,6 +67,7 @@ class PingMonitorPlugin(MonitorPlugin):
         self.timeout = int(cfg.get("timeout", 5))
         # hosts: dict of {hostname: {warning: x, critical: y}} or list of hostnames
         raw_hosts = cfg.get("hosts", {})
+        self.hosts: Dict[str, Any]
         if isinstance(raw_hosts, list):
             self.hosts = {h: {} for h in raw_hosts}
         else:

hbd/common/__init__.py

@@ -1,3 +1,3 @@
 """Common utilities shared between hbc and hbd."""
 
-from hbd import __version__
+from hbd import __version__  # noqa: F401

hbd/common/proto.py

@@ -8,7 +8,7 @@ Message Types:
     PLG: Plugin data message (client -> server)
 """
 
-from typing import Dict, Any, Union
+from typing import Dict, Any
 import json
 import zlib
 
@@ -92,7 +92,7 @@ def stodict(msg: bytes):
     '!' and decodes accordingly. Returns a dict with key 'ID' set to the
     message ID and the parsed key/value pairs.
     """
-    d = {}
+    d: Dict[str, Any] = {}
     if len(msg) > 0 and chr(msg[0]) == "!":
         # message is: b'!ID:' + compressed_payload
         # original code used msg[1:4].decode() for ID (3 bytes including colon)
@@ -159,4 +159,3 @@ def decode_plugin_data(msg: bytes) -> Dict[str, Any]:
         Dictionary with 'ID', 'plugin', and plugin data fields
     """
     return stodict(msg)
-

hbd/server/__init__.py

@@ -1,3 +1,3 @@
 """HeartBeat Daemon (hbd) - Server/daemon component."""
 
-from hbd import __version__
+from hbd import __version__  # noqa: F401

hbd/server/cli.py

@@ -120,7 +120,7 @@ def cmd_passwd(args):
 def cmd_notify(args):
     """Send a test message via a single notification channel."""
     from .config import load_config
-    from .notify import Notification, _dispatch_to_channel, setup
+    from .notify import Notification, setup
 
     config = load_config(args.configfile)
     setup(config)
@@ -193,7 +193,9 @@ def _read_pid(configfile) -> int | None:
 
 
 def cmd_stop(args):
-    import os, signal as _signal, time
+    import os
+    import signal as _signal
+    import time
     pid = _read_pid(args.configfile)
     if pid is None:
         sys.exit(1)
@@ -212,7 +214,8 @@ def cmd_stop(args):
 
 
 def cmd_reload(args):
-    import os, signal as _signal
+    import os
+    import signal as _signal
     pid = _read_pid(args.configfile)
     if pid is None:
         sys.exit(1)
@@ -222,7 +225,10 @@ def cmd_reload(args):
 
 
 def cmd_restart(args):
-    import os, signal as _signal, time, subprocess
+    import os
+    import signal as _signal
+    import time
+    import subprocess
     pid = _read_pid(args.configfile)
     if pid is not None:
         print(f"Stopping hbd (pid {pid})...")

hbd/server/config.py

@@ -7,7 +7,7 @@ import os
 try:
     import yaml
 except Exception:
-    yaml = None
+    yaml = None  # type: ignore[assignment]
 
 SERVER_DEFAULTS = {
     # Network settings
@@ -205,9 +205,6 @@ class ReloadableConfig:
                 # Load new config
                 new_config = load_config(path)
 
-                # Store old config for rollback if needed
-                old_config = self._config
-                
                 # Update config
                 self._config = new_config
                 self._logger.info(f"Configuration reloaded from {path}")

hbd/server/configio.py

@@ -15,6 +15,7 @@ def _make_yaml() -> YAML:
     y.preserve_quotes = True
     return y
 
+
 # Top-level keys managed by the 'server' logical section
 _SERVER_KEYS = [
     "hbd_port", "hbd_host", "ws_port", "wss_port", "hb_port",

hbd/server/data.py

@@ -1,4 +1,6 @@
-msgs = []  # in-memory list of recent messages for new websocket clients; also logged to file via notify.eventlog
+msgs: list = []  # in-memory list of recent messages for new websocket clients; also logged to file via notify.eventlog
+
+
 class Data:
 
     def __init__(self, config):

hbd/server/dns.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 from subprocess import Popen, PIPE, STDOUT
-from typing import Optional
+from typing import Any, Callable, Optional
 import asyncio
 import logging
 
@@ -79,8 +79,8 @@ async def dns_update_worker(
     hbdclass,
     cfg: dict,
     async_queue=None,
-    log: Optional[callable] = None,
-    pushmsg: Optional[callable] = None,
+    log: Optional[Callable[..., Any]] = None,
+    pushmsg: Optional[Callable[..., Any]] = None,
     loop: Optional[asyncio.AbstractEventLoop] = None,
 ):
     """Pure async DNS worker that processes updates from asyncio.Queue.
@@ -164,7 +164,7 @@ async def dns_update_worker(
 def start_dns_worker(
     hbdclass,
     cfg: dict,
-    log: Optional[callable] = None,
+    log: Optional[Callable[..., Any]] = None,
     loop: Optional[asyncio.AbstractEventLoop] = None,
 ):
     """Start the async DNS worker and return the Task.
@@ -176,7 +176,7 @@ def start_dns_worker(
         loop = asyncio.get_event_loop()
 
     # Create asyncio.Queue and wrap in a bridge for thread-safe puts
-    async_q = asyncio.Queue()
+    async_q: asyncio.Queue[Any] = asyncio.Queue()
 
     class _QueueBridge:
         """Thread-safe wrapper around asyncio.Queue for synchronous callers."""

hbd/server/hbdclass.py

@@ -8,6 +8,7 @@ import time
 import json
 import copy
 import queue
+from typing import Any, Dict
 
 num = 0
 
@@ -24,7 +25,7 @@ def log(host, m):
 class Connection:
     # map of addrs to names
 
-    htab = {}
+    htab: Dict[str, Any] = {}
     UNKNOWN = "unknown"
     UP = "up"
     DOWN = "down"
@@ -278,8 +279,8 @@ class Connection:
 #
 class Host:
     # Table of Hosts
-    hosts = {}
-    dnsQ = queue.Queue()
+    hosts: Dict[str, "Host"] = {}
+    dnsQ: "queue.Queue[Any]" = queue.Queue()
 
     def __init__(self, name):
         global num

hbd/server/http.py

@@ -109,6 +109,7 @@ def _insert_threshold_metric(thresholds: dict, metric_path: str, values: dict) -
         else:
             thresholds[plugin].setdefault(intermediate, {})[leaf] = cfg
 
+
 def _render_template(html_str: str, **context) -> str:
     tmpl = jinja2.Template(html_str)
     return tmpl.render(**context)
@@ -1132,7 +1133,6 @@ async def start(
                 elif is_mon:
                     monitored.append(hostname)
 
-
         # Resolve notification channel configs for display
         notif_channels = []
         if current_user:

hbd/server/journal.py

@@ -11,7 +11,7 @@ import os
 import asyncio
 from datetime import datetime
 from pathlib import Path
-from typing import Dict, Any, Optional, Callable, Iterable, List, Tuple, Union
+from typing import Dict, Any, Optional, Callable, Iterable, List, TextIO, Tuple, Union
 
 logger = logging.getLogger(__name__)
 
@@ -52,7 +52,7 @@ class MessageJournal:
         self.enabled = self.config.get('journal_enabled', True)
 
         # Runtime state
-        self._file_handle = None
+        self._file_handle: Optional[TextIO] = None
         self._current_size = 0
         self._lock = asyncio.Lock()
         self._initialized = False

hbd/server/main.py

@@ -5,8 +5,6 @@ import logging
 import socket
 import time
 import signal
-import sys
-import ssl
 from . import __version__
 
 from . import udp
@@ -23,6 +21,7 @@ eventlog = notify_mod.eventlog
 
 # shared runtime collections and helpers
 
+
 def save_state(config, hbdclass):
     """Save current state to pickle file. Safe to call at any time."""
     import pickle

hbd/server/monitor.py

@@ -8,8 +8,6 @@ This module can be extended for additional monitoring tasks.
 """
 
 from __future__ import annotations
-import asyncio
-import time
 from . import notify as notify_mod
 
 DROPOVERDUE = 7 * 24 * 3600
@@ -25,4 +23,3 @@ async def cleanup_connections(hbdclass):
         for conn_type, conn in host.connections.items():
             if hasattr(conn, 'cancel_overdue_timer'):
                 conn.cancel_overdue_timer()
-

hbd/server/notify.py

@@ -21,7 +21,7 @@ import smtplib
 import subprocess
 import time
 import sys
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from typing import Optional
 
 from . import data
@@ -49,6 +49,7 @@ logf = None
 
 _LEVEL_ORDER = {"RECOVER": 0, "INFO": 0, "WARNING": 1, "CRITICAL": 2}
 
+
 def _level_value(level: str) -> int:
     return _LEVEL_ORDER.get(level.upper(), 0)
 
@@ -103,7 +104,6 @@ def initlog(logfile):
 
 
 def closelog():
-    global logf
     if logf and logf != sys.stderr:
         try:
             logf.close()
@@ -331,8 +331,6 @@ async def _send_sms_voipms_async(channel_cfg: dict, notif: Notification) -> bool
         return False
 
 
-
-
 async def _send_matrix_async(channel_cfg: dict, notif: Notification) -> bool:
     """Send a Matrix message using matrix-nio."""
     try:
@@ -341,7 +339,6 @@ async def _send_matrix_async(channel_cfg: dict, notif: Notification) -> bool:
         logger.error("matrix-nio not installed; pip install matrix-nio")
         return False
 
-    from nio import AsyncClient
     homeserver = channel_cfg.get("homeserver", "")
     access_token = channel_cfg.get("access_token", "")
     room_id = channel_cfg.get("room_id", "")
@@ -359,7 +356,6 @@ async def _send_matrix_async(channel_cfg: dict, notif: Notification) -> bool:
     client = AsyncClient(homeserver)
     client.access_token = access_token
     try:
-        from nio import RoomSendResponse
         content = {
             "msgtype": "m.text",
             "body": text,

hbd/server/oauth.py

@@ -128,7 +128,7 @@ def get_providers(config: dict) -> list[ResolvedProvider]:
     Entries with missing required fields or unknown types are skipped with
     a warning log.  Order follows config declaration order.
     """
-    result = []
+    result: list = []
     oauth_cfg = config.get("oauth", {})
     if not isinstance(oauth_cfg, dict):
         return result

hbd/server/threshold.py

@@ -13,13 +13,14 @@ import asyncio
 import logging
 import time
 from enum import Enum
-from typing import Dict, List, Any, Optional, Tuple, Callable
+from typing import Dict, List, Any, Optional, Tuple
 from . import notify as notify_mod
 from .config import THRESHOLD_DEFAULTS
 
 logger = logging.getLogger(__name__)
 eventlog = notify_mod.eventlog
 
+
 class AlertLevel(Enum):
     """Alert severity levels."""
     OK = 0
@@ -52,16 +53,16 @@ class AlertState:
         self.metric_path = metric_path
         self.level = AlertLevel.OK
         self.since = time.time()
-        self.last_value = None
+        self.last_value: Any = None
         self.last_check = time.time()
         self.notification_count = 0
-        self.last_notification = None
-        self.threshold_value = None  # The threshold value that triggered alert
-        self.operator = None  # The comparison operator (>, <, >=, etc.)
+        self.last_notification: Optional[float] = None
+        self.threshold_value: Optional[float] = None  # The threshold value that triggered alert
+        self.operator: Optional[str] = None  # The comparison operator (>, <, >=, etc.)
         self.hysteresis: Optional[float] = None  # Hysteresis fraction used for recovery
-        self.formatted_message = None  # Formatted display message for UI
+        self.formatted_message: Optional[str] = None  # Formatted display message for UI
         self.acknowledged = False  # Whether alert has been acknowledged
-        self.acknowledged_at = None  # Timestamp when acknowledged
+        self.acknowledged_at: Optional[float] = None  # Timestamp when acknowledged
         self.consecutive_count = 0  # Consecutive exceedances while still OK (for count gating)
         self.pending_since: Optional[float] = None  # non-None while waiting out grace period before notifying
 
@@ -182,6 +183,7 @@ class AlertState:
     def __str__(self):
         return self.to_dict().__str__()
 
+
 class ThresholdConfig:
     """Configuration for a single threshold check."""
 
@@ -360,13 +362,13 @@ class ThresholdChecker:
             journal: Optional MessageJournal instance for logging threshold events
         """
         # Named threshold configurations (pre-merged: defaults + overrides): {config_name: {metric_path: ThresholdConfig}}
-        self.threshold_configs = {}
+        self.threshold_configs: Dict[str, Dict[str, ThresholdConfig]] = {}
 
         # Raw overrides only for each named config (no defaults baked in): {config_name: {metric_path: ThresholdConfig}}
         self.threshold_raw_configs: Dict[str, Dict[str, ThresholdConfig]] = {}
 
         # Single threshold set for backward compatibility: {metric_path: ThresholdConfig}
-        self.thresholds = {}
+        self.thresholds: Dict[str, ThresholdConfig] = {}
 
         # Host to ordered list of config names: {host_name: [config_name, ...]}
         self.host_config_mapping: Dict[str, List[str]] = {}
@@ -827,7 +829,7 @@ class ThresholdChecker:
             Dictionary of thresholds for this host
         """
         # Legacy mode: single threshold set for all hosts
-        if self.thresholds and not self.threshold_configs:
+        if self.thresholds:
             return self.thresholds
 
         if not self.threshold_configs:
@@ -946,6 +948,7 @@ class ThresholdChecker:
             self._check_pending_or_renotify(host_name, alert_state, metric_path, value, threshold, None)
 
         return None
+
     def _find_threshold(
         self, thresholds: Dict[str, "ThresholdConfig"], metric_path: str
     ) -> Tuple[Optional["ThresholdConfig"], Optional[str]]:
@@ -1273,7 +1276,7 @@ class ThresholdChecker:
 
     def _format_display(
         self,
-        display_format: str,
+        display_format: Optional[str],
         value: Any,
         threshold_value: Optional[float],
         op_symbol: str,

hbd/server/udp.py

@@ -5,6 +5,7 @@ import socket
 import struct
 import time
 import zlib
+from typing import Optional
 import logging
 
 from platform import system as platform_system
@@ -25,6 +26,7 @@ _STALE_INTERVAL_MULTIPLIER = 3
 # Supported on Linux, FreeBSD, and macOS.  The constant is not exposed by
 # Python's socket module on all platforms
 platform = platform_system()
+_SO_TIMESTAMP: Optional[int]
 if platform == "Darwin":
     _SO_TIMESTAMP = 1024  # SO_TIMESTAMP on macOS (not in Python's socket module)
 elif platform == "Linux":
@@ -45,6 +47,8 @@ def enable_kernel_timestamps(sock) -> bool:
     Returns True if the kernel will supply receive timestamps, False otherwise
     (unsupported platform, older kernel, or insufficient permissions).
     """
+    if _SO_TIMESTAMP is None:
+        return False
     try:
         sock.setsockopt(socket.SOL_SOCKET, _SO_TIMESTAMP, 1)
         return True
@@ -71,6 +75,7 @@ class RecvmsgTransport:
     Exposes the same sendto() / close() interface as asyncio's DatagramTransport
     so the rest of the code does not need to know which path is in use.
     """
+
     def __init__(self, loop, sock):
         self._loop = loop
         self._sock = sock
@@ -422,15 +427,20 @@ def handle_datagram(msg: dict, addr, transport, ctx: dict):
         except Exception as e:
             logger.debug(f"Failed to log message to journal: {e}")
 
+    # Only named HTB/PLG datagrams come from clients; anything else (e.g. an internet scanner's
+    # SIP probe) is journaled above but must not sign on a host.
+    if msg.get("ID") not in ("HTB", "PLG") or not msg.get("name"):
+        logger.debug("Ignoring non-client datagram from %s", addr)
+        return
+
     cfg = ctx.get("config", {})
-    hbdcls = ctx.get("hbdclass")
+    hbdcls = ctx["hbdclass"]
     msg_to_websockets = ctx.get("msg_to_websockets")
     DEBUG = ctx.get("DEBUG", 0)
-    verbose = ctx.get("verbose", False)
 
     # normalize addr (ip, port)
     ip = addr[0] if isinstance(addr, (list, tuple)) else addr
-    name = msg.get("name", "unknown")
+    name = msg["name"]
     from ..common.utils import shortname
     from . import config as config_mod
 
@@ -453,10 +463,8 @@ def handle_datagram(msg: dict, addr, transport, ctx: dict):
         newh = False
 
     cid = msg.get("id", 0)
-    try:
-        rtt = float(msg.get("rtt"))
-    except TypeError:
-        rtt = None
+    rtt_raw = msg.get("rtt")
+    rtt = float(rtt_raw) if rtt_raw is not None else None
 
     if msg.get("ID") == "HTB":
         host.doesack = msg.get("acks", -1)

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "hbd"
-version = "5.4.5"
+version = "5.4.7"
 description = "Heartbeat monitoring system — client (hbc) and server (hbd)"
 readme = "README.md"
 requires-python = ">=3.11"
@@ -58,9 +58,12 @@ all = [
 # Development dependencies
 dev = [
   "pytest>=7.0",
+  "pytest-asyncio>=0.23",
   "pytest-cov>=4.0",
   "flake8>=5.0",
   "mypy>=1.10",
+  "types-PyYAML",
+  "types-psutil",
   "black>=23.0",
   "isort>=5.0",
   "tox>=4.0",
@@ -83,10 +86,18 @@ include = ["hbd*"]
 
 
 [tool.black]
-line-length = 111
+line-length = 165
 
 [tool.flake8]
-max-line-length = 111
+max-line-length = 165
 
 [tool.pylint.format]
 max-line-length = 111
+
+[[tool.mypy.overrides]]
+# optional notification libraries that ship neither stubs nor py.typed
+module = ["nio", "nio.*", "mattermostdriver", "mattermostdriver.*"]
+ignore_missing_imports = true
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]

scripts/c/Makefile

@@ -1,6 +1,8 @@
 CC      ?= cc
 CFLAGS  = -O2 -Wall -Wextra -std=c11
 LDFLAGS = -lz -lpthread -lm
+# res_search() lives in libresolv on older glibc; in libc on newer glibc, musl and the BSDs
+LDFLAGS += $(shell echo 'int main(void){return 0;}' | $(CC) -x c - -lresolv -o /dev/null 2>/dev/null && echo -lresolv)
 TARGET  = hbc_mini
 SRC     = hbc_mini.c
 

scripts/c/hbc_mini.c

@@ -20,6 +20,7 @@
 #endif
 
 #include <arpa/inet.h>
+#include <arpa/nameser.h>
 #include <netinet/in.h>
 #include <ctype.h>
 #include <errno.h>
@@ -27,6 +28,7 @@
 #include <math.h>
 #include <netdb.h>
 #include <pthread.h>
+#include <resolv.h>
 #include <signal.h>
 #include <stdarg.h>
 #include <stdbool.h>
@@ -576,6 +578,7 @@ static int conn_send(conn_t *c, const char *msg_id, kvdict_t *d) {
 
     struct sockaddr_storage sa; socklen_t salen = 0;
     memset(&sa, 0, sizeof(sa));
+    pthread_mutex_lock(&c->mu);     /* conn_retarget() may change addr/af/sockfd */
     if (c->af == AF_INET) {
         struct sockaddr_in *s4 = (struct sockaddr_in *)&sa;
         s4->sin_family = AF_INET;
@@ -590,7 +593,6 @@ static int conn_send(conn_t *c, const char *msg_id, kvdict_t *d) {
         salen = sizeof(*s6);
     }
 
-    pthread_mutex_lock(&c->mu);
     c->last_send = now_ts();
     ssize_t sent = sendto(c->sockfd, wire, wlen, 0, (struct sockaddr *)&sa, salen);
     pthread_mutex_unlock(&c->mu);
@@ -1655,6 +1657,199 @@ static void usage(const char *prog) {
         "  -h          Help\n", prog);
 }
 
+/* ============================================================
+ * Re-resolution of the server hosts at the TTL of their A/AAAA records
+ * getaddrinfo() hides the TTL, so a res_search() query is made only to learn it; the
+ * addresses still come from getaddrinfo(). TTL unknown (IP literal, /etc/hosts name, no
+ * resolv.conf, timeout) -> DNS_DEFAULT_TTL. Runs in a detached thread so a slow resolver
+ * cannot delay heartbeats; the main loop applies the result.
+ * ============================================================ */
+
+#define DNS_DEFAULT_TTL 300     /* used when the real TTL is unknown */
+#define DNS_MIN_TTL     30      /* a TTL of 0 must not turn into a resolution loop */
+#define DNS_FAILED_RETRY 60
+
+typedef struct { int af; char addr[INET6_ADDRSTRLEN]; } dns_addr_t;
+
+typedef struct {
+    const char **hosts;
+    int          nhost, port, af_filter;
+} dns_ctx_t;
+
+static struct {
+    pthread_mutex_t mu;
+    bool            ready;      /* a result is waiting for the main loop */
+    int             n;
+    dns_addr_t      addrs[MAX_HOSTS];
+} g_dns = { PTHREAD_MUTEX_INITIALIZER, false, 0, {{0, ""}} };
+
+/* Smallest A/AAAA TTL (CNAMEs included) of host, or -1 if it cannot be determined. */
+static long dns_host_ttl(const char *host, int af_filter) {
+    unsigned char probe[sizeof(struct in6_addr)];
+    if (inet_pton(AF_INET, host, probe) == 1 || inet_pton(AF_INET6, host, probe) == 1) return -1;
+
+    long best = -1;
+    int types[2] = { ns_t_a, ns_t_aaaa };
+    for (int t = 0; t < 2; t++) {
+        if (af_filter == AF_INET && t == 1) continue;
+        if (af_filter == AF_INET6 && t == 0) continue;
+        unsigned char ans[NS_PACKETSZ * 2];
+        int len = res_search(host, ns_c_in, types[t], ans, sizeof(ans));
+        ns_msg msg;
+        if (len < 0 || ns_initparse(ans, len, &msg) < 0) continue;
+        long min = -1; bool found = false;
+        for (int i = 0; i < ns_msg_count(msg, ns_s_an); i++) {
+            ns_rr rr;
+            if (ns_parserr(&msg, ns_s_an, i, &rr) < 0) continue;
+            int ty = ns_rr_type(rr);
+            if (ty != types[t] && ty != ns_t_cname) continue;
+            if (ty == types[t]) found = true;
+            if (min < 0 || (long)ns_rr_ttl(rr) < min) min = (long)ns_rr_ttl(rr);
+        }
+        if (found && (best < 0 || min < best)) best = min;
+    }
+    return best;
+}
+
+/* Seconds until the next resolution: the smallest known TTL, floored at DNS_MIN_TTL. */
+static int dns_refresh_delay(const dns_ctx_t *x) {
+    long best = -1;
+    for (int i = 0; i < x->nhost; i++) {
+        long t = dns_host_ttl(x->hosts[i], x->af_filter);
+        if (t >= 0 && (best < 0 || t < best)) best = t;
+    }
+    if (best < 0) return DNS_DEFAULT_TTL;
+    return best < DNS_MIN_TTL ? DNS_MIN_TTL : (best > 86400 ? 86400 : (int)best);
+}
+
+/* Every (family, address) the hosts resolve to, deduplicated; -1 if any lookup fails. */
+static int dns_resolve_all(const dns_ctx_t *x, dns_addr_t *out, int max) {
+    int n = 0;
+    for (int i = 0; i < x->nhost; i++) {
+        struct addrinfo hints = {0}, *res = NULL;
+        hints.ai_socktype = SOCK_DGRAM;
+        hints.ai_protocol = IPPROTO_UDP;
+        hints.ai_family   = x->af_filter;
+        char ps[16]; snprintf(ps, sizeof(ps), "%d", x->port);
+        if (getaddrinfo(x->hosts[i], ps, &hints, &res) != 0) {
+            LOGW("cannot resolve %s", x->hosts[i]);
+            return -1;
+        }
+        for (struct addrinfo *ai = res; ai && n < max; ai = ai->ai_next) {
+            dns_addr_t a; a.af = ai->ai_family;
+            void *ip = (ai->ai_family == AF_INET)
+                ? (void *)&((struct sockaddr_in  *)ai->ai_addr)->sin_addr
+                : (void *)&((struct sockaddr_in6 *)ai->ai_addr)->sin6_addr;
+            inet_ntop(ai->ai_family, ip, a.addr, sizeof(a.addr));
+            bool dup = false;
+            for (int k = 0; k < n; k++) if (strcmp(out[k].addr, a.addr) == 0) dup = true;
+            if (!dup) out[n++] = a;
+        }
+        freeaddrinfo(res);
+    }
+    return n;
+}
+
+static void *dns_thread(void *arg) {
+    const dns_ctx_t *x = arg;
+    int delay = dns_refresh_delay(x);
+    while (g_running) {
+        for (int s = 0; s < delay && g_running; s++) sleep(1);
+        if (!g_running) break;
+        dns_addr_t tmp[MAX_HOSTS];
+        int n = dns_resolve_all(x, tmp, MAX_HOSTS);
+        if (n <= 0) { delay = DNS_FAILED_RETRY; continue; }
+        pthread_mutex_lock(&g_dns.mu);
+        memcpy(g_dns.addrs, tmp, sizeof(tmp));
+        g_dns.n = n; g_dns.ready = true;
+        pthread_mutex_unlock(&g_dns.mu);
+        delay = dns_refresh_delay(x);
+    }
+    return NULL;
+}
+
+/* Point c at a new server address; the socket is replaced if the family changed. */
+static void conn_retarget(conn_t *c, int af, const char *addr) {
+    LOGI("server address %s -> %s", c->addr, addr);
+    pthread_mutex_lock(&c->mu);
+    if (af != c->af) {
+        int fd = socket(af, SOCK_DGRAM, 0);
+        if (fd < 0) { LOGE("socket: %s", strerror(errno)); pthread_mutex_unlock(&c->mu); return; }
+        fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK);
+        if (c->sockfd >= 0) close(c->sockfd);
+        c->sockfd = fd; c->af = af;
+    }
+    snprintf(c->addr, sizeof(c->addr), "%s", addr);
+    pthread_mutex_unlock(&c->mu);
+}
+
+/* Main-loop side: follow the latest resolution. Mirrors hbd.client.dnsttl.retarget(). Gone
+ * addresses are re-pointed at new ones in list order, so g_conns[0] (the one plugin data
+ * uses) survives as long as anything resolves; the rest are retired, surplus new ones added. */
+static void dns_apply(int *conn_id, const char *iam, int port) {
+    dns_addr_t r[MAX_HOSTS]; int nr;
+    pthread_mutex_lock(&g_dns.mu);
+    if (!g_dns.ready) { pthread_mutex_unlock(&g_dns.mu); return; }
+    memcpy(r, g_dns.addrs, sizeof(r)); nr = g_dns.n; g_dns.ready = false;
+    pthread_mutex_unlock(&g_dns.mu);
+
+    int gone[MAX_HOSTS], ng = 0, fresh[MAX_HOSTS], nf = 0;
+    for (int i = 0; i < g_nconns; i++) {
+        bool wanted = false;
+        for (int j = 0; j < nr; j++) if (strcmp(g_conns[i].addr, r[j].addr) == 0) wanted = true;
+        if (!wanted) gone[ng++] = i;
+    }
+    for (int j = 0; j < nr; j++) {
+        bool have = false;
+        for (int i = 0; i < g_nconns; i++) if (strcmp(g_conns[i].addr, r[j].addr) == 0) have = true;
+        if (!have) fresh[nf++] = j;
+    }
+
+    int paired = ng < nf ? ng : nf;
+    for (int k = 0; k < paired; k++) conn_retarget(&g_conns[gone[k]], r[fresh[k]].af, r[fresh[k]].addr);
+
+    /* drop[] holds connection indexes to retire, in ascending order */
+    int drop[MAX_HOSTS], nd = 0;
+    for (int k = paired; k < ng; k++) drop[nd++] = gone[k];
+    if (nd > 0 && drop[0] == 0) {
+        int keep = -1;
+        for (int i = 1; i < g_nconns && keep < 0; i++) {
+            bool isgone = false;
+            for (int k = 0; k < ng; k++) if (gone[k] == i) isgone = true;
+            if (!isgone) keep = i;
+        }
+        if (keep > 0) {
+            conn_retarget(&g_conns[0], g_conns[keep].af, g_conns[keep].addr);
+            drop[0] = keep;
+            for (int a = 0; a < nd - 1 && drop[a] > drop[a + 1]; a++) {
+                int tmp = drop[a]; drop[a] = drop[a + 1]; drop[a + 1] = tmp;
+            }
+        }
+    }
+    for (int k = nd - 1; k >= 0; k--) {
+        int i = drop[k];
+        if (i == 0) continue;
+        LOGI("dropping server address %s", g_conns[i].addr);
+        conn_close(&g_conns[i]);
+        pthread_mutex_destroy(&g_conns[i].mu);
+        int last = --g_nconns;
+        if (i != last) {
+            memcpy(&g_conns[i], &g_conns[last], sizeof(conn_t));
+            pthread_mutex_init(&g_conns[i].mu, NULL);
+        }
+    }
+
+    for (int k = paired; k < nf && g_nconns < MAX_HOSTS; k++) {
+        conn_t *c = &g_conns[g_nconns];
+        memset(c, 0, sizeof(*c));
+        c->conn_id = (*conn_id)++; c->port = port;
+        c->af = r[fresh[k]].af; c->sockfd = -1;
+        snprintf(c->name, sizeof(c->name), "%s", iam);
+        snprintf(c->addr, sizeof(c->addr), "%s", r[fresh[k]].addr);
+        if (conn_open(c)) { g_nconns++; LOGI("adding server address %s", c->addr); }
+    }
+}
+
 int main(int argc, char **argv) {
     bool        do_boot   = false;
     bool        do_daemon = false;
@@ -1776,9 +1971,17 @@ int main(int argc, char **argv) {
         else LOGW("pthread_create(probe): %s", strerror(prc));
     }
 
+    /* Re-resolution at the DNS TTL; detached, the process exit reaps it */
+    dns_ctx_t dns_ctx = { hosts, nhost, cfg.hb_port, af_filter };
+    pthread_t dtid;
+    int drc = pthread_create(&dtid, NULL, dns_thread, &dns_ctx);
+    if (drc == 0) pthread_detach(dtid);
+    else LOGW("pthread_create(dns): %s", strerror(drc));
+
     /* Main heartbeat + receive loop */
     time_t last_hb = 0;
     while (g_running) {
+        dns_apply(&conn_id, iam, cfg.hb_port);
         time_t now = time(NULL);
         if (now - last_hb >= cfg.interval) {
             for (int i = 0; i < g_nconns; i++) {

scripts/c/test_probe.c

@@ -226,6 +226,55 @@ static int live_mode(void) {
     return 0;
 }
 
+/* ---- server address re-resolution (dns_apply) ---- */
+
+static void dns_conn(int i, int af, const char *addr) {
+    memset(&g_conns[i], 0, sizeof(conn_t));
+    g_conns[i].conn_id = i + 1; g_conns[i].af = af; g_conns[i].port = 5;
+    snprintf(g_conns[i].addr, sizeof(g_conns[i].addr), "%s", addr);
+    conn_open(&g_conns[i]);
+}
+
+static void dns_push(int n, const dns_addr_t *a) {
+    memcpy(g_dns.addrs, a, n * sizeof(*a)); g_dns.n = n; g_dns.ready = true;
+}
+
+static void test_dns_apply_follows_changes(void) {
+    int id = 3;
+    dns_conn(0, AF_INET, "1.1.1.1"); dns_conn(1, AF_INET6, "2001:db8::1"); g_nconns = 2;
+
+    dns_addr_t moved[2] = {{AF_INET, "9.9.9.9"}, {AF_INET6, "2001:db8::9"}};
+    dns_push(2, moved); dns_apply(&id, "me", 5);
+    CHECK(g_nconns == 2);
+    CHECK_STR(g_conns[0].addr, "9.9.9.9");
+    CHECK_STR(g_conns[1].addr, "2001:db8::9");
+
+    /* the first connection's address vanishes and nothing replaces it: it survives, the other goes */
+    dns_addr_t v6only[1] = {{AF_INET6, "2001:db8::9"}};
+    dns_push(1, v6only); dns_apply(&id, "me", 5);
+    CHECK(g_nconns == 1);
+    CHECK_STR(g_conns[0].addr, "2001:db8::9");
+    CHECK(g_conns[0].af == AF_INET6);
+
+    dns_addr_t grown[2] = {{AF_INET6, "2001:db8::9"}, {AF_INET, "5.5.5.5"}};
+    dns_push(2, grown); dns_apply(&id, "me", 5);
+    CHECK(g_nconns == 2);
+    CHECK_STR(g_conns[1].addr, "5.5.5.5");
+    CHECK(g_conns[1].conn_id == 3);
+
+    dns_push(2, grown); dns_apply(&id, "me", 5);        /* unchanged: nothing happens */
+    CHECK(g_nconns == 2 && id == 4);
+    for (int i = 0; i < g_nconns; i++) conn_close(&g_conns[i]);
+    g_nconns = 0;
+}
+
+static void test_dns_ttl_unknown_for_ip_literals(void) {
+    CHECK(dns_host_ttl("192.0.2.1", 0) == -1);
+    CHECK(dns_host_ttl("2001:db8::1", 0) == -1);
+    const char *h[] = {"192.0.2.1"}; dns_ctx_t x = {h, 1, 5, 0};
+    CHECK(dns_refresh_delay(&x) == DNS_DEFAULT_TTL);
+}
+
 int main(int argc, char **argv) {
     if (argc > 1 && !strcmp(argv[1], "--live")) return live_mode();
     test_gateway_parsers();
@@ -238,6 +287,8 @@ int main(int argc, char **argv) {
     test_route_loss_then_return_pings_immediately();
     test_missing_ping_stays_unknown_even_after_route_loss();
     test_unsupported_family_is_unknown();
+    test_dns_apply_follows_changes();
+    test_dns_ttl_unknown_for_ip_literals();
     printf("%d checks, %d failed\n", g_run, g_fail);
     return g_fail ? 1 : 0;
 }

scripts/hbc_mini.py

@@ -28,6 +28,7 @@ IPv4) path as an outage. Config, in the network_monitor section of ~/.hbc.json:
 
 import argparse
 import asyncio
+import ipaddress
 import json
 import logging
 import os
@@ -36,6 +37,7 @@ import re
 import shutil
 import signal
 import socket
+import struct
 import subprocess
 import sys
 import time
@@ -48,7 +50,7 @@ from pathlib import Path
 from typing import Any, Awaitable, Callable, Dict, List, Optional, Set, Tuple
 
 # updated by scripts/bumpminor.sh
-__version__ = "5.4.5"
+__version__ = "5.4.7"
 
 # ---------------------------------------------------------------------------
 # Protocol  (mirrors hbd/common/proto.py)
@@ -1164,6 +1166,14 @@ class AsyncConnection:
             self._transport.close()
             self._transport = None
 
+    def set_target(self, addr: str, af: int):
+        """Re-point at a new server address; the socket is reopened on the next send if the family changed."""
+        if af != self.af:
+            self.close()
+            self.af = af
+        self.addr = addr
+        self._log = logging.getLogger(f"hbc.conn.{addr}")
+
     def _handle_ack(self, msg: Dict[str, Any], now: float):
         rtt = (now - self.lastsend) * 1000.0
         self.rtts.append(rtt)
@@ -1188,6 +1198,207 @@ class AsyncConnection:
         self.lastsend = time.time()
 
 
+# ---------------------------------------------------------------------------
+# Re-resolution of the server hosts at the TTL of their A/AAAA records
+# getaddrinfo() hides the TTL, so a minimal stub-resolver query is made only to learn it; the
+# addresses still come from getaddrinfo(). TTL unknown (IP literal, /etc/hosts name, no
+# resolv.conf, timeout) -> _DNS_DEFAULT_TTL.
+# ---------------------------------------------------------------------------
+
+log = logging.getLogger("hbc.dns")
+
+_DNS_DEFAULT_TTL = 300       # used when the real TTL is unknown
+_DNS_MIN_TTL = 30            # a TTL of 0 must not turn into a resolution loop
+_DNS_FAILED_RETRY = 60       # re-try delay after a failed resolution
+_DNS_QUERY_TIMEOUT = 2.0
+
+_DNS_TYPE_A, _DNS_TYPE_CNAME, _DNS_TYPE_AAAA = 1, 5, 28
+
+
+def _dns_parse_resolv_conf(path: str = "/etc/resolv.conf") -> Tuple[List[str], List[str]]:
+    """Return (nameservers, search domains)."""
+    servers: List[str] = []
+    search: List[str] = []
+    try:
+        with open(path) as f:
+            for line in f:
+                parts = line.split()
+                if len(parts) < 2:
+                    continue
+                if parts[0] == "nameserver":
+                    servers.append(parts[1].split("%")[0])
+                elif parts[0] in ("search", "domain"):
+                    search = parts[1:]
+    except OSError:
+        pass
+    return servers, search
+
+
+def _dns_build_query(name: str, qtype: int, qid: int) -> bytes:
+    out = struct.pack("!HHHHHH", qid, 0x0100, 1, 0, 0, 0)
+    for label in name.rstrip(".").split("."):
+        out += bytes([len(label)]) + label.encode("idna")
+    return out + b"\x00" + struct.pack("!HH", qtype, 1)
+
+
+def _dns_skip_name(buf: bytes, off: int) -> int:
+    while True:
+        n = buf[off]
+        if n == 0:
+            return off + 1
+        if n & 0xC0:
+            return off + 2
+        off += n + 1
+
+
+def _dns_min_answer_ttl(resp: bytes, qid: int, qtype: int) -> Optional[int]:
+    """Smallest TTL among the answer RRs (CNAMEs included), or None if there is no qtype answer."""
+    try:
+        rid, flags, qd, an = struct.unpack("!HHHH", resp[:8])
+        if rid != qid or flags & 0x000F:
+            return None
+        off = 12
+        for _ in range(qd):
+            off = _dns_skip_name(resp, off) + 4
+        ttls: List[int] = []
+        found = False
+        for _ in range(an):
+            off = _dns_skip_name(resp, off)
+            rtype, _cls, ttl, rdlen = struct.unpack("!HHIH", resp[off:off + 10])
+            off += 10 + rdlen
+            if rtype in (qtype, _DNS_TYPE_CNAME):
+                ttls.append(ttl)
+            found = found or rtype == qtype
+        return min(ttls) if found else None
+    except (struct.error, IndexError):
+        return None
+
+
+def _dns_query_ttl(server: str, name: str, qtype: int) -> Optional[int]:
+    qid = struct.unpack("!H", os.urandom(2))[0]
+    try:
+        af = socket.AF_INET6 if ":" in server else socket.AF_INET
+        with socket.socket(af, socket.SOCK_DGRAM) as s:
+            s.settimeout(_DNS_QUERY_TIMEOUT)
+            s.sendto(_dns_build_query(name, qtype, qid), (server, 53))
+            return _dns_min_answer_ttl(s.recv(4096), qid, qtype)
+    except (OSError, UnicodeError):
+        return None
+
+
+def _dns_lookup_ttl(host: str, af_filter: int = 0) -> Optional[int]:
+    """Blocking. Smallest A/AAAA TTL for host, or None when it cannot be determined."""
+    try:
+        ipaddress.ip_address(host)
+        return None
+    except ValueError:
+        pass
+    servers, search = _dns_parse_resolv_conf()
+    if host.endswith("."):
+        names = [host]
+    elif "." in host:
+        names = [host] + [f"{host}.{d}" for d in search]
+    else:
+        names = [f"{host}.{d}" for d in search] + [host]
+    qtypes = [t for t, af in ((_DNS_TYPE_A, socket.AF_INET), (_DNS_TYPE_AAAA, socket.AF_INET6))
+              if af_filter in (0, af)]
+    for name in names:
+        ttls = []
+        for qtype in qtypes:
+            for server in servers:
+                ttl = _dns_query_ttl(server, name, qtype)
+                if ttl is not None:
+                    ttls.append(ttl)
+                    break
+        if ttls:
+            return min(ttls)
+    return None
+
+
+def _dns_refresh_delay(ttls: List[Optional[int]]) -> int:
+    """Seconds until the next resolution: the smallest known TTL, floored at _DNS_MIN_TTL."""
+    known = [t for t in ttls if t is not None]
+    return max(min(known), _DNS_MIN_TTL) if known else _DNS_DEFAULT_TTL
+
+
+def _dns_resolve(hosts: List[str], port: int, af_filter: int) -> Optional[List[Tuple[int, str]]]:
+    """Blocking. Every (family, address) the hosts resolve to, deduplicated, or None if any lookup fails."""
+    found: List[Tuple[int, str]] = []
+    for host in hosts:
+        try:
+            infos = socket.getaddrinfo(host, port, af_filter, 0, socket.SOL_UDP)
+        except socket.gaierror as e:
+            log.warning("cannot resolve %s: %s", host, e)
+            return None
+        for info in infos:
+            entry = (info[0], str(info[4][0]))
+            if entry not in found:
+                found.append(entry)
+    return found or None
+
+
+def _dns_retarget(connections: list, resolved: List[Tuple[int, str]]) -> Tuple[List[Tuple[int, str]], list]:
+    """Point the connections at the resolved addresses, in place.
+
+    A connection whose address is gone is re-pointed at an address that is new, in list order, so
+    the first connection (the one plugin data uses) survives as long as anything resolves.
+    Returns (new addresses still needing a connection, connections to drop).
+    """
+    current = {c.addr for c in connections}
+    wanted = {addr for _, addr in resolved}
+    new = [(af, addr) for af, addr in resolved if addr not in current]
+    gone = [c for c in connections if c.addr not in wanted]
+    paired = min(len(gone), len(new))
+    for conn, (af, addr) in zip(gone, new):
+        log.info("server address %s -> %s", conn.addr, addr)
+        conn.set_target(addr, af)
+    drop = gone[paired:]
+    if connections and connections[0] in drop:
+        # plugin data goes out on the first connection: hand it a surviving address and drop that one instead
+        keep = next(c for c in connections if c not in gone)
+        connections[0].set_target(keep.addr, keep.af)
+        drop.remove(connections[0])
+        drop.append(keep)
+    return new[paired:], drop
+
+
+async def _dns_refresher(connections: List["AsyncConnection"], hosts: List[str], port: int, af_filter: int,
+                         name: str, interval: int, probe: Optional[_Connectivity], next_id: int):
+    """Re-resolve the server hosts every TTL seconds and follow address changes.
+
+    A failed lookup leaves the connections untouched. Addresses that disappear are re-pointed at
+    new ones where possible; the rest are retired, and surplus new ones get a connection.
+    """
+    loop = asyncio.get_running_loop()
+
+    def ttl_delay() -> int:
+        return _dns_refresh_delay([_dns_lookup_ttl(h, af_filter) for h in hosts])
+
+    delay = await loop.run_in_executor(None, ttl_delay)
+    while _running:
+        await _sleep(delay)
+        if not _running:
+            break
+        resolved = await loop.run_in_executor(None, _dns_resolve, hosts, port, af_filter)
+        if resolved is None:
+            delay = _DNS_FAILED_RETRY
+            continue
+        new, gone = _dns_retarget(connections, resolved)
+        for conn in gone:
+            if len(connections) > 1:
+                log.info("dropping server address %s", conn.addr)
+                conn._dead = True
+                conn.close()
+                connections.remove(conn)
+        for af, addr in new:
+            conn = AsyncConnection(next_id, addr, port, af, name)
+            next_id += 1
+            log.info("adding server address %s", addr)
+            connections.append(conn)
+            _active_tasks.append(asyncio.create_task(_heartbeat_sender(conn, interval, probe)))
+        delay = await loop.run_in_executor(None, ttl_delay)
+
+
 # ---------------------------------------------------------------------------
 # Server command handlers
 # ---------------------------------------------------------------------------
@@ -1463,6 +1674,9 @@ async def _async_main(args, cfg: Dict[str, Any]) -> int:
     for conn in connections:
         _active_tasks.append(asyncio.create_task(_heartbeat_sender(conn, interval, _probe)))
 
+    _active_tasks.append(asyncio.create_task(
+        _dns_refresher(connections, args.hosts, port, af_filter, iam, interval, _probe, conn_id)))
+
     if plugins and connections:
         _active_tasks.append(asyncio.create_task(_plugin_collector(connections[0], plugins)))
 

tests/test_client_dnsttl.py

@@ -0,0 +1,119 @@
+"""TTL-driven re-resolution of the server hostnames in the full client."""
+import socket
+import struct
+
+from hbd.client import dnsttl
+
+
+def _rr(rtype, ttl, rdata):
+    return b"\xc0\x0c" + struct.pack("!HHIH", rtype, 1, ttl, len(rdata)) + rdata
+
+
+def _response(qid, qtype, answers, rcode=0):
+    q = dnsttl.build_query("hbd.example.com", qtype, qid)[12:]
+    return struct.pack("!HHHHHH", qid, 0x8180 | rcode, 1, len(answers), 0, 0) + q + b"".join(answers)
+
+
+def test_min_ttl_covers_cname_and_address():
+    resp = _response(7, dnsttl.TYPE_A, [_rr(dnsttl.TYPE_CNAME, 60, b"\x01x\x00"),
+                                        _rr(dnsttl.TYPE_A, 300, b"\x01\x02\x03\x04")])
+    assert dnsttl.min_answer_ttl(resp, 7, dnsttl.TYPE_A) == 60
+
+
+def test_no_answer_of_requested_type_is_unknown():
+    assert dnsttl.min_answer_ttl(_response(7, dnsttl.TYPE_AAAA, []), 7, dnsttl.TYPE_AAAA) is None
+
+
+def test_error_rcode_and_foreign_id_are_unknown():
+    ans = [_rr(dnsttl.TYPE_A, 300, b"\x01\x02\x03\x04")]
+    assert dnsttl.min_answer_ttl(_response(7, 1, ans, rcode=3), 7, 1) is None
+    assert dnsttl.min_answer_ttl(_response(7, 1, ans), 8, 1) is None
+
+
+def test_truncated_response_is_unknown():
+    assert dnsttl.min_answer_ttl(b"\x00\x07\x81", 7, 1) is None
+
+
+def test_refresh_delay():
+    assert dnsttl.refresh_delay([300, 120, None]) == 120
+    assert dnsttl.refresh_delay([0]) == dnsttl.MIN_TTL
+    assert dnsttl.refresh_delay([None]) == dnsttl.DEFAULT_TTL
+
+
+def test_ip_literal_has_no_ttl():
+    assert dnsttl.lookup_ttl("192.0.2.1") is None
+    assert dnsttl.lookup_ttl("2001:db8::1") is None
+
+
+class _C:
+    def __init__(self, addr, af=socket.AF_INET):
+        self.addr, self.af = addr, af
+
+    def set_target(self, addr, af):
+        self.addr, self.af = addr, af
+
+
+V4, V6 = socket.AF_INET, socket.AF_INET6
+
+
+def test_retarget_follows_address_change_in_place():
+    first, second = _C("1.1.1.1"), _C("2001:db8::1", V6)
+    new, gone = dnsttl.retarget([first, second], [(V4, "9.9.9.9"), (V6, "2001:db8::9")])
+    assert (new, gone) == ([], [])
+    assert (first.addr, second.addr) == ("9.9.9.9", "2001:db8::9")
+
+
+def test_retarget_unchanged_is_a_noop():
+    conns = [_C("1.1.1.1")]
+    assert dnsttl.retarget(conns, [(V4, "1.1.1.1")]) == ([], [])
+
+
+def test_retarget_reports_surplus_new_and_gone():
+    a, b = _C("1.1.1.1"), _C("2.2.2.2")
+    new, gone = dnsttl.retarget([a, b], [(V4, "3.3.3.3")])
+    assert new == [] and gone == [b] and a.addr == "3.3.3.3"
+    new, gone = dnsttl.retarget([a], [(V4, "3.3.3.3"), (V4, "4.4.4.4")])
+    assert new == [(V4, "4.4.4.4")] and gone == []
+
+
+def test_retarget_never_drops_the_first_connection():
+    a, b = _C("1.1.1.1"), _C("2.2.2.2")
+    new, gone = dnsttl.retarget([a, b], [(V4, "2.2.2.2")])
+    assert new == [] and gone == [b] and a.addr == "2.2.2.2"
+
+
+def test_resolve_failure_returns_none(monkeypatch):
+    def boom(*a):
+        raise socket.gaierror("nope")
+    monkeypatch.setattr(socket, "getaddrinfo", boom)
+    assert dnsttl.resolve(["x.invalid"], 1, 0) is None
+
+
+def test_refresher_follows_a_change_and_adds_a_family(monkeypatch):
+    import asyncio
+    from hbd.client import main as hbc
+
+    answers = [[(V4, "9.9.9.9"), (V6, "2001:db8::9")], None, [(V4, "9.9.9.9")]]
+    monkeypatch.setattr(dnsttl, "resolve", lambda *a: answers.pop(0) if len(answers) > 1 else answers[0])
+    monkeypatch.setattr(dnsttl, "lookup_ttl", lambda *a: None)
+    monkeypatch.setattr(dnsttl, "DEFAULT_TTL", 0.01)
+    monkeypatch.setattr(dnsttl, "FAILED_RETRY", 0.01)
+
+    async def idle(*a):
+        await asyncio.sleep(10)
+    monkeypatch.setattr(hbc, "heartbeat_sender", idle)
+
+    async def go():
+        hbc.shutdown_event = asyncio.Event()
+        hbc.active_tasks = []
+        conns = [hbc.AsyncConnection(1, "1.1.1.1", 5, V4, "me")]
+        task = asyncio.create_task(hbc.dns_refresher(conns, ["h"], 5, 0, "me", 10, None, 2))
+        await asyncio.sleep(0.2)
+        hbc.shutdown_event.set()
+        await task
+        for t in hbc.active_tasks:
+            t.cancel()
+        return conns
+
+    conns = asyncio.run(go())
+    assert [(c.conn_id, c.addr) for c in conns] == [(1, "9.9.9.9")]

tests/test_configio.py

@@ -1,5 +1,3 @@
-import glob
-import os
 import pytest
 from hbd.server import configio
 

tests/test_dns.py

@@ -1,9 +1,8 @@
-import time
-import queue
+import asyncio
 import unittest
 from unittest.mock import patch, MagicMock
 
-import hbd.dns as dns
+import hbd.server.dns as dns
 
 
 class TestDNS(unittest.TestCase):
@@ -17,7 +16,7 @@ class TestDNS(unittest.TestCase):
         p = dns.create_nsupdate_payload("host", "2001:db8::1", "example")
         self.assertIn("update add host.dy.example 5 AAAA 2001:db8::1", p)
 
-    @patch("hbd.dns.Popen")
+    @patch("hbd.server.dns.Popen")
     def test_nsupdate_success(self, mock_popen):
         proc = MagicMock()
         proc.communicate.return_value = (b"status: NOERROR", None)
@@ -41,7 +40,7 @@ class TestDNS(unittest.TestCase):
             stderr=dns.STDOUT,
         )
 
-    @patch("hbd.dns.Popen")
+    @patch("hbd.server.dns.Popen")
     def test_nsupdate_failure(self, mock_popen):
         proc = MagicMock()
         proc.communicate.return_value = (b"some error", None)
@@ -57,83 +56,7 @@ class TestDNS(unittest.TestCase):
         self.assertIsNotNone(err)
         self.assertIn("some error", err)
 
-    def test_dnsupdatethread_processes_queue(self):
-        # patch nsupdate to succeed
-        with patch("hbd.dns.nsupdate", return_value=None):
-            logs = []
-
-            def log(h, m):
-                logs.append((h, m))
-
-            emails = []
-
-            def email(s, m):
-                emails.append((s, m))
-
-            class FakeHost:
-                dnsQ = queue.Queue()
-
-            class FakeHbd:
-                Host = FakeHost
-
-            # start the thread (daemon) that processes the queue
-            t = dns.start_dns_thread(
-                FakeHbd, {"dyndomains": ["example"]}, log=log, email=email
-            )
-            self.assertTrue(t.is_alive())
-
-            # enqueue one item and wait for it to be processed (polling with timeout)
-            FakeHbd.Host.dnsQ.put(("testhost", "1.2.3.4"))
-
-            for _ in range(30):
-                if logs:
-                    break
-                time.sleep(0.1)
-
-            self.assertTrue(logs, "dnsupdatethread did not call log")
-            self.assertTrue(
-                any("changed address" in m or "DNS updated" in m for (_h, m) in logs)
-            )
-
-    def test_dnsupdatethread_calls_email_on_failure(self):
-        # patch nsupdate to fail with an error message
-        with patch("hbd.dns.nsupdate", return_value="error: failed"):
-            logs = []
-
-            def log(h, m):
-                logs.append((h, m))
-
-            emails = []
-
-            def email(s, m):
-                emails.append((s, m))
-
-            class FakeHost:
-                dnsQ = queue.Queue()
-
-            class FakeHbd:
-                Host = FakeHost
-
-            dns.start_dns_thread(
-                FakeHbd, {"dyndomains": ["example"]}, log=log, email=email
-            )
-            # enqueue and wait for the email to be sent
-            FakeHbd.Host.dnsQ.put(("testhost", "1.2.3.4"))
-
-            for _ in range(30):
-                if emails:
-                    break
-                time.sleep(0.1)
-
-            self.assertTrue(emails, "dnsupdatethread did not call email on failure")
-            self.assertTrue(
-                any(
-                    "nsupdate failed" in s or "nsupdate failed" in m or "error" in m
-                    for (s, m) in emails
-                )
-            )
-
-    @patch("hbd.dns.Popen")
+    @patch("hbd.server.dns.Popen")
     def test_nsupdate_raises_oserror(self, mock_popen):
         mock_popen.side_effect = OSError("noexec")
         err = dns.nsupdate(
@@ -147,5 +70,37 @@ class TestDNS(unittest.TestCase):
         self.assertIn("execution failed", err)
 
 
+class TestDNSWorker(unittest.IsolatedAsyncioTestCase):
+
+    async def _run_worker(self, nsupdate_result):
+        """Feed one update through dns_update_worker; returns the (name, level, message) log calls."""
+        logs = []
+        queue = asyncio.Queue()
+        await queue.put(("testhost", "1.2.3.4"))
+        await queue.put(None)
+        with patch("hbd.server.dns.nsupdate", return_value=nsupdate_result):
+            await dns.dns_update_worker(
+                None, {"dyndomains": ["example"]}, async_queue=queue,
+                log=lambda name, level, msg: logs.append((name, level, msg)),
+            )
+        return logs
+
+    async def test_worker_logs_successful_update(self):
+        logs = await self._run_worker(None)
+        self.assertEqual(len(logs), 1)
+        name, level, msg = logs[0]
+        self.assertEqual((name, level), ("testhost", "INFO"))
+        self.assertIn("DNS updated testhost.dy.example", msg)
+        self.assertIn("1.2.3.4", msg)
+
+    async def test_worker_logs_error_on_failed_update(self):
+        logs = await self._run_worker("error: failed")
+        self.assertEqual(len(logs), 1)
+        name, level, msg = logs[0]
+        self.assertEqual((name, level), ("testhost", "ERROR"))
+        self.assertIn("DNS update failed", msg)
+        self.assertIn("error: failed", msg)
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_handle_datagram.py

@@ -1,5 +1,5 @@
-from hbd.udp import handle_datagram, parse_message
-from hbd.proto import dicttos
+from hbd.server.udp import handle_datagram, parse_message
+from hbd.common.proto import dicttos, stodict
 
 
 class FakeTransport:
@@ -17,7 +17,7 @@ def dummy_noop(*a, **k):
 def test_handle_cmd_sends_command():
     ftr = FakeTransport()
     # prepare ctx linking to hbdclass
-    import hbdclass
+    from hbd.server import hbdclass
 
     ctx = {
         "config": {"watchhosts": []},
@@ -34,7 +34,7 @@ def test_handle_cmd_sends_command():
     # create host by sending initial heartbeat
     msg = parse_message(dicttos("HTB", {"name": "cmdhost", "interval": 10}))
     handle_datagram(msg, ("127.0.0.1", 50000), ftr, ctx)
-    assert ftr.sent[0][0] == b"ACK"
+    assert ftr.sent[0][0].startswith(b"!ACK:")
 
     # queue a CMD for the host and send another heartbeat; expect command sent
     h = hbdclass.Host.hosts["cmdhost"]
@@ -42,8 +42,6 @@ def test_handle_cmd_sends_command():
     ftr.sent.clear()
     msg2 = parse_message(dicttos("HTB", {"name": "cmdhost", "interval": 10}))
     handle_datagram(msg2, ("127.0.0.1", 50000), ftr, ctx)
-    # should have sent ACK and the command; last send should be non-empty
-    assert len(ftr.sent) >= 1
-    # the command for ??0 will be sent as raw cmd string
-    # so at least one send contains b'doit' or similar
-    assert any(b"doit" in s[0] for s in ftr.sent)
+    # the ACK and the queued command are both sent; the command is a compressed CMD packet
+    decoded = [stodict(data) for data, _addr in ftr.sent]
+    assert any(d["ID"] == "CMD" and d["cmd"] == "doit" for d in decoded)

tests/test_hbc_mini_dnsttl.py

@@ -0,0 +1,72 @@
+"""TTL-driven re-resolution in the standalone scripts/hbc_mini.py."""
+import asyncio
+import importlib.util
+import pathlib
+import socket
+import struct
+import sys
+
+_PATH = pathlib.Path(__file__).resolve().parents[1] / "scripts" / "hbc_mini.py"
+_SPEC = importlib.util.spec_from_file_location("hbc_mini", _PATH)
+mini = importlib.util.module_from_spec(_SPEC)
+sys.modules["hbc_mini"] = mini
+_SPEC.loader.exec_module(mini)
+
+V4, V6 = socket.AF_INET, socket.AF_INET6
+
+
+def _rr(rtype, ttl, rdata):
+    return b"\xc0\x0c" + struct.pack("!HHIH", rtype, 1, ttl, len(rdata)) + rdata
+
+
+def _response(qid, qtype, answers, rcode=0):
+    q = mini._dns_build_query("hbd.example.com", qtype, qid)[12:]
+    return struct.pack("!HHHHHH", qid, 0x8180 | rcode, 1, len(answers), 0, 0) + q + b"".join(answers)
+
+
+def test_min_ttl_covers_cname_and_address():
+    resp = _response(7, 1, [_rr(5, 60, b"\x01x\x00"), _rr(1, 300, b"\x01\x02\x03\x04")])
+    assert mini._dns_min_answer_ttl(resp, 7, 1) == 60
+
+
+def test_unknown_ttl_cases():
+    assert mini._dns_min_answer_ttl(_response(7, 28, []), 7, 28) is None
+    assert mini._dns_min_answer_ttl(_response(7, 1, [_rr(1, 9, b"1234")], rcode=3), 7, 1) is None
+    assert mini._dns_min_answer_ttl(b"\x00\x07\x81", 7, 1) is None
+    assert mini._dns_lookup_ttl("192.0.2.1") is None
+
+
+def test_refresh_delay():
+    assert mini._dns_refresh_delay([300, 120, None]) == 120
+    assert mini._dns_refresh_delay([0]) == mini._DNS_MIN_TTL
+    assert mini._dns_refresh_delay([None]) == mini._DNS_DEFAULT_TTL
+
+
+def test_refresher_follows_a_change_and_adds_a_family(monkeypatch):
+    answers = [[(V4, "9.9.9.9"), (V6, "2001:db8::9")], None, [(V4, "9.9.9.9")]]
+    monkeypatch.setattr(mini, "_dns_resolve", lambda *a: answers.pop(0) if len(answers) > 1 else answers[0])
+    monkeypatch.setattr(mini, "_dns_lookup_ttl", lambda *a: None)
+    monkeypatch.setattr(mini, "_DNS_DEFAULT_TTL", 0.01)
+    monkeypatch.setattr(mini, "_DNS_FAILED_RETRY", 0.01)
+
+    async def idle(*a):
+        await asyncio.sleep(10)
+    monkeypatch.setattr(mini, "_heartbeat_sender", idle)
+
+    async def go():
+        mini._running = True
+        mini._shutdown_event = asyncio.Event()
+        mini._active_tasks = []
+        conns = [mini.AsyncConnection(1, "1.1.1.1", 5, V4, "me")]
+        task = asyncio.create_task(mini._dns_refresher(conns, ["h"], 5, 0, "me", 10, None, 2))
+        await asyncio.sleep(0.2)
+        mini._running = False
+        mini._shutdown_event.set()
+        await task
+        for t in mini._active_tasks:
+            t.cancel()
+        return conns
+
+    conns = asyncio.run(go())
+    assert [(c.conn_id, c.addr) for c in conns] == [(1, "9.9.9.9")]
+    assert conns[0].af == V4

tests/test_http_config_api.py

@@ -1,5 +1,4 @@
 """Tests for the config read/write API helpers in http.py."""
-import pytest
 from hbd.server import http
 
 

tests/test_http_users_me.py

@@ -1,5 +1,4 @@
 """Tests for PUT /api/0/users/me logic."""
-import pytest
 from hbd.server import users as users_mod
 
 
@@ -87,7 +86,6 @@ def test_put_users_me_notification_channels(tmp_path):
 
 def test_visible_channels_excludes_private_from_others():
     """Private channels owned by another user must not appear in the visible set."""
-    from hbd.server import settings as settings_mod
 
     config = {
         "notification_channels": {

tests/test_nagios_runner.py

@@ -1,15 +1,11 @@
 import asyncio
 import logging
-import os
 import stat
 
-from hbd.client.plugins.nagios_runner import (
-    NagiosRunnerPlugin,
-    NAGIOS_OK,
-    NAGIOS_WARNING,
-    NAGIOS_CRITICAL,
-    NAGIOS_UNKNOWN,
-)
+from hbd.client.plugins.nagios_runner import NagiosRunnerPlugin, NAGIOS_UNKNOWN
+
+# Nagios plugin exit codes (the module itself only needs UNKNOWN)
+NAGIOS_OK, NAGIOS_WARNING, NAGIOS_CRITICAL = 0, 1, 2
 
 
 def test_no_commands_sets_skip_reason():

tests/test_notification_channels_api.py

@@ -1,5 +1,4 @@
 """Tests for notification channel CRUD via configio helpers and visibility logic."""
-import pytest
 from hbd.server import configio, settings as settings_mod
 
 

tests/test_oauth.py

@@ -37,7 +37,6 @@ def reset_users_dict():
     users_mod.users = original
 
 
-
 def test_make_state_returns_unique_tokens():
     s1 = oauth.make_state()
     s2 = oauth.make_state()
@@ -124,7 +123,6 @@ def test_provision_oauth_user_survives_config_reload():
     assert "oauthonly" in users_mod.users
 
 
-
 # ---------------------------------------------------------------------------
 # Integration-style tests: callback logic chain
 # ---------------------------------------------------------------------------

tests/test_proto.py

@@ -1,19 +1,19 @@
-from hbd.proto import dicttos, stodict, oldmtodict
+from hbd.common.proto import dicttos, stodict, oldmtodict
 
 
-def test_dicttos_and_stodict_uncompressed():
+def test_dicttos_and_stodict_roundtrip():
     msg = dicttos("HTB", {"name": "host.example", "interval": 10})
+    assert msg.startswith(b"!HTB:")
     d = stodict(msg)
-    assert d["ID"].startswith("HTB")
+    assert d["ID"] == "HTB"
     assert d["name"] == "host.example"
     assert d["interval"] == 10
 
 
-def test_dicttos_and_stodict_compressed():
-    msg = dicttos("ACK", {"time": 12345}, compress=True)
-    d = stodict(msg)
-    # for compressed the original code included the colon in ID slice; ensure no crash
-    assert "ID" in d
+def test_dicttos_and_stodict_ack():
+    d = stodict(dicttos("ACK", {"time": 12345}))
+    assert d["ID"] == "ACK"
+    assert d["time"] == 12345
 
 
 def test_oldmtodict():

tests/test_settings_sections.py

@@ -1,4 +1,3 @@
-import pytest
 from hbd.server import settings as settings_mod
 
 CFG = {

tests/test_threshold.py

@@ -13,13 +13,12 @@ Tests cover:
 """
 
 import sys
-import time
 from pathlib import Path
 
 # Add parent directory to path for imports
 sys.path.insert(0, str(Path(__file__).parent.parent))
 
-from hbd.threshold import (
+from hbd.server.threshold import (  # noqa: E402
     ThresholdChecker,
     ThresholdConfig,
     AlertLevel,
@@ -170,23 +169,23 @@ def test_alert_state():
 
     # Update to WARNING - should trigger notification
     changed = alert.update(AlertLevel.WARNING, 85.0)
-    assert changed == True, "State change should return True"
+    assert changed is True, "State change should return True"
     assert alert.level == AlertLevel.WARNING
     assert alert.last_value == 85.0
 
     # Update with same level - no notification
     changed = alert.update(AlertLevel.WARNING, 86.0)
-    assert changed == False, "No state change should return False"
+    assert changed is False, "No state change should return False"
     assert alert.last_value == 86.0
 
     # Escalate to CRITICAL
     changed = alert.update(AlertLevel.CRITICAL, 95.0)
-    assert changed == True, "Escalation should trigger notification"
+    assert changed is True, "Escalation should trigger notification"
     assert alert.level == AlertLevel.CRITICAL
 
     # Recover to OK
     changed = alert.update(AlertLevel.OK, 50.0)
-    assert changed == True, "Recovery should trigger notification"
+    assert changed is True, "Recovery should trigger notification"
     assert alert.level == AlertLevel.OK
 
     print("  ✓ Alert state tracking works correctly")
@@ -257,7 +256,16 @@ def test_threshold_checker_parsing():
     print(f"  ✓ Parsed {len(checker.thresholds)} thresholds correctly")
 
 
-def test_check_plugin_data():
+def _checker_with_notifications(config, notifications, monkeypatch):
+    """ThresholdChecker with no grace period that records notifications instead of sending them."""
+    monkeypatch.setattr(
+        ThresholdChecker, "_send_notification",
+        lambda self, host, lvl, message, *a: notifications.append(f"[{lvl}] {host} {message}"),
+    )
+    return ThresholdChecker({"grace": 0, **config})
+
+
+def test_check_plugin_data(monkeypatch):
     """Test checking plugin data against thresholds."""
     print("\nTest 6: Plugin data checking...")
 
@@ -277,11 +285,7 @@ def test_check_plugin_data():
     }
 
     notifications = []
-    
-    def notification_callback(msg):
-        notifications.append(msg)
-    
-    checker = ThresholdChecker(config, notification_callback=notification_callback)
+    checker = _checker_with_notifications(config, notifications, monkeypatch)
     alert_states = {}
 
     # First check - OK
@@ -353,12 +357,12 @@ def test_check_plugin_data():
 
     assert len(state_changes) == 2, "Two metrics should recover"
     assert len(notifications) == 2, "Two recovery notifications"
-    assert any("RECOVERED" in n for n in notifications), "Should have recovery notification"
+    assert any("RECOVER" in n for n in notifications), "Should have recovery notification"
 
     print("  ✓ Plugin data checking and notifications work")
 
 
-def test_nested_metrics():
+def test_nested_metrics(monkeypatch):
     """Test checking nested metrics like disk partitions."""
     print("\nTest 7: Nested metrics (partitions)...")
 
@@ -384,7 +388,7 @@ def test_nested_metrics():
     }
 
     notifications = []
-    checker = ThresholdChecker(config, notification_callback=lambda m: notifications.append(m))
+    checker = _checker_with_notifications(config, notifications, monkeypatch)
     alert_states = {}
 
     plugin_data = {

tests/test_udp.py

@@ -1,5 +1,5 @@
-from hbd.udp import parse_message
-from hbd.proto import dicttos
+from hbd.server.udp import parse_message
+from hbd.common.proto import dicttos
 
 
 def test_parse_message_uncompressed():

tests/test_udp_stray_datagrams.py

@@ -0,0 +1,40 @@
+"""Datagrams that are not heartbeat traffic (internet scanners) must not create hosts."""
+from hbd.common.proto import dicttos
+from hbd.server import hbdclass
+from hbd.server.udp import handle_datagram, parse_message
+
+
+class _FakeTransport:
+    sent = 0
+
+    def sendto(self, data, addr):
+        _FakeTransport.sent += 1
+
+
+def _ctx():
+    return {"config": {}, "hbdclass": hbdclass, "msg_to_websockets": None,
+            "DEBUG": 0, "verbose": False}
+
+
+def _handle(raw):
+    handle_datagram(parse_message(raw), ("192.0.2.9", 50000), _FakeTransport(), _ctx())
+
+
+def test_sip_invite_does_not_create_a_host():
+    hbdclass.Host.hosts.pop("unknown", None)
+    _handle(b"INVITE sip:10000@192.0.2.1:50003 SIP/2.0\r\nTo: <sip:10000@192.0.2.1>\r\n\r\n")
+    assert "unknown" not in hbdclass.Host.hosts
+
+
+def test_nameless_heartbeat_does_not_create_a_host_or_get_acked():
+    hbdclass.Host.hosts.pop("unknown", None)
+    _FakeTransport.sent = 0
+    _handle(dicttos("HTB", {"id": 0}))
+    assert "unknown" not in hbdclass.Host.hosts
+    assert _FakeTransport.sent == 0
+
+
+def test_named_heartbeat_still_creates_its_host():
+    hbdclass.Host.hosts.pop("stray-ok", None)
+    _handle(dicttos("HTB", {"name": "stray-ok", "id": 0}))
+    assert "stray-ok" in hbdclass.Host.hosts

tox.ini

@@ -21,5 +21,5 @@ commands =
     mypy hbd
 
 [flake8]
-max-line-length = 111
+max-line-length = 165
 extend-ignore = E203