diff --git a/config.py b/config.py index a1324cc..ada7ed8 100644 --- a/config.py +++ b/config.py @@ -204,6 +204,46 @@ "208.67.222.222", # OpenDNS ] +# DNS interception detection. See docs/dns_interception_srd.md. +# +# A macOS DNS proxy network extension (Tailscale's, a corporate VPN client, a +# captive portal) captures DNS flows system-wide, so a query to ANY address on +# port 53 is answered locally. That makes every port-53 reachability check +# report success unconditionally: measured 2026-08-29, a query to 240.0.0.1 +# came back NOERROR in under 10 ms. +# +# 240.0.0.1 is in reserved space (240.0.0.0/4) and is not routable, so nothing +# can legitimately answer from it. Any response at all means DNS is intercepted. +# The timeout is short because a clean host pays it in full on every probe, +# while an intercepted one answers almost immediately. +DNS_INTERCEPTION_PROBE_ADDRESS = "240.0.0.1" +DNS_INTERCEPTION_PROBE_TIMEOUT = 1.5 +DNS_INTERCEPTION_PROBE_NAME = "example.com" + +# DNS over TLS. Not intercepted by a DNS proxy extension, so this is how a +# resolver's reachability can actually be tested on such a host. +DOT_PORT = 853 +DOT_QUERY_TIMEOUT = 6 + +# The internet check needs one TCP handshake with a host on the internet; it +# does not need DNS. It used port 53, which made it unable to fail on an +# intercepted host. Port 443 is not intercepted (measured: documentation +# addresses time out, real hosts connect). +INTERNET_CHECK_PORT = 443 +INTERNET_CHECK_TIMEOUT = 3 +INTERNET_CHECK_HOSTS = [ + "1.1.1.1", # Cloudflare + "8.8.8.8", # Google + "9.9.9.9", # Quad9 +] + +# Unprivileged ICMP (SOCK_DGRAM with IPPROTO_ICMP) works without root on macOS. +# Used only as a weaker, clearly-labelled signal where no DNS signal is +# trustworthy. Note G-root and L-root do not answer ICMP at all, so a +# non-response is never reported as "down". +ICMP_PROBE_TIMEOUT = 3 +ICMP_PROBE_ATTEMPTS = 2 + # Common websites for connectivity testing CONNECTIVITY_TEST_SITES = [ "https://www.google.com", diff --git a/docs/dns_interception_srd.md b/docs/dns_interception_srd.md new file mode 100644 index 0000000..e20d062 --- /dev/null +++ b/docs/dns_interception_srd.md @@ -0,0 +1,293 @@ +# SRD: DNS Checks That Cannot Fail + +## 1. Purpose + +Three tools report success unconditionally on any host running a macOS DNS +proxy network extension. `check_internet_connection` will report "Connected" +with the WAN cable unplugged. `check_dns_root_servers` and `check_dns_resolvers` +will report every server reachable when none of them is. + +This document records what was measured, fixes the two tools that can be fixed, +and makes the third report honestly that it cannot run rather than passing. + +## 2. Background + +### 2.1 How it was found + +A consumer of this toolkit (a voice assistant that speaks tool results aloud) +removed a language model from its narration path, cutting a network-check turn +from roughly 22 seconds to under one. The operator noticed that a check of +thirteen root servers across the internet was answering in under a second, +including a subprocess spawn, and asked why. Nothing in this repository changed. + +### 2.2 What is intercepted + +Measured 2026-08-29 on macOS 25.6.0. Every row has a negative control: an +address that must not be able to answer. `240.0.0.1` is in reserved space and is +not routable; `192.0.2.1`, `198.51.100.1` and `203.0.113.1` are RFC 5737 +documentation ranges. + +| Signal | Real host | Negative control | Verdict | +| --- | --- | --- | --- | +| UDP/53 | answers | **answers** | intercepted | +| TCP/53 | answers | **answers** | intercepted | +| UDP/53, `IP_BOUND_IF` to en0 | `ENXIO` | `ENXIO` | no bypass | +| UDP/53, source bound to en0 | answers | **answers** | no bypass | +| TCP/443 | connects | times out | clean | +| TCP/853 (DoT) | connects | times out | clean | +| ICMP echo | replies 2.6-65.9 ms | times out | clean | + +The answers returned are *correct* — they match Cloudflare DoH byte for byte, +and a nonexistent name correctly yields NXDOMAIN — so this is a real resolver +answering on behalf of every destination, not a wildcard hijacker. + +### 2.3 What is doing it + +Nothing is listening on port 53 (`lsof -nP -iUDP:53 -iTCP:53` is empty) and the +routing table is ordinary (default via `en0` to the LAN gateway, for the root +server addresses and for `240.0.0.1` alike). The host runs Tailscale, whose +`io.tailscale.ipn.macsys.network-extension` is loaded, with `scutil --dns` +reporting nameserver `100.100.100.100` and a `.ts.net` search domain. + +That is a **`NEDNSProxyProvider`**: a macOS DNS proxy network extension, which +captures DNS flows system-wide at the flow level rather than by binding a port +or installing a route. This is why no socket option reaches it. It is not a +misconfiguration and it is working as designed; the defect is that these tools +cannot tell. + +**Do not treat this as Tailscale-specific.** Any DNS proxy extension, corporate +VPN client, or captive portal produces the same result. The fix must detect the +*condition*, not the vendor. + +### 2.4 The consequence, stated plainly + +A check that cannot fail is worse than no check. It converts "I do not know" +into "everything is fine", and it does so at exactly the moment a user is asking +because they suspect it is not. + +## 3. Scope + +In scope: + +- `check_internet_connection`: move off port 53 entirely. +- `check_dns_resolvers`: query over DoT, which is not intercepted. +- `check_dns_root_servers`: detect interception and refuse to pass; report ICMP + reachability as a clearly-labelled weaker signal. +- A shared `dns_interception_detected()` helper. +- A third result state, "could not be checked", distinct from reachable and + unreachable. + +Out of scope: + +- Defeating the extension. Section 5.1 records that it cannot be done from an + unprivileged process, with the measurements behind that. +- Changing anyone's Tailscale configuration. +- IPv6. The tables here are IPv4 and the probes were IPv4. +- The other DNS tools in the registry (`dns_diagnostics.py`, reverse lookups), + which resolve names rather than test specific servers and are not making a + reachability claim. + +## 4. Goals + +1. No tool reports a server reachable on the strength of a query that never left + the machine. +2. Every reachability claim is backed by a signal with a passing negative + control. +3. "Could not be checked" is a distinct outcome and never renders as success. +4. The detection works for any DNS proxy, not just Tailscale. +5. Tools that can be genuinely fixed are genuinely fixed, not merely annotated. + +## 5. Architecture + +### 5.1 There is no bypass, and this is measured rather than assumed + +Two approaches were tried and both are recorded so they are not re-proposed: + +- **`IP_BOUND_IF`** (`setsockopt(IPPROTO_IP, 25, if_index)` on `en0`) failed with + `OSError: [Errno 6] Device not configured` for every destination including a + real root server, so it did not even reach the question. +- **Source-address binding** (`bind(("192.168.1.244", 0))`) succeeded in sending, + and the negative control still answered. The flow is captured regardless of + which interface address it leaves from. + +This follows from what a `NEDNSProxyProvider` is: the system hands matching +flows to the extension before they reach the network stack's routing decision. +An unprivileged process cannot opt out. Do not spend time here again. + +### 5.2 Detection: a negative control at runtime + +``` +dns_interception_detected() -> bool +``` + +Sends one A query for a fixed name to `240.0.0.1` — reserved space, not +routable, guaranteed to have nothing behind it — with a short timeout. **Any +response at all means DNS is being intercepted**, because a response is +impossible otherwise. + +Three properties matter: + +- It is cheap in the case that matters. When interception is present the reply + arrives in under 10 ms, so the tools that need to know pay almost nothing. + When it is absent the probe costs its timeout, so the timeout is short + (`DNS_INTERCEPTION_PROBE_TIMEOUT`, 1.5 s) and the result is cached for the + process lifetime. +- It tests the condition, not the vendor. A corporate VPN or captive portal that + answers for every destination is caught by the same probe. +- It fails toward *reporting a problem*, not toward silence. If the probe itself + errors, the answer is "unknown", which the callers treat as "cannot verify" — + the same direction as detecting interception, never as an all-clear. + +### 5.3 `check_internet_connection`: move off port 53 + +Currently `socket.create_connection((DNS_TEST_SERVERS[0], 53), timeout=3)`. The +port is incidental — nothing about this check needs DNS, it needs one TCP +handshake with a host on the internet. + +Port 443 is not intercepted (measured: `1.1.1.1:443` connects, both +documentation addresses time out at 4 s), so the fix is to connect on 443 and to +try more than one host before concluding the internet is down. This tool is +fully repaired: it will report Disconnected when the WAN is down, which it +cannot do today. + +### 5.4 `check_dns_resolvers`: query over DoT + +DoT (TCP/853) is not intercepted and reaches the real resolver. Measured against +all ten entries in `DEFAULT_DNS_RESOLVERS`: + +| Resolver | TCP/853 | DoT query | +| --- | --- | --- | +| Google primary and secondary | open | answered | +| Cloudflare primary and secondary | open | answered | +| OpenDNS primary and secondary | open | answered | +| Quad9 primary and secondary | open | answered | +| **Comodo primary and secondary** | **timeout** | **not available** | +| negative controls | timeout | timed out | + +Eight of ten are genuinely checkable. **Comodo Secure DNS offers no DoT**, and +its port 53 answer is the interceptor's, so on an intercepted host it cannot be +checked at all. It is reported as `UNKNOWN`, not as reachable and not as +unreachable — see 5.6. + +Each resolver needs its DoT hostname for certificate validation, so +`DEFAULT_DNS_RESOLVERS` gains a per-entry hostname. Where a resolver has none, +that is the marker that it is not DoT-capable. + +When `dns_interception_detected()` is false, plain port-53 queries are used as +before: they are faster, they test the actual service on the actual port, and +there is nothing wrong with them on a clean host. + +### 5.5 `check_dns_root_servers`: it cannot be fixed, so it must not pass + +The root servers are the one case with no repair available. + +- They serve DNS on port 53 only, which is captured. +- Encrypted DNS is not a uniform escape route. Measured across five roots: B + accepts TCP/853 but presents a certificate that does not verify, F accepts + TCP/443, A and K time out on both, M refuses on 443. There is no path that + works for all thirteen. + +So under interception this tool reports that the check **could not be performed** +and why. It never reports the roots reachable on the strength of a query that +did not leave the machine. + +ICMP is offered alongside, explicitly labelled as a weaker claim — it shows a +host answers a ping, not that it serves DNS. Two caveats, both measured, and +both of which must reach the output rather than being smoothed over: + +- **G-root (192.112.36.4) and L-root (199.7.83.42) do not answer ICMP**, across + repeated attempts with retries. Eleven of thirteen reply, in 2.6 to 65.9 ms. + Reporting 11/13 as "two root servers are down" would be a false negative, so + non-response to ICMP is reported as "did not answer a ping", never as down. +- **Reply validation is load-bearing.** A first version of the probe counted any + received ICMP packet as a reply, and a router's destination-unreachable for + `240.0.0.1` read as a 2310 ms success. A reply counts only if it is an echo + reply (type 0) carrying our own payload and coming from the address queried. + The negative controls exist to catch exactly this, and did. + +Unprivileged ICMP works on macOS via `SOCK_DGRAM` with `IPPROTO_ICMP`, so no +elevation is needed. On a platform where that raises `PermissionError`, ICMP is +skipped and only the interception verdict is reported. + +### 5.6 A third result state + +Both DNS tools return three states per server rather than two: + +| State | Meaning | +| --- | --- | +| `REACHABLE` | A signal with a passing negative control confirmed it | +| `UNREACHABLE` | Such a signal was attempted and failed | +| `UNKNOWN` | No trustworthy signal was available from this host | + +`UNKNOWN` is the whole point of this document. It covers Comodo on any host, and +every root server on an intercepted one. Summary lines must never fold `UNKNOWN` +into either of the others, and the count of `UNKNOWN` servers is always stated. + +### 5.7 A predicted defect, deliberately not asserted + +Root servers are not recursive: queried for `example.com A` they return a +*referral* to the `.com` nameservers, with an empty answer section. +`dns.resolver.Resolver.resolve()` raises `NoAnswer` on an empty answer section. +If both hold, `check_dns_server` has always reported every root server +unreachable on a clean host, and only the interception has been masking it. + +**This is a prediction, not a measurement.** It could not be verified here: the +interception prevents seeing a real root response, and the one root accepting +DoT presents an unverifiable certificate. It is recorded rather than acted on. + +The redesign sidesteps the question regardless by querying `.` `NS`, which the +roots answer authoritatively from the zone they actually serve. That is the +correct query for "is this root server serving" whether or not the prediction +holds. Confirm it on a host with no DNS proxy before closing this section. + +## 6. Configuration + +New constants in `config.py`: + +| Constant | Value | Meaning | +| --- | --- | --- | +| `DNS_INTERCEPTION_PROBE_ADDRESS` | `240.0.0.1` | Reserved; nothing may answer | +| `DNS_INTERCEPTION_PROBE_TIMEOUT` | `1.5` | Short: the clean case pays this | +| `DOT_PORT` | `853` | DNS over TLS | +| `INTERNET_CHECK_PORT` | `443` | Was 53, which is intercepted | +| `INTERNET_CHECK_HOSTS` | 3 addresses | More than one before declaring down | +| `ICMP_PROBE_TIMEOUT` | `3` | Per attempt | + +## 7. Failure modes + +| Failure | Behaviour | +| --- | --- | +| DNS intercepted | Root check reports UNKNOWN with the reason; resolvers use DoT | +| Interception probe errors | Treated as "cannot verify", never as clean | +| DoT unavailable for a resolver | That resolver is UNKNOWN, others unaffected | +| ICMP needs root on this platform | ICMP skipped; interception verdict still reported | +| A root does not answer ICMP | "did not answer a ping", never "down" | +| Everything clean | Plain port-53 queries, as today, but for `.` NS | + +No row produces a false pass. That is the one property this document exists to +guarantee. + +## 8. Testing + +- `dns_interception_detected()` against a stub that answers the reserved address + (must report intercepted) and one that times out (must report clean). +- The ICMP reply validator against an echo reply, a destination-unreachable, a + reply from a different address, and a truncated packet. Only the first counts. +- Three-state reporting: a summary containing an `UNKNOWN` must not claim all + servers reachable, asserted for every summary-line generator. +- `check_internet_connection` against an unreachable host list must return + Disconnected — the case that is impossible to produce today. + +## 9. Known limitations + +1. **The root server check cannot be repaired on an intercepted host.** It can + only be honest. Turning off the DNS proxy's "override local DNS" setting is + the only way to restore it, and that is the operator's decision, not this + tool's. +2. **DoT tests the resolver's DoT service**, which is not byte-for-byte the same + service as its port 53. A resolver could in principle serve one and not the + other. This is a far smaller gap than the one it replaces. +3. **Comodo cannot be checked from an intercepted host at all.** +4. Section 5.7's prediction is unverified. +5. Measured on one host, one OS version, one VPN client. The negative-control + discipline is what makes the result portable, not the specific numbers. diff --git a/network_diagnostics.py b/network_diagnostics.py index 492acc4..1c1d4a5 100644 --- a/network_diagnostics.py +++ b/network_diagnostics.py @@ -27,7 +27,9 @@ from utils import create_success_result, create_error_result, wrap_legacy_result, standardize_tool_output # Import centralized configuration -from config import get_dns_servers, DNS_TEST_SERVERS, COMMON_PORTS +from config import (get_dns_servers, DNS_TEST_SERVERS, COMMON_PORTS, + INTERNET_CHECK_HOSTS, INTERNET_CHECK_PORT, + INTERNET_CHECK_TIMEOUT) # Import v3 pentest tools try: @@ -608,13 +610,34 @@ def _get_all_interfaces_mac(system: str) -> Dict[str, str]: @standardize_tool_output() def check_internet_connection() -> str: - """Check if the internet is reachable""" - try: - # Try to connect to a reliable server - socket.create_connection((DNS_TEST_SERVERS[0], 53), timeout=3) - return "Connected" - except Exception: - return "Disconnected" + """Check if the internet is reachable. + + Opens a TCP connection to a well-known host and reports whether it + succeeded. Tries several hosts before concluding the internet is down, so + one provider having a bad day is not read as an outage. + + This used to connect on port 53, which made it unable to fail. A macOS DNS + proxy network extension captures port 53 for every destination, so the + connection always succeeded locally and the tool reported "Connected" with + the WAN unplugged. Port 443 is not intercepted: measured 2026-08-29, + documentation-range addresses time out on 443 while real hosts connect in + milliseconds. Nothing about this check needs DNS -- it needs one handshake + with something on the internet -- so the port was incidental all along. + See docs/dns_interception_srd.md section 5.3. + + Returns: + str: "Connected" if any host accepted a connection, otherwise + "Disconnected". + """ + for host in INTERNET_CHECK_HOSTS: + try: + connection = socket.create_connection( + (host, INTERNET_CHECK_PORT), timeout=INTERNET_CHECK_TIMEOUT) + connection.close() + return "Connected" + except OSError: + continue + return "Disconnected" @standardize_tool_output() diff --git a/network_tools/dns_check.py b/network_tools/dns_check.py index c42df56..91c1671 100644 --- a/network_tools/dns_check.py +++ b/network_tools/dns_check.py @@ -7,12 +7,20 @@ provide the IP addresses of individual domain names. """ +import dns.exception +import dns.message +import dns.query import dns.resolver import time from datetime import datetime from typing import Dict, List, Tuple, Optional from colorama import Fore, Style +from .dns_interception import ( + describe_icmp_result, dns_interception_detected, icmp_reachable, + interception_notice, +) + # List of DNS root servers with their IP addresses DNS_ROOT_SERVERS = { "A": "198.41.0.4", @@ -31,46 +39,94 @@ } -def check_dns_server(name: str, ip: str, query_name: str = "example.com") -> Tuple[bool, Optional[str]]: +def check_dns_server(name: str, ip: str, query_name: str = ".") -> Tuple[bool, Optional[str]]: """Check if a specific DNS root server is reachable. - + + Queries the server directly for the root zone's NS records, using a plain + DNS message rather than a Resolver. + + Both of those are deliberate. A root server is not recursive: asked for + 'example.com A' it returns a REFERRAL to the .com nameservers with an empty + answer section, and dns.resolver.Resolver.resolve() raises NoAnswer on an + empty answer section -- so the previous implementation may have been + reporting every root server unreachable whenever it was not being masked by + a local DNS proxy answering on their behalf. That is a prediction rather + than a measurement (it could not be confirmed on the intercepted host where + this was written; see docs/dns_interception_srd.md section 5.7), but '.' NS + is the correct question either way: it asks the server for the zone it + actually serves, and it is answered authoritatively. + Args: name: The name of the root server (e.g., "A", "B", etc.) ip: The IP address of the root server - query_name: The domain name to query (default: example.com) - + query_name: The name to query (default: the root zone) + Returns: Tuple containing: - Boolean indicating if server is reachable - Error message (None if reachable) """ try: - resolver = dns.resolver.Resolver() - resolver.timeout = 5 - resolver.lifetime = 5 - resolver.nameservers = [ip] - query_response = resolver.resolve(query_name, "A") + query = dns.message.make_query(query_name, "NS") + response = dns.query.udp(query, ip, timeout=5) + if not (response.answer or response.authority): + raise dns.exception.DNSException("empty response") print(f"{Fore.GREEN} - Successfully queried the '{name}' root server at {ip} for '{query_name}'{Style.RESET_ALL}") return True, None - except Exception as e: + except (OSError, dns.exception.DNSException) as e: print(f"{Fore.RED} - Failed to query {name} root server at {ip}: {e}{Style.RESET_ALL}") return False, str(e) -def check_dns_root_servers(servers: Optional[Dict[str, str]] = None, retry_failed: bool = True) -> Tuple[List[str], List[str]]: +def _icmp_fallback(servers: Dict[str, str]) -> List[str]: + """Probe each root server with ICMP when no DNS signal can be trusted. + + This answers a different and weaker question than the DNS check: whether the + host replies to a ping, not whether it is serving the root zone. Callers + must present it that way. + + Note that some root servers legitimately drop ICMP -- G (192.112.36.4) and + L (199.7.83.42) both did on every attempt, while the other eleven replied in + 2.6 to 65.9 ms -- so a non-reply is reported as "did not answer a ping" and + never as "down". + + Args: + servers: Mapping of root server name to IP address. + + Returns: + One description line per server, for the "could not be checked" section. + """ + descriptions = [] + for name, ip in servers.items(): + answered, round_trip_ms = icmp_reachable(ip) + print(describe_icmp_result(f"'{name}' root server", ip, answered, + round_trip_ms)) + if answered: + descriptions.append(f"- {name} ({ip}) - DNS not testable here; " + f"answers a ping in {round_trip_ms:.0f} ms") + else: + descriptions.append(f"- {name} ({ip}) - DNS not testable here; " + f"did not answer a ping either") + return descriptions + + +def check_dns_root_servers(servers: Optional[Dict[str, str]] = None, retry_failed: bool = True) -> Tuple[List[str], List[str], List[str]]: """Check if DNS root servers are reachable. - + Args: servers: Optional dictionary of DNS root servers to check (name -> IP) If None, uses the default DNS_ROOT_SERVERS retry_failed: Whether to retry unreachable servers after a delay - + Returns: Tuple containing: - List of reachable server descriptions - List of unreachable server descriptions + - List of servers that could not be checked at all. This third list + is the point: on a host where DNS is intercepted it holds every + server, and it must never be folded into either of the others. """ if servers is None: servers = DNS_ROOT_SERVERS @@ -78,6 +134,16 @@ def check_dns_root_servers(servers: Optional[Dict[str, str]] = None, retry_faile reachable_servers = [] unreachable_servers = [] + # The root servers serve DNS on port 53 and nothing else, and they offer no + # uniform encrypted alternative (measured: B accepts 853 with a certificate + # that does not verify, F accepts 443, A and K time out, M refuses). So on a + # host where a DNS proxy answers for every destination there is no way to + # test them at all -- and the honest report is that the check could not be + # run, never that they are fine. ICMP is offered instead as an explicitly + # weaker signal. See docs/dns_interception_srd.md section 5.5. + if dns_interception_detected() is not False: + return [], [], _icmp_fallback(servers) + # First round of checks for name, ip in servers.items(): is_reachable, error = check_dns_server(name, ip) @@ -101,16 +167,21 @@ def check_dns_root_servers(servers: Optional[Dict[str, str]] = None, retry_faile reachable_servers.append(f"- {name_part} ({ip_part})") unreachable_servers = new_unreachable - return reachable_servers, unreachable_servers + return reachable_servers, unreachable_servers, [] -def generate_dns_report(reachable: List[str], unreachable: List[str]) -> str: +def generate_dns_report(reachable: List[str], unreachable: List[str], + unknown: Optional[List[str]] = None) -> str: """Generate a formatted report of DNS root server reachability. - + Args: reachable: List of reachable server descriptions unreachable: List of unreachable server descriptions - + unknown: List of servers that could not be checked at all. Reported in + its own section and never folded into either of the others: a + server nobody could reach a verdict on is not a server that + answered, and it is not one that failed. + Returns: str: Formatted report """ @@ -129,7 +200,26 @@ def generate_dns_report(reachable: List[str], unreachable: List[str]) -> str: for server in unreachable: report += server + "\n" - if not unreachable: + unknown = unknown or [] + if unknown: + report += "\n" + interception_notice() + "\n" + report += ("The root servers offer no encrypted alternative to port 53, " + "so their DNS reachability cannot be tested from this host at " + "all. Ping results below answer a weaker question -- whether " + "the host replies -- and some root servers drop ICMP even " + "when healthy.\n") + report += "\nDNS Root Servers That Could Not Be Checked:\n" + for server in unknown: + report += server + "\n" + + if unknown and not reachable and not unreachable: + report += (f"\nDNS Root Servers reachability summary: could not be " + f"determined for any of the {len(unknown)} root servers.\n") + elif unknown: + report += (f"\nDNS Root Servers reachability summary: {len(reachable)} " + f"reachable, {len(unreachable)} unreachable, {len(unknown)} " + f"could not be checked.\n") + elif not unreachable: report += "\nDNS Root Servers reachability summary: All DNS Root Servers are reachable.\n" else: report += "\nDNS Root Servers reachability summary: Some DNS Root Servers are unreachable.\n" @@ -149,8 +239,8 @@ def main(silent: bool = False, polite: bool = False) -> str: """ print(f"Starting DNS Root Servers check at {datetime.now()}\n") - reachable, unreachable = check_dns_root_servers(DNS_ROOT_SERVERS) - report = generate_dns_report(reachable, unreachable) + reachable, unreachable, unknown = check_dns_root_servers(DNS_ROOT_SERVERS) + report = generate_dns_report(reachable, unreachable, unknown) # Only print detailed output if not in silent mode if not silent: diff --git a/network_tools/dns_interception.py b/network_tools/dns_interception.py new file mode 100644 index 0000000..a958e0f --- /dev/null +++ b/network_tools/dns_interception.py @@ -0,0 +1,227 @@ +"""DNS interception detection and unprivileged ICMP reachability. + +A macOS DNS proxy network extension -- Tailscale's, a corporate VPN client, a +captive portal -- captures DNS flows system-wide at the flow level, so a query +addressed to ANY IP on port 53 is answered locally. Every port-53 reachability +check on such a host therefore reports success unconditionally, including for +servers that are completely unreachable. + +This module provides the two signals the DNS tools need in order to tell the +difference: a runtime negative control that detects the condition, and an ICMP +probe that is not intercepted and can stand in as a weaker claim. + +Full design and the measurements behind it: docs/dns_interception_srd.md. +""" + +import os +import socket +import struct +import time +from typing import Optional, Tuple + +import dns.exception +import dns.message +import dns.query +from colorama import Fore, Style + +from config import ( + DNS_INTERCEPTION_PROBE_ADDRESS, + DNS_INTERCEPTION_PROBE_NAME, + DNS_INTERCEPTION_PROBE_TIMEOUT, + ICMP_PROBE_ATTEMPTS, + ICMP_PROBE_TIMEOUT, +) + +# Result states. UNKNOWN is the reason this module exists: it is what a check +# must report when no trustworthy signal was available, and it must never be +# folded into either of the other two. +REACHABLE = "REACHABLE" +UNREACHABLE = "UNREACHABLE" +UNKNOWN = "UNKNOWN" + +# The probe is stable for the life of a process and costs a full timeout on a +# clean host, so it is answered once. +_interception_cache: Optional[bool] = None + +# Carried in the ICMP payload so a reply can be told apart from an unrelated +# packet arriving on the same socket. +_ICMP_MAGIC = b"instability-reach" + + +def dns_interception_detected(force: bool = False) -> Optional[bool]: + """Detect whether DNS on this host is being intercepted. + + Sends one A query to a reserved, unroutable address. Nothing can legitimately + answer from there, so any response at all proves a local proxy is answering + on behalf of every destination -- which means no port-53 reachability check + on this host can be trusted. + + Args: + force: Re-probe instead of using the per-process cached answer. + + Returns: + True if DNS is intercepted, False if it appears clean, or None if the + probe could not be carried out. None means "cannot verify" and callers + must treat it the same way they treat True: as grounds to withhold a + pass, never as an all-clear. + """ + global _interception_cache + if _interception_cache is not None and not force: + return _interception_cache + + query = dns.message.make_query(DNS_INTERCEPTION_PROBE_NAME, "A") + try: + dns.query.udp(query, DNS_INTERCEPTION_PROBE_ADDRESS, + timeout=DNS_INTERCEPTION_PROBE_TIMEOUT) + except dns.exception.Timeout: + # The expected result on a healthy host: the packet went out and nothing + # came back, because nothing is there. + _interception_cache = False + return False + except OSError: + # No route, or the network is down entirely. That is not evidence either + # way about interception. + return None + except dns.exception.DNSException: + # A malformed or unexpected response is still a response, and nothing + # should be responding at all. + _interception_cache = True + return True + else: + _interception_cache = True + return True + + +def interception_notice() -> str: + """One line explaining why a DNS-based check could not be trusted.""" + return (f"DNS on this host is being intercepted: a query to " + f"{DNS_INTERCEPTION_PROBE_ADDRESS} (reserved, unroutable) was " + f"answered. A DNS proxy is replying for every destination, so " + f"port 53 reachability cannot be tested from here.") + + +def _checksum(data: bytes) -> int: + """The standard internet checksum over an ICMP message.""" + if len(data) % 2: + data += b"\x00" + total = 0 + for index in range(0, len(data), 2): + total += (data[index] << 8) + data[index + 1] + total = (total >> 16) + (total & 0xFFFF) + return ~(total + (total >> 16)) & 0xFFFF + + +def _is_echo_reply(packet: bytes) -> bool: + """Whether a received packet is an echo reply to one of our own probes. + + Load-bearing, and not defensive tidiness. A first version of this probe + counted any received ICMP packet as a reply, so a router's + destination-unreachable for the unroutable test address read as a 2310 ms + success. Only an echo reply (type 0) carrying our payload counts. + + macOS may or may not prepend the IP header on a SOCK_DGRAM ICMP socket, so + both offsets are tried. + + Args: + packet: The bytes returned by recvfrom. + + Returns: + True only for a genuine echo reply to one of our probes. + """ + for offset in (0, 20): + if len(packet) < offset + 8: + continue + if packet[offset] == 0 and _ICMP_MAGIC in packet[offset:]: + return True + return False + + +def icmp_reachable(host: str, timeout: int = None, + attempts: int = None) -> Tuple[bool, Optional[float]]: + """Whether a host answers an ICMP echo request, and how quickly. + + Unprivileged: macOS allows SOCK_DGRAM with IPPROTO_ICMP without root, which + is what lets this run from the same process as everything else. + + This proves a host is reachable. It does NOT prove the host is serving DNS, + and a caller must not present it as though it did. Note also that some hosts + legitimately drop ICMP -- G-root and L-root both do -- so a False here means + "did not answer a ping", never "down". + + Args: + host: The IP address to probe. + timeout: Seconds to wait per attempt. + attempts: How many echo requests to send before giving up. + + Returns: + (answered, round_trip_ms). round_trip_ms is None when answered is False. + """ + timeout = ICMP_PROBE_TIMEOUT if timeout is None else timeout + attempts = ICMP_PROBE_ATTEMPTS if attempts is None else attempts + + for sequence in range(1, attempts + 1): + try: + probe = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, + socket.IPPROTO_ICMP) + except (PermissionError, OSError): + # A platform that requires root for this. Skipping is correct: the + # caller falls back to reporting UNKNOWN rather than guessing. + return False, None + + identifier = os.getpid() & 0xFFFF + body = struct.pack("!BBHHH", 8, 0, 0, identifier, sequence) + _ICMP_MAGIC + packet = (struct.pack("!BBHHH", 8, 0, _checksum(body), identifier, + sequence) + _ICMP_MAGIC) + deadline = time.monotonic() + timeout + try: + start = time.monotonic() + probe.sendto(packet, (host, 0)) + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + break + probe.settimeout(remaining) + try: + data, address = probe.recvfrom(2048) + except socket.timeout: + break + # Keep waiting on anything that is not our reply from our host, + # rather than accepting the first packet that arrives. + if address[0] == host and _is_echo_reply(data): + return True, (time.monotonic() - start) * 1000 + except OSError: + pass + finally: + probe.close() + + return False, None + + +def describe_icmp_result(name: str, host: str, answered: bool, + round_trip_ms: Optional[float]) -> str: + """One console line for an ICMP probe, in the repository's output style.""" + if answered: + return (f"{Fore.GREEN} - {name} ({host}) answered a ping in " + f"{round_trip_ms:.1f} ms{Style.RESET_ALL}") + return (f"{Fore.YELLOW} - {name} ({host}) did not answer a ping " + f"(some servers drop ICMP; this is not proof it is down)" + f"{Style.RESET_ALL}") + + +def get_module_tools(): + """Expose nothing from this module as a runnable tool. + + The registry uses this function when a module defines it and never falls + back to scanning for public functions, so declaring an empty mapping here + keeps these helpers out of the tool list without adding four more names to + the central exclusion set in core/tools_registry.py. + + That matters beyond tidiness: these are the signals the DNS tools use to + decide whether their own results can be trusted, not diagnostics in their + own right. Exposing them would invite a caller to run the probe on its own + and read an interception verdict as though it were a network check. + + Returns: + An empty dict. + """ + return {} diff --git a/network_tools/resolver_check.py b/network_tools/resolver_check.py index d7d0f92..43db7d3 100644 --- a/network_tools/resolver_check.py +++ b/network_tools/resolver_check.py @@ -6,12 +6,21 @@ resolvers, which is critical for diagnosing network connectivity issues. """ +import dns.exception +import dns.message +import dns.query import dns.resolver import time from datetime import datetime from typing import Dict, List, Tuple, Optional from colorama import Fore, Style +from config import DOT_PORT, DOT_QUERY_TIMEOUT +from .dns_interception import ( + REACHABLE, UNKNOWN, UNREACHABLE, + dns_interception_detected, interception_notice, +) + # List of DNS resolvers and their IP addresses DEFAULT_DNS_RESOLVERS = { "Google Public DNS - Primary": "8.8.8.8", @@ -26,6 +35,26 @@ "Comodo Secure DNS - Secondary": "8.20.247.20" } +# DoT hostnames, used for certificate validation. On a host where DNS is +# intercepted, port 53 is answered locally for every destination, so DoT on +# port 853 is the only way to reach the actual resolver -- see +# docs/dns_interception_srd.md section 5.4. +# +# A resolver absent from this table has no DoT service and therefore cannot be +# checked at all on an intercepted host. Measured 2026-08-29: the eight below +# all answered over DoT with a passing negative control, while both Comodo +# addresses time out on 853. +DOT_HOSTNAMES = { + "8.8.8.8": "dns.google", + "8.8.4.4": "dns.google", + "1.1.1.1": "cloudflare-dns.com", + "1.0.0.1": "cloudflare-dns.com", + "208.67.222.222": "dns.opendns.com", + "208.67.220.220": "dns.opendns.com", + "9.9.9.9": "dns.quad9.net", + "149.112.112.112": "dns.quad9.net", +} + def get_local_default_dns_resolver() -> str: """Get the IP address of the local default DNS resolver. @@ -75,16 +104,56 @@ def check_resolver(resolver_name: str, resolver_ip: str, query_domain: str = 'ex # If we get an answer, consider the resolver reachable if answer: return True, response_time, None - + except (dns.resolver.NoAnswer, dns.resolver.NXDOMAIN, dns.resolver.Timeout, dns.exception.DNSException) as e: if attempt == retry_attempts - 1: return False, None, str(e) time.sleep(2) # Sleep for 2 seconds before retrying - + # If we reach here, all attempts failed but didn't raise exceptions return False, None, "No valid DNS response received" +def check_resolver_over_tls(resolver_ip: str, query_domain: str = 'example.com', + timeout: int = None) -> Tuple[bool, Optional[float], Optional[str]]: + """Check a DNS resolver over DoT, which a DNS proxy extension cannot capture. + + On a host running a macOS DNS proxy network extension, a plain port-53 query + to any address is answered locally, so check_resolver above cannot fail and + its result means nothing. DoT runs on TCP 853, which is not intercepted, and + reaches the real resolver. + + Args: + resolver_ip: IP address of the resolver. + query_domain: Domain to query. + timeout: Seconds to wait for the response. + + Returns: + Tuple of (reachable, response_time_seconds, error_message). A resolver + with no entry in DOT_HOSTNAMES returns (False, None, reason) and the + caller reports it as UNKNOWN rather than unreachable -- not being + checkable is not the same as being down. + """ + timeout = DOT_QUERY_TIMEOUT if timeout is None else timeout + hostname = DOT_HOSTNAMES.get(resolver_ip) + if hostname is None: + return False, None, "no DoT service; cannot be checked on this host" + + query = dns.message.make_query(query_domain, 'A') + start_time = time.monotonic() + try: + response = dns.query.tls(query, resolver_ip, port=DOT_PORT, + timeout=timeout, server_hostname=hostname) + except dns.exception.Timeout: + return False, None, "timed out over DoT" + except (OSError, dns.exception.DNSException) as exc: + return False, None, f"DoT query failed: {exc}" + + if not response.answer: + return False, None, "DoT connection succeeded but returned no answer" + return True, time.monotonic() - start_time, None + + def monitor_dns_resolvers(custom_resolvers: Optional[Dict[str, str]] = None) -> str: """Monitor the reachability of DNS resolvers. @@ -97,10 +166,20 @@ def monitor_dns_resolvers(custom_resolvers: Optional[Dict[str, str]] = None) -> """ reachable_resolvers = [] unreachable_resolvers = [] + unknown_resolvers = [] results = "" results += f"Starting DNS Resolver monitoring report at: {datetime.now()}\n" results += "This will check the reachability of several of the most popular DNS resolvers.\n" - + + # If a DNS proxy is answering for every destination, a port-53 query proves + # nothing about the resolver it was addressed to. Switch to DoT, which is + # not captured, and report anything with no DoT service as UNKNOWN rather + # than inventing a verdict. See docs/dns_interception_srd.md section 5.4. + intercepted = dns_interception_detected() + if intercepted is not False: + results += f"{interception_notice()}\n" + results += "Falling back to DNS over TLS on port 853, which is not intercepted.\n" + # Start with default resolvers resolvers_to_check = DEFAULT_DNS_RESOLVERS.copy() @@ -117,8 +196,18 @@ def monitor_dns_resolvers(custom_resolvers: Optional[Dict[str, str]] = None) -> # Iterate through the list of DNS resolvers and check their reachability for resolver_name, resolver_ip in resolvers_to_check.items(): - is_reachable, response_time, error = check_resolver(resolver_name, resolver_ip) - + if intercepted is not False: + checkable = resolver_ip in DOT_HOSTNAMES + if not checkable: + reason = "no DoT service, and port 53 is intercepted here" + print(f"{Fore.YELLOW} - Cannot check {resolver_name} " + f"({resolver_ip}): {reason}{Style.RESET_ALL}") + unknown_resolvers.append(f"{resolver_name}: not checkable: {reason}") + continue + is_reachable, response_time, error = check_resolver_over_tls(resolver_ip) + else: + is_reachable, response_time, error = check_resolver(resolver_name, resolver_ip) + if is_reachable and response_time is not None: print(f"{Fore.GREEN} - Successfully queried {resolver_name} ({resolver_ip}): Response time {response_time:.3f} seconds{Style.RESET_ALL}") reachable_resolvers.append(f"{resolver_name}: Response Time: {response_time:.3f} seconds") @@ -132,13 +221,29 @@ def monitor_dns_resolvers(custom_resolvers: Optional[Dict[str, str]] = None) -> for resolver_info in reachable_resolvers: results += f"- {resolver_info}\n" - if not unreachable_resolvers: - results += "\nAll DNS resolvers are reachable.\n" - else: + if unreachable_resolvers: results += "\nUnreachable DNS Resolvers:\n" for resolver_info in unreachable_resolvers: results += f"- {resolver_info}\n" + # Stated separately and never folded into either of the others. A resolver + # nobody could check is not a resolver that answered, and it is not one that + # failed either. + if unknown_resolvers: + results += "\nDNS Resolvers That Could Not Be Checked:\n" + for resolver_info in unknown_resolvers: + results += f"- {resolver_info}\n" + + if not unreachable_resolvers and not unknown_resolvers: + results += "\nAll DNS resolvers are reachable.\n" + elif not unreachable_resolvers: + results += (f"\nAll {len(reachable_resolvers)} checkable DNS resolvers are " + f"reachable; {len(unknown_resolvers)} could not be checked.\n") + else: + results += (f"\nSummary: {len(reachable_resolvers)} reachable, " + f"{len(unreachable_resolvers)} unreachable, " + f"{len(unknown_resolvers)} not checkable.\n") + return results diff --git a/tests/test_dns_interception.py b/tests/test_dns_interception.py new file mode 100644 index 0000000..8599ba0 --- /dev/null +++ b/tests/test_dns_interception.py @@ -0,0 +1,187 @@ +#!/usr/bin/env python3 +""" +Test script for DNS interception detection and the checks that depend on it + +A macOS DNS proxy network extension captures DNS flows system-wide, so a query +addressed to any IP on port 53 is answered locally. Three tools reported success +unconditionally as a result. These tests cover the properties that keep the +fixed versions honest. + +Offline: every network call is stubbed. See docs/dns_interception_srd.md. +""" + +import os +import struct +import sys + +import dns.exception +from colorama import Fore, Style, init + +init(autoreset=True) + +parent_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if parent_dir not in sys.path: + sys.path.append(parent_dir) + +from network_tools import dns_check, dns_interception, resolver_check + +FAILURES = [] + + +def check(condition, label): + if condition: + print(f"{Fore.GREEN} [PASS] {label}{Style.RESET_ALL}") + else: + print(f"{Fore.RED} [FAIL] {label}{Style.RESET_ALL}") + FAILURES.append(label) + + +def _echo_reply(magic=dns_interception._ICMP_MAGIC, icmp_type=0): + return struct.pack("!BBHHH", icmp_type, 0, 0, 1234, 1) + magic + + +def test_interception_detection(): + """A response from a reserved, unroutable address can only be a proxy.""" + print("Detection:") + saved = dns_interception.dns.query.udp + + dns_interception._interception_cache = None + dns_interception.dns.query.udp = lambda *a, **k: "a response" + check(dns_interception.dns_interception_detected(force=True) is True, + "a response from the unroutable probe address means intercepted") + + dns_interception._interception_cache = None + def timeout(*args, **kwargs): + raise dns.exception.Timeout("no answer") + dns_interception.dns.query.udp = timeout + check(dns_interception.dns_interception_detected(force=True) is False, + "a timeout from the probe address means clean") + + dns_interception._interception_cache = None + def unreachable(*args, **kwargs): + raise OSError("network is down") + dns_interception.dns.query.udp = unreachable + check(dns_interception.dns_interception_detected(force=True) is None, + "a probe that could not run returns None, not False") + + dns_interception.dns.query.udp = saved + dns_interception._interception_cache = None + + +def test_unknown_is_never_an_all_clear(): + """The single property this whole change exists to guarantee.""" + print("No false all-clear:") + unknown = ["- A (198.41.0.4) - DNS not testable here; answers a ping in 12 ms"] + report = dns_check.generate_dns_report([], [], unknown) + check("All DNS Root Servers are reachable" not in report, + "a report with unknowns never claims all root servers reachable") + check("could not be determined" in report, + "it says the reachability could not be determined") + check("intercepted" in report, + "and it says why") + + mixed = dns_check.generate_dns_report(["- B (199.9.14.201)"], [], unknown) + check("All DNS Root Servers are reachable" not in mixed, + "a partly-unknown report does not claim an all-clear either") + + +def test_icmp_reply_validation(): + """A router's destination-unreachable is not an echo reply. + + The first version of this probe counted any received ICMP packet, so an + unreachable message for the unroutable test address read as a 2310 ms + success. The negative control caught it; this test keeps it caught. + """ + print("ICMP reply validation:") + check(dns_interception._is_echo_reply(_echo_reply()) is True, + "an echo reply carrying our payload counts") + check(dns_interception._is_echo_reply(_echo_reply(icmp_type=3)) is False, + "a destination-unreachable (type 3) does not count") + check(dns_interception._is_echo_reply(_echo_reply(icmp_type=11)) is False, + "a time-exceeded (type 11) does not count") + check(dns_interception._is_echo_reply(_echo_reply(magic=b"someone-else")) is False, + "an echo reply that is not ours does not count") + check(dns_interception._is_echo_reply(b"\x00\x00") is False, + "a truncated packet does not count") + check(dns_interception._is_echo_reply(b"") is False, + "an empty packet does not count") + check(dns_interception._is_echo_reply(b"\x00" * 20 + _echo_reply()) is True, + "an echo reply behind an IP header still counts") + + +def test_root_servers_report_unknown_when_intercepted(): + """The roots serve port 53 only and have no encrypted alternative, so on an + intercepted host the honest answer is that the check could not run.""" + print("Root servers under interception:") + saved = dns_interception.dns_interception_detected + dns_check.dns_interception_detected = lambda *a, **k: True + dns_check.icmp_reachable = lambda ip, **k: (True, 5.0) + try: + reachable, unreachable, unknown = dns_check.check_dns_root_servers() + check(reachable == [], "nothing is reported reachable") + check(unreachable == [], "nothing is reported unreachable either") + check(len(unknown) == len(dns_check.DNS_ROOT_SERVERS), + "every root server is reported as not checkable") + finally: + dns_check.dns_interception_detected = saved + dns_check.icmp_reachable = dns_interception.icmp_reachable + + +def test_a_root_that_drops_icmp_is_not_called_down(): + """G-root and L-root drop ICMP even when healthy.""" + print("ICMP non-response wording:") + line = dns_interception.describe_icmp_result("'G' root server", + "192.112.36.4", False, None) + check("not proof it is down" in line, "a non-reply is not reported as down") + check("did not answer a ping" in line, "it is reported as what it is") + + +def test_resolvers_without_dot_are_unknown(): + """Comodo offers no DoT, so on an intercepted host it cannot be checked at + all -- which is not the same as being unreachable.""" + print("Resolvers without DoT:") + for ip in ("8.26.56.26", "8.20.247.20"): + check(ip not in resolver_check.DOT_HOSTNAMES, + f"{ip} is correctly recorded as having no DoT service") + reachable, _, error = resolver_check.check_resolver_over_tls("8.26.56.26") + check(reachable is False and "cannot be checked" in (error or ""), + "a resolver with no DoT hostname reports why, not a failure to reach") + + for ip in ("8.8.8.8", "1.1.1.1", "9.9.9.9", "208.67.222.222"): + check(ip in resolver_check.DOT_HOSTNAMES, f"{ip} has a DoT hostname") + + +def test_internet_check_does_not_use_port_53(): + """The port was incidental and it is what made the check unable to fail.""" + print("Internet check port:") + import config + check(config.INTERNET_CHECK_PORT != 53, + "the internet check no longer uses port 53") + check(config.INTERNET_CHECK_PORT == 443, + "it uses 443, which is not intercepted") + check(len(config.INTERNET_CHECK_HOSTS) > 1, + "more than one host is tried before declaring the internet down") + + +def run_all(): + print(f"{Fore.CYAN}DNS interception tests{Style.RESET_ALL}\n") + test_interception_detection() + test_unknown_is_never_an_all_clear() + test_icmp_reply_validation() + test_root_servers_report_unknown_when_intercepted() + test_a_root_that_drops_icmp_is_not_called_down() + test_resolvers_without_dot_are_unknown() + test_internet_check_does_not_use_port_53() + + print() + if FAILURES: + print(f"{Fore.RED}{len(FAILURES)} check(s) failed:{Style.RESET_ALL}") + for failure in FAILURES: + print(f" - {failure}") + return 1 + print(f"{Fore.GREEN}All DNS interception checks passed{Style.RESET_ALL}") + return 0 + + +if __name__ == "__main__": + sys.exit(run_all())