91 lines
2.8 KiB
Python
91 lines
2.8 KiB
Python
"""DNS resolution that always uses an explicitly configured server.
|
|
|
|
The system/router-provided resolver (as configured via /etc/resolv.conf,
|
|
usually pointing at your router) is never consulted. Each lookup opens a
|
|
fresh, unconfigured dnspython resolver and points it only at the DNS
|
|
server given for that specific endpoint.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import ipaddress
|
|
import socket
|
|
import threading
|
|
|
|
import dns.resolver
|
|
|
|
# Guards socket.getaddrinfo patching below. Checks must run sequentially
|
|
# while a patch is active since getaddrinfo is a process-global function;
|
|
# the monitor's main loop does exactly that (one endpoint at a time).
|
|
_patch_lock = threading.Lock()
|
|
|
|
|
|
def is_ip_literal(host: str) -> bool:
|
|
"""Return True if `host` is already a literal IPv4 or IPv6 address.
|
|
|
|
Such hosts have nothing to resolve — dnspython does not special-case
|
|
them, it will happily send a DNS query for e.g. "192.168.0.2" as if it
|
|
were a hostname and get NXDOMAIN back. Callers should skip resolution
|
|
entirely for these.
|
|
"""
|
|
try:
|
|
ipaddress.ip_address(host)
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def format_host_for_url(host: str) -> str:
|
|
"""Format `host` for embedding directly in a URL.
|
|
|
|
IPv6 literals need brackets (e.g. "::1" -> "[::1]"); everything else
|
|
(hostnames, IPv4 literals) is returned unchanged.
|
|
"""
|
|
try:
|
|
if ipaddress.ip_address(host).version == 6:
|
|
return f"[{host}]"
|
|
except ValueError:
|
|
pass
|
|
return host
|
|
|
|
|
|
def resolve(host: str, dns_server: str, timeout: float) -> str:
|
|
"""Resolve `host` to an IPv4 address using *only* `dns_server`.
|
|
|
|
Raises dns.exception.DNSException (or a subclass) on failure —
|
|
NXDOMAIN, timeout, no answer, etc.
|
|
"""
|
|
resolver = dns.resolver.Resolver(configure=False)
|
|
resolver.nameservers = [dns_server]
|
|
resolver.lifetime = timeout
|
|
resolver.timeout = timeout
|
|
answer = resolver.resolve(host, "A")
|
|
return str(answer[0])
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def pin_resolution(hostname: str, ip: str):
|
|
"""Force socket.getaddrinfo(hostname, ...) to return `ip` for the
|
|
duration of the block.
|
|
|
|
This lets `requests`/urllib3 connect to the IP we already resolved via
|
|
the explicit DNS server above, without doing any resolution of its
|
|
own (which would otherwise go through the system resolver). The
|
|
hostname in the URL is left untouched, so the Host header and TLS
|
|
SNI/certificate verification behave exactly as normal.
|
|
"""
|
|
original_getaddrinfo = socket.getaddrinfo
|
|
|
|
def patched(host, *args, **kwargs):
|
|
if host == hostname:
|
|
host = ip
|
|
return original_getaddrinfo(host, *args, **kwargs)
|
|
|
|
with _patch_lock:
|
|
socket.getaddrinfo = patched
|
|
try:
|
|
yield
|
|
finally:
|
|
socket.getaddrinfo = original_getaddrinfo
|