diff --git a/README.md b/README.md index b914621..3d3972c 100644 --- a/README.md +++ b/README.md @@ -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). diff --git a/hbd/client/dnsttl.py b/hbd/client/dnsttl.py new file mode 100644 index 0000000..6270222 --- /dev/null +++ b/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 diff --git a/hbd/client/main.py b/hbd/client/main.py index 011e286..29b9d4c 100644 --- a/hbd/client/main.py +++ b/hbd/client/main.py @@ -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 @@ -95,6 +96,14 @@ class AsyncConnection: self.transport.close() 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 +509,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") @@ -640,6 +691,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)) diff --git a/scripts/c/Makefile b/scripts/c/Makefile index 2831662..c2a19c5 100644 --- a/scripts/c/Makefile +++ b/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 diff --git a/scripts/c/hbc_mini.c b/scripts/c/hbc_mini.c index 6316912..1bc0aa8 100644 --- a/scripts/c/hbc_mini.c +++ b/scripts/c/hbc_mini.c @@ -20,6 +20,7 @@ #endif #include +#include #include #include #include @@ -27,6 +28,7 @@ #include #include #include +#include #include #include #include @@ -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++) { diff --git a/scripts/c/test_probe.c b/scripts/c/test_probe.c index 3d2e570..2f8626d 100644 --- a/scripts/c/test_probe.c +++ b/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; } diff --git a/scripts/hbc_mini.py b/scripts/hbc_mini.py index 3863b0f..9477007 100755 --- a/scripts/hbc_mini.py +++ b/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 @@ -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))) diff --git a/tests/test_client_dnsttl.py b/tests/test_client_dnsttl.py new file mode 100644 index 0000000..8ebe351 --- /dev/null +++ b/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")] diff --git a/tests/test_hbc_mini_dnsttl.py b/tests/test_hbc_mini_dnsttl.py new file mode 100644 index 0000000..fac2496 --- /dev/null +++ b/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