import random import socket import struct import time RCODES = {0: "NOERROR", 1: "FORMERR", 2: "SERVFAIL", 3: "NXDOMAIN", 4: "NOTIMP", 5: "REFUSED"} RESOLV_CONF = "/etc/resolv.conf" HOSTS_FILE = "/etc/hosts" def _ms(started): return round((time.monotonic() - started) * 1000, 1) def read_resolv_conf(path=RESOLV_CONF): nameservers, search, options = [], [], [] try: with open(path) as fh: for line in fh: parts = line.split("#", 1)[0].split() if not parts: continue if parts[0] == "nameserver" and len(parts) > 1: nameservers.append(parts[1]) elif parts[0] in ("search", "domain"): search.extend(parts[1:]) elif parts[0] == "options": options.extend(parts[1:]) except OSError as exc: return {"error": str(exc), "nameservers": [], "search": [], "options": []} return {"nameservers": nameservers, "search": search, "options": options} def hosts_entries(name, path=HOSTS_FILE): matches = [] try: with open(path) as fh: for line in fh: parts = line.split("#", 1)[0].split() if len(parts) > 1 and name in parts[1:]: matches.append(parts[0]) except OSError: pass return matches def _skip_name(data, offset): while True: length = data[offset] if length == 0: return offset + 1 if length & 0xC0 == 0xC0: return offset + 2 offset += 1 + length def _read_name(data, offset): labels = [] while True: length = data[offset] if length == 0: return ".".join(labels) if length & 0xC0 == 0xC0: pointer = struct.unpack(">H", data[offset:offset + 2])[0] & 0x3FFF return ".".join(labels + [_read_name(data, pointer)]) offset += 1 labels.append(data[offset:offset + length].decode(errors="replace")) offset += length def build_query(name, ident, qtype=1): header = struct.pack(">HHHHHH", ident, 0x0100, 1, 0, 0, 0) question = b"".join(bytes([len(label)]) + label.encode() for label in name.rstrip(".").split(".")) return header + question + b"\x00" + struct.pack(">HH", qtype, 1) def parse_response(data, ident): if len(data) < 12: raise ValueError("response too short") rid, flags, qdcount, ancount, _, _ = struct.unpack(">HHHHHH", data[:12]) if rid != ident: raise ValueError("response id mismatch") offset = 12 for _ in range(qdcount): offset = _skip_name(data, offset) + 4 addresses, cnames = [], [] for _ in range(ancount): offset = _skip_name(data, offset) rtype, _, _, rdlength = struct.unpack(">HHIH", data[offset:offset + 10]) offset += 10 if rtype == 1 and rdlength == 4: addresses.append(socket.inet_ntoa(data[offset:offset + 4])) elif rtype == 28 and rdlength == 16: addresses.append(socket.inet_ntop(socket.AF_INET6, data[offset:offset + 16])) elif rtype == 5: cnames.append(_read_name(data, offset)) offset += rdlength rcode = flags & 0xF return {"rcode": RCODES.get(rcode, str(rcode)), "addresses": addresses, "cnames": cnames} def dns_query(server, name, timeout=2.0, qtype=1, port=53): ident = random.randint(0, 0xFFFF) family = socket.AF_INET6 if ":" in server else socket.AF_INET started = time.monotonic() result = {"server": server, "type": "AAAA" if qtype == 28 else "A"} try: with socket.socket(family, socket.SOCK_DGRAM) as sock: sock.settimeout(timeout) sock.sendto(build_query(name, ident, qtype), (server, port)) data, _ = sock.recvfrom(4096) result.update(parse_response(data, ident)) except (OSError, ValueError) as exc: result["error"] = str(exc) or type(exc).__name__ result["ms"] = _ms(started) return result def resolve(host, port): started = time.monotonic() try: infos = socket.getaddrinfo(host, port, proto=socket.IPPROTO_TCP) addresses = sorted({info[4][0] for info in infos}) return {"addresses": addresses, "ms": _ms(started)} except OSError as exc: return {"addresses": [], "error": str(exc), "ms": _ms(started)} def tcp_connect(address, port, timeout): started = time.monotonic() try: with socket.create_connection((address, port), timeout=timeout): return {"address": address, "ok": True, "ms": _ms(started)} except OSError as exc: return {"address": address, "ok": False, "error": str(exc) or type(exc).__name__, "ms": _ms(started)} def probe(host, port, timeout=3.0): started = time.monotonic() resolv = read_resolv_conf() result = { "ts": time.time(), "host": host, "port": port, "resolv_conf": resolv, "hosts_file": hosts_entries(host), "getaddrinfo": resolve(host, port), "nameservers": [dns_query(server, host, timeout=min(timeout, 2.0)) for server in resolv["nameservers"]], "tcp": [], } candidates = list(result["getaddrinfo"]["addresses"]) for answer in result["nameservers"]: for address in answer.get("addresses", []): if address not in candidates: candidates.append(address) result["tcp"] = [tcp_connect(address, port, timeout) for address in candidates] result["ms"] = _ms(started) result["summary"] = summarize(result) return result def summarize(result): lines = [] gai = result["getaddrinfo"] if gai.get("error"): lines.append(f"System resolver failed for {result['host']}: {gai['error']}") else: lines.append(f"System resolver: {result['host']} -> {', '.join(gai['addresses'])}") if result["hosts_file"]: lines.append(f"/etc/hosts maps it to {', '.join(result['hosts_file'])}") if not result["resolv_conf"]["nameservers"]: lines.append("No nameservers in /etc/resolv.conf") for answer in result["nameservers"]: if answer.get("error"): lines.append(f"Nameserver {answer['server']}: no reply ({answer['error']})") elif answer["addresses"]: lines.append(f"Nameserver {answer['server']}: {answer['rcode']} -> {', '.join(answer['addresses'])}") else: lines.append(f"Nameserver {answer['server']}: {answer['rcode']}, no address") answers = [set(a.get("addresses", [])) for a in result["nameservers"] if not a.get("error")] if len(answers) > 1 and any(a != answers[0] for a in answers[1:]): lines.append("Nameservers disagree; the result depends on which one the resolver asks first") if not result["tcp"]: lines.append("No address to try a TCP connection to") for attempt in result["tcp"]: if attempt["ok"]: lines.append(f"TCP {attempt['address']}:{result['port']} connected in {attempt['ms']} ms") else: lines.append(f"TCP {attempt['address']}:{result['port']} failed: {attempt['error']}") return lines