v0.2.1: Resolve both connection endpoints via DNS + local cache

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "fwtop"
-version = "0.2.0"
+version = "0.2.1"
 description = "Real-time router/firewall traffic visualizer TUI for Linux (interfaces, conntrack, firewall drops)"
 readme = "README.md"
 authors = [

src/fwtop/app.py

@@ -12,6 +12,7 @@
 from fwtop.collector import Collector
 from fwtop.config import ZONE_CYCLE, Config
 from fwtop.resolve import Resolver
+from fwtop.sources.demo import demo_ptr_lookup
 from fwtop.widgets import (
     ConnectionsTable,
     ConntrackPanel,
@@ -197,9 +198,17 @@ def compose(self) -> ComposeResult:
         yield Label("", id="status-label", markup=False)
         yield Footer()
 
+    def _make_resolver(self) -> Resolver:
+        # Production resolves every address (local or public) through the
+        # system DNS resolver. In demo mode the synthetic addresses don't
+        # exist, so swap in a fake DNS backend that returns their PTR records;
+        # the cache/queue path is identical either way.
+        lookup = demo_ptr_lookup if self.collector.demo else None
+        return Resolver(lookup=lookup)
+
     def on_mount(self) -> None:
         if self.resolve_enabled:
-            self.resolver = Resolver()
+            self.resolver = self._make_resolver()
         self._start_time = monotonic()
         self._update_subtitle()
 
@@ -306,7 +315,7 @@ def action_toggle_pause(self) -> None:
     def action_toggle_resolve(self) -> None:
         self.resolve_enabled = not self.resolve_enabled
         if self.resolve_enabled and self.resolver is None:
-            self.resolver = Resolver()
+            self.resolver = self._make_resolver()
         self._update_subtitle()
 
     def action_change_interval(self) -> None:

src/fwtop/resolve.py

@@ -3,20 +3,47 @@
 import queue
 import socket
 import threading
+from typing import Callable, Optional
+
+
+def _ptr_lookup(ip: str) -> str:
+    """Resolve ``ip`` to a hostname via the system resolver (reverse DNS).
+
+    Returns the IP unchanged when there is no PTR record or the lookup fails.
+    This goes through the OS resolver, so on a router whose local DNS serves
+    PTR records for its DHCP leases, private addresses resolve here too — no
+    hosts file or static map required.
+    """
+    try:
+        return socket.gethostbyaddr(ip)[0]
+    except Exception:
+        return ip
 
 
 class Resolver:
-    """Background reverse-DNS resolver with a cache.
+    """Background reverse-DNS resolver with a program-local cache.
 
-    ``display(ip)`` returns the resolved hostname if it is already known,
+    ``display(ip)`` returns the resolved hostname if it is already cached,
     otherwise the IP itself while scheduling a lookup in the background so a
     later call can show the name. Lookups never block the caller — the UI keeps
-    showing the raw IP until the PTR record (if any) comes back.
+    showing the raw IP until the PTR record (if any) comes back. Every result
+    (including "no PTR -> the IP itself") is cached in an in-memory map, so each
+    address is looked up at most once.
+
+    All resolution goes through DNS via ``lookup`` (the system resolver by
+    default). The hook exists only so tests and the synthetic ``--demo`` mode
+    can stand in a fake DNS backend for addresses that don't exist for real; it
+    is not a static names table.
     """
 
-    def __init__(self, max_workers: int = 4) -> None:
+    def __init__(
+        self,
+        max_workers: int = 4,
+        lookup: Optional[Callable[[str], str]] = None,
+    ) -> None:
         self._cache: dict[str, str] = {}  # ip -> hostname (or ip if no PTR)
         self._pending: set[str] = set()
+        self._lookup = lookup or _ptr_lookup
         self._lock = threading.Lock()
         self._queue: queue.Queue[str] = queue.Queue()
         self._stop = threading.Event()
@@ -45,10 +72,7 @@ def _worker(self) -> None:
                 ip = self._queue.get(timeout=0.5)
             except queue.Empty:
                 continue
-            try:
-                host = socket.gethostbyaddr(ip)[0]
-            except Exception:
-                host = ip  # no PTR record or lookup failed -> fall back to IP
+            host = self._lookup(ip)
             with self._lock:
                 self._cache[ip] = host
                 self._pending.discard(ip)

src/fwtop/sources/demo.py

@@ -20,6 +20,28 @@
 _WAN_HOSTS = ("203.0.113.9", "198.51.100.23", "8.8.8.8", "1.1.1.1", "140.82.121.4")
 _WAN_IP = "203.0.113.2"  # the router's public address (NAT egress)
 
+# PTR answers a router's local DNS would return for the synthetic hosts. The
+# demo's addresses are fictional (TEST-NET / RFC1918), so a real reverse-DNS
+# lookup can't resolve them; demo_ptr_lookup() stands in as that DNS backend
+# so the resolver still exercises its real cache/lookup path.
+_DEMO_PTR = {
+    "192.168.1.10": "desktop.lan",
+    "192.168.1.42": "laptop.lan",
+    "192.168.1.77": "nas.lan",
+    "10.0.10.5": "iot-hub.lan",
+    "203.0.113.2": "router.wan",
+    "203.0.113.9": "vpn-peer.example.net",
+    "198.51.100.23": "mail.example.org",
+    "8.8.8.8": "dns.google",
+    "1.1.1.1": "one.one.one.one",
+    "140.82.121.4": "lb-140-82-121-4-fra.github.com",
+}
+
+
+def demo_ptr_lookup(ip: str) -> str:
+    """Stand-in DNS backend for the demo: return the simulated PTR or the IP."""
+    return _DEMO_PTR.get(ip, ip)
+
 
 class DemoSource:
     """Synthetic data source producing plausible, time-varying router metrics.