84 lines
2.5 KiB
Python
84 lines
2.5 KiB
Python
"""IPv4 mDNS resolver using only Python stdlib. Used as no-dep fallback."""
|
|
import socket
|
|
import struct
|
|
import select
|
|
import time
|
|
|
|
MDNS_ADDR4 = "224.0.0.251"
|
|
MDNS_PORT = 5353
|
|
QTYPE_A = 1
|
|
|
|
|
|
def build_query(hostname: str, qtype: int = QTYPE_A) -> bytes:
|
|
header = struct.pack(">HHHHHH", 0, 0, 1, 0, 0, 0)
|
|
q = b""
|
|
for part in hostname.rstrip(".").split("."):
|
|
q += bytes([len(part)]) + part.encode()
|
|
q += b"\x00" + struct.pack(">HH", qtype, 1)
|
|
return header + q
|
|
|
|
|
|
def skip_name(data: bytes, off: int) -> int:
|
|
while off < len(data):
|
|
b = data[off]
|
|
if b == 0:
|
|
return off + 1
|
|
if b & 0xC0 == 0xC0:
|
|
return off + 2
|
|
off += b + 1
|
|
return off
|
|
|
|
|
|
def parse_all_a(data: bytes) -> list[tuple[str, int]]:
|
|
"""Return every A record (ip, ttl) in the response."""
|
|
out: list[tuple[str, int]] = []
|
|
if len(data) < 12:
|
|
return out
|
|
qd, an = struct.unpack(">HH", data[4:8])
|
|
off = 12
|
|
for _ in range(qd):
|
|
off = skip_name(data, off) + 4
|
|
for _ in range(an):
|
|
off = skip_name(data, off)
|
|
if off + 10 > len(data):
|
|
break
|
|
rtype, _, ttl, rdlen = struct.unpack(">HHIH", data[off:off + 10])
|
|
off += 10
|
|
if rtype == QTYPE_A and rdlen == 4 and off + 4 <= len(data):
|
|
out.append((socket.inet_ntoa(data[off:off + 4]), ttl))
|
|
off += rdlen
|
|
return out
|
|
|
|
|
|
def query_all(hostname: str, timeout: float) -> list[tuple[str, int]]:
|
|
"""Collect all A records advertised for hostname during the time window."""
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
try:
|
|
mreq = socket.inet_aton(MDNS_ADDR4) + socket.inet_aton("0.0.0.0")
|
|
s.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
|
|
except OSError:
|
|
pass
|
|
s.setblocking(False)
|
|
try:
|
|
s.sendto(build_query(hostname, QTYPE_A), (MDNS_ADDR4, MDNS_PORT))
|
|
except OSError:
|
|
s.close()
|
|
return []
|
|
found: list[tuple[str, int]] = []
|
|
end = time.time() + timeout
|
|
while time.time() < end:
|
|
r, _, _ = select.select([s], [], [], min(0.1, max(0.0, end - time.time())))
|
|
if r:
|
|
try:
|
|
data, _ = s.recvfrom(4096)
|
|
except OSError:
|
|
continue
|
|
recs = parse_all_a(data)
|
|
if recs:
|
|
found.extend(recs)
|
|
# Give a brief grace window for additional A packets, then stop.
|
|
end = min(end, time.time() + 0.15)
|
|
s.close()
|
|
return found
|