From 8270d2d7cab1e4f36279194ea6c21ed6df54bf69 Mon Sep 17 00:00:00 2001 From: Franck Zoccolo Date: Wed, 5 Aug 2026 11:32:18 +0200 Subject: [PATCH] Add continuous querying and caching Replace one-shot, blind-sleep mDNS lookups with a warm cache backed by persistent background queriers: - PTR (browse): a single ServiceBrowser tracks every service type seen so far (recreated with the enlarged type set when a new one shows up, rather than one browser per type), so zeroconf continuously refreshes it in the background; answers are read straight from the mDNS cache instead of blocking timeout seconds on every query. - TXT/SRV: still resolved via get_service_info() (already cache-first internally), but a background job now periodically re-primes the cache so it doesn't go cold between queries. - A/AAAA (bare hostnames): answers are read from the mDNS cache first; only a first-ever query for a name blocks waiting for a response, with a background job keeping it warm afterwards. Periodic refresh jobs are scheduled by a single heap-based scheduler thread and executed by a fixed pool of 4 worker threads, so the background thread count stays constant no matter how many distinct names/services get queried over the proxy's lifetime. All returned TTLs now reflect the real remaining TTL of the underlying mDNS record instead of a fixed constant, and goodbye packets (TTL=0) are handled for free since eviction is delegated to zeroconf's own cache. Also add .gitignore for the project-local .venv/ and __pycache__/. --- .gitignore | 2 + proxy.py | 174 ++++++++++++++++++++++++++++++++++++++--------------- 2 files changed, 127 insertions(+), 49 deletions(-) create mode 100644 .gitignore diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..a230a78 --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +.venv/ +__pycache__/ diff --git a/proxy.py b/proxy.py index 536a7ea..5ce4679 100644 --- a/proxy.py +++ b/proxy.py @@ -1,9 +1,12 @@ #!/usr/bin/env python3 -from zeroconf import Zeroconf, ServiceBrowser, ServiceStateChange, ServiceInfo, DNSQuestion, DNSOutgoing, RecordUpdateListener, IPVersion -from zeroconf.const import _TYPE_A, _TYPE_AAAA, _CLASS_IN, _FLAGS_QR_QUERY +from zeroconf import Zeroconf, ServiceBrowser, ServiceInfo, DNSQuestion, DNSOutgoing, IPVersion, current_time_millis +from zeroconf.const import _TYPE_A, _TYPE_AAAA, _TYPE_PTR, _TYPE_SRV, _TYPE_TXT, _CLASS_IN, _FLAGS_QR_QUERY import sys import time +import threading +import heapq +import queue from twisted.internet import reactor, defer, threads from twisted.names import client, dns, error, server from twisted.python import log @@ -15,6 +18,8 @@ port = int(sys.argv[2]) ttl = 120 timeout = 2 negative_ttl = 10 +refresh_interval = 90 +refresh_pool_size = 4 ipv6_ula_network = ipaddress.ip_network('fc00::/7') def soa_record(): @@ -26,7 +31,73 @@ def soa_record(): class DynamicResolver(object): def __init__(self): self.zeroconf = Zeroconf(ip_version=IPVersion.All) - + self.lock = threading.Lock() + self.browser = None + self.browser_types = set() + self.interests = set() + self.refresh_heap = [] + self.refresh_queue = queue.Queue() + for _ in range(refresh_pool_size): + threading.Thread(target=self._refresh_worker, daemon=True).start() + threading.Thread(target=self._refresh_scheduler, daemon=True).start() + + def _ensure_browser(self, localname): + """Make sure a persistent ServiceBrowser covers this service type. + A single ServiceBrowser (and thus a single thread) tracks every + type seen so far; when a genuinely new type shows up, it is + recreated with the enlarged type set rather than starting a + second one, so the thread count stays independent of how many + distinct service types get queried. Returns True the first time + this service type is seen.""" + with self.lock: + if localname in self.browser_types: + return False + self.browser_types.add(localname) + if self.browser is not None: + self.browser.cancel() + self.browser = ServiceBrowser(self.zeroconf, list(self.browser_types), [lambda *a, **kw: None]) + return True + + def _keep_warm(self, key, refresh): + """Register (once) a periodic `refresh()` job for `key`, so the + zeroconf cache stays populated between DNS queries instead of + going cold and forcing a fresh mDNS round trip on every lookup. + Jobs are executed by a small fixed-size worker pool rather than + one thread per key. Returns True the first time this key is + seen.""" + with self.lock: + if key in self.interests: + return False + self.interests.add(key) + heapq.heappush(self.refresh_heap, (time.time() + refresh_interval, key, refresh)) + return True + + def _refresh_scheduler(self): + """Single background thread: wakes up due refresh jobs and hands + them to the worker pool via self.refresh_queue.""" + while True: + job = None + with self.lock: + if self.refresh_heap and self.refresh_heap[0][0] <= time.time(): + _, key, refresh = heapq.heappop(self.refresh_heap) + job = (key, refresh) + if job: + self.refresh_queue.put(job) + else: + time.sleep(1) + + def _refresh_worker(self): + """Fixed pool of worker threads (refresh_pool_size of them) that + execute due refresh jobs and reschedule them for their next run.""" + while True: + key, refresh = self.refresh_queue.get() + try: + refresh() + except Exception: + pass + with self.lock: + heapq.heappush(self.refresh_heap, (time.time() + refresh_interval, key, refresh)) + def _dynamicResponseRequired(self, query): if str(query.name).endswith(domain): return True @@ -40,26 +111,19 @@ class DynamicResolver(object): localname = str(query.name)[:-len(domain)] + "local." def browse(localname): - services = [] - def handler(zeroconf, service_type, name, state_change): - if state_change is ServiceStateChange.Added: - services.append(name) - - sb = ServiceBrowser(self.zeroconf, localname, [handler]) - time.sleep(timeout) - sb.cancel() - - answers, additional = [], [] - for service in services: - answers.append(dns.RRHeader(name=localname[:-6] + domain, ttl=ttl, type=dns.PTR, payload=dns.Record_PTR( - name=service[:-6] + domain - ))) - #txt_ans, _, _ = txt(service) - #srv_ans, _, a_ans = srv(service) - #additional += a_ans + txt_ans + srv_ans - return answers, ([] if answers else [soa_record()]), additional + if self._ensure_browser(localname): + time.sleep(timeout) + + now = current_time_millis() + records = [r for r in self.zeroconf.cache.get_all_by_details(localname, _TYPE_PTR, _CLASS_IN) if not r.is_expired(now)] + answers = [dns.RRHeader(name=localname[:-6] + domain, ttl=int(r.get_remaining_ttl(now)), type=dns.PTR, payload=dns.Record_PTR( + name=r.alias[:-6] + domain + )) for r in records] + return answers, ([] if answers else [soa_record()]), [] def txt(localname): + self._keep_warm(('info', localname), lambda: self.zeroconf.get_service_info(localname, localname, timeout*1000)) + if localname.endswith('._device-info._tcp.local.'): info = ServiceInfo(localname, localname) info.request(self.zeroconf, timeout*1000) @@ -80,17 +144,26 @@ class DynamicResolver(object): i += length data = [b"%s=%s" % (p, info.properties[p]) for p in sorted(info.properties, key=lambda k: order.index(k) if k in order else 1000)] - answers = [dns.RRHeader(name=localname[:-6] + domain, ttl=ttl, type=dns.TXT, payload=dns.Record_TXT( + now = current_time_millis() + cached = self.zeroconf.cache.get_by_details(localname, _TYPE_TXT, _CLASS_IN) + record_ttl = int(cached.get_remaining_ttl(now)) if cached else ttl + answers = [dns.RRHeader(name=localname[:-6] + domain, ttl=record_ttl, type=dns.TXT, payload=dns.Record_TXT( *data ))] return answers, [], [] def srv(localname): + self._keep_warm(('info', localname), lambda: self.zeroconf.get_service_info(localname, localname, timeout*1000)) + info = self.zeroconf.get_service_info(localname, localname, timeout*1000) if info is None: return [], [soa_record()], [] - answers = [dns.RRHeader(name=localname[:-6] + domain, ttl=ttl, type=dns.SRV, payload=dns.Record_SRV( + now = current_time_millis() + cached = self.zeroconf.cache.get_by_details(localname, _TYPE_SRV, _CLASS_IN) + srv_ttl = int(cached.get_remaining_ttl(now)) if cached else ttl + + answers = [dns.RRHeader(name=localname[:-6] + domain, ttl=srv_ttl, type=dns.SRV, payload=dns.Record_SRV( info.priority, info.weight, info.port, info.server[:-6] + domain ))] target = info.server[:-6] + domain @@ -104,36 +177,39 @@ class DynamicResolver(object): def host(localname, qtype): if qtype == dns.AAAA: - mdns_type, addr_len, parse_addr = _TYPE_AAAA, 16, ipaddress.IPv6Address + mdns_type, parse_addr = _TYPE_AAAA, ipaddress.IPv6Address keep_addr = lambda addr: addr in ipv6_ula_network else: - mdns_type, addr_len, parse_addr = _TYPE_A, 4, ipaddress.IPv4Address + mdns_type, parse_addr = _TYPE_A, ipaddress.IPv4Address keep_addr = lambda addr: not addr.is_link_local - - class listener(RecordUpdateListener): - def __init__(self): - self.addrs = [] - self.time = time.time() - def update_record(self, zc, now, record): - if record.type == mdns_type and len(record.address) == addr_len: - addr = parse_addr(record.address) - if keep_addr(addr): - self.addrs.append(str(addr)) - - l = listener() - q = DNSQuestion(localname, mdns_type, _CLASS_IN) - self.zeroconf.add_listener(l, q) - out = DNSOutgoing(_FLAGS_QR_QUERY) - out.add_question(q) - self.zeroconf.send(out) - while len(l.addrs) == 0 and time.time() - l.time < timeout: - time.sleep(0.1) - self.zeroconf.remove_listener(l) - record_cls = dns.Record_AAAA if qtype == dns.AAAA else dns.Record_A - answers = [dns.RRHeader(name=query.name.name, ttl=ttl, type=qtype, payload=record_cls( - addr - )) for addr in l.addrs] + + def send_question(): + q = DNSQuestion(localname, mdns_type, _CLASS_IN) + out = DNSOutgoing(_FLAGS_QR_QUERY) + out.add_question(q) + self.zeroconf.send(out) + + def answers_from_cache(): + now = current_time_millis() + result = [] + for r in self.zeroconf.cache.get_all_by_details(localname, mdns_type, _CLASS_IN): + if r.is_expired(now): + continue + addr = parse_addr(r.address) + if keep_addr(addr): + result.append(dns.RRHeader(name=query.name.name, ttl=int(r.get_remaining_ttl(now)), type=qtype, payload=record_cls(str(addr)))) + return result + + is_new = self._keep_warm((localname, mdns_type), send_question) + + answers = answers_from_cache() + if not answers and is_new: + send_question() + deadline = time.time() + timeout + while not answers and time.time() < deadline: + time.sleep(0.1) + answers = answers_from_cache() return answers, ([] if answers else [soa_record()]), []