61 lines
2.0 KiB
Python
61 lines
2.0 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 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 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
|