timka123andClaude Opus 5
ec7eedbc66
`[tool.ruff]` set a line length and a target version and nothing else, so which rules ran depended on the installed ruff's defaults — fine while nobody ran it, not fine for a CI job that can turn a push red because a tool changed its mind. The set is named explicitly (E, F, I, UP, SIM, B, RUF) with three ignores that are each a real decision rather than a shrug: B008, because FastAPI's whole dependency idiom is a call in a default argument; E501, because wrapping prose is a formatter's business; and RUF001, because the reports and the bot speak Russian and "ambiguous character" is the wrong complaint about a string the user is meant to read. The 73 findings that remained are fixed — mostly `datetime.timezone.utc` → `datetime.UTC`, `typing.Callable` → `collections.abc.Callable`, import order, and unused `noqa` markers — plus seven by hand: `contextlib.suppress` for two try/except/pass pairs, unpacking instead of list concatenation, one negated return. `ruff format` is deliberately *not* adopted: it would reformat 51 files, and some of the layout here is hand-set (the comment tables, the DNS wire-format structs) in ways a formatter flattens. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
377 lines
14 KiB
Python
377 lines
14 KiB
Python
"""The DNS side of website blocking: routing, the stub, and the drop-in file.
|
|
|
|
Everything here runs without root and without touching the real machine: the
|
|
stub binds a port the kernel picks, "upstream" is a socket in this process, the
|
|
resolved drop-in goes to a temp directory, and the helper is a shell script that
|
|
only writes down what it was called with. The one thing worth testing obsessively
|
|
is the order of operations — the failure mode is a machine whose DNS points at a
|
|
stub that is not listening, and that is not a thing to discover in production.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.machinery
|
|
import importlib.util
|
|
import socket
|
|
import struct
|
|
import threading
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from kadence_desktop.blocking.resolver import (
|
|
ResolvedRoute,
|
|
StubResolver,
|
|
_parse_resolv_conf,
|
|
split_routable,
|
|
)
|
|
|
|
HELPER_SOURCE = Path(__file__).resolve().parents[1] / "helper" / "kadence-helper"
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def helper():
|
|
"""The privileged helper, imported as a module — it is all functions.
|
|
|
|
It has no `.py` suffix, so the loader has to be named explicitly.
|
|
"""
|
|
loader = importlib.machinery.SourceFileLoader("kadence_helper", str(HELPER_SOURCE))
|
|
spec = importlib.util.spec_from_loader("kadence_helper", loader)
|
|
module = importlib.util.module_from_spec(spec)
|
|
assert spec.loader is not None
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
# ==========================================================================
|
|
# Which masks can be routed
|
|
# ==========================================================================
|
|
|
|
|
|
def test_a_mask_anchored_on_a_zone_is_routable():
|
|
routable, unroutable, zones = split_routable(
|
|
["*.youtube.com", "*gram.com", "*.t.me", "*.site", "*gram*", "*"]
|
|
)
|
|
assert routable == ["*.youtube.com", "*gram.com", "*.t.me"]
|
|
assert unroutable == ["*.site", "*gram*", "*"]
|
|
assert zones == ["youtube.com", "gram.com", "t.me"]
|
|
|
|
|
|
def test_zones_are_reported_once_even_for_several_masks():
|
|
_, _, zones = split_routable(["*.youtube.com", "*.music.youtube.com", "*.video.youtube.com"])
|
|
assert zones == ["youtube.com", "music.youtube.com", "video.youtube.com"]
|
|
|
|
|
|
# ==========================================================================
|
|
# The DNS wire format, from the test's side
|
|
# ==========================================================================
|
|
|
|
TYPE_A, TYPE_AAAA, TYPE_HTTPS = 1, 28, 65
|
|
|
|
|
|
def query(name: str, qtype: int = TYPE_A, qid: int = 0x1234) -> bytes:
|
|
labels = b"".join(bytes([len(label)]) + label.encode() for label in name.split(".")) + b"\x00"
|
|
return struct.pack("!HHHHHH", qid, 0x0100, 1, 0, 0, 0) + labels + struct.pack("!HH", qtype, 1)
|
|
|
|
|
|
def _question_end(packet: bytes) -> int | None:
|
|
"""Where the question section ends, or None for anything malformed."""
|
|
offset = 12
|
|
while offset < len(packet) and packet[offset]:
|
|
offset += 1 + packet[offset]
|
|
offset += 1 + 4
|
|
return offset if offset <= len(packet) else None
|
|
|
|
|
|
def _answer(rdata: bytes, record_type: int = TYPE_A) -> bytes:
|
|
return b"\xc0\x0c" + struct.pack("!HHIH", record_type, 1, 60, len(rdata)) + rdata
|
|
|
|
|
|
def upstream_answer(request: bytes, address: str) -> bytes | None:
|
|
"""What a real upstream would say: an A record with a routable address."""
|
|
end = _question_end(request)
|
|
if end is None:
|
|
return None
|
|
rdata = socket.inet_aton(address)
|
|
flags = struct.unpack_from("!H", request, 2)[0] | 0x8000
|
|
return struct.pack("!HHHHHH", 0x1234, flags, 1, 1, 0, 0) + request[12:end] + _answer(rdata)
|
|
|
|
|
|
def read_answer(response: bytes) -> list[tuple[int, bytes]]:
|
|
"""The answer records of a response, as `(type, rdata)` pairs."""
|
|
count = struct.unpack_from("!H", response, 6)[0]
|
|
offset = _question_end(response)
|
|
assert offset is not None
|
|
records: list[tuple[int, bytes]] = []
|
|
for _ in range(count):
|
|
offset += 2 # the name, always a pointer to the question here
|
|
record_type, _, _, length = struct.unpack_from("!HHIH", response, offset)
|
|
offset += 10
|
|
records.append((record_type, response[offset : offset + length]))
|
|
offset += length
|
|
return records
|
|
|
|
|
|
def read_ip(response: bytes) -> str | None:
|
|
for record_type, rdata in read_answer(response):
|
|
if record_type == TYPE_A:
|
|
return socket.inet_ntoa(rdata)
|
|
return None
|
|
|
|
|
|
class Upstream:
|
|
"""A stand-in for the machine's real DNS server, answering one address."""
|
|
|
|
def __init__(self, address: str = "203.0.113.9") -> None:
|
|
self.answer = address
|
|
self.queries: list[bytes] = []
|
|
self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
self.socket.bind(("127.0.0.1", 0))
|
|
self.address = self.socket.getsockname()
|
|
self.thread = threading.Thread(target=self._serve, daemon=True)
|
|
self.thread.start()
|
|
|
|
def _serve(self) -> None:
|
|
while True:
|
|
try:
|
|
request, peer = self.socket.recvfrom(4096)
|
|
except OSError:
|
|
return
|
|
self.queries.append(request)
|
|
response = upstream_answer(request, self.answer)
|
|
if response is not None:
|
|
self.socket.sendto(response, peer)
|
|
|
|
def stop(self) -> None:
|
|
self.socket.close()
|
|
|
|
|
|
# ==========================================================================
|
|
# The stub
|
|
# ==========================================================================
|
|
|
|
|
|
@pytest.fixture
|
|
def stub(tmp_path):
|
|
"""A stub on a kernel-chosen port, forwarding to [Upstream] by default."""
|
|
upstream = Upstream()
|
|
conf = tmp_path / "resolv.conf"
|
|
conf.write_text(f"nameserver {upstream.address[0]}:{upstream.address[1]}\n")
|
|
resolver = StubResolver(ports=(0,), upstream_path=str(conf))
|
|
assert resolver.start()
|
|
resolver.set_rules(
|
|
["*.youtube.com"], ["music.youtube.com"], "127.0.0.1", active=True
|
|
)
|
|
yield resolver
|
|
resolver.stop()
|
|
upstream.stop()
|
|
|
|
|
|
def ask(stub: StubResolver, packet: bytes, *, tcp: bool = False) -> bytes:
|
|
if tcp:
|
|
with socket.create_connection(("127.0.0.1", stub.port), timeout=2) as sock:
|
|
sock.sendall(struct.pack("!H", len(packet)) + packet)
|
|
length = struct.unpack("!H", sock.recv(2))[0]
|
|
return sock.recv(length)
|
|
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
|
|
sock.settimeout(2)
|
|
sock.sendto(packet, ("127.0.0.1", stub.port))
|
|
response, _ = sock.recvfrom(4096)
|
|
return response
|
|
|
|
|
|
def test_a_blocked_name_is_answered_with_the_redirect_address(stub):
|
|
response = ask(stub, query("web.youtube.com"))
|
|
assert read_ip(response) == "127.0.0.1"
|
|
|
|
|
|
def test_ipv6_and_service_records_get_no_answer_at_all(stub):
|
|
# The address hints in an HTTPS record would hand the browser a real address
|
|
# without it ever asking the A record, so every non-A query is NODATA.
|
|
for qtype in (TYPE_AAAA, TYPE_HTTPS):
|
|
response = ask(stub, query("web.youtube.com", qtype=qtype))
|
|
assert struct.unpack_from("!H", response, 6)[0] == 0
|
|
|
|
|
|
def test_the_bare_zone_is_not_matched_by_a_wildcard(stub):
|
|
# `*.youtube.com` never covered `youtube.com` itself; the hosts file has that
|
|
# one, and the stub must not start disagreeing with it.
|
|
response = ask(stub, query("youtube.com"))
|
|
assert read_ip(response) == "203.0.113.9"
|
|
|
|
|
|
def test_an_allow_listed_subdomain_is_forwarded(stub):
|
|
response = ask(stub, query("music.youtube.com"))
|
|
assert read_ip(response) == "203.0.113.9"
|
|
|
|
|
|
def test_an_unrelated_name_is_forwarded(stub):
|
|
response = ask(stub, query("example.com"))
|
|
assert read_ip(response) == "203.0.113.9"
|
|
|
|
|
|
def test_when_the_block_is_off_everything_is_forwarded(stub):
|
|
stub.set_rules(["*.youtube.com"], [], "127.0.0.1", active=False)
|
|
assert read_ip(ask(stub, query("web.youtube.com"))) == "203.0.113.9"
|
|
|
|
|
|
def test_the_same_answer_comes_over_tcp(stub):
|
|
response = ask(stub, query("web.youtube.com"), tcp=True)
|
|
assert read_ip(response) == "127.0.0.1"
|
|
|
|
|
|
def test_junk_on_the_socket_does_not_take_the_stub_down(stub):
|
|
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
|
|
sock.sendto(b"\x01\x02 not a dns packet", ("127.0.0.1", stub.port))
|
|
sock.sendto(b"", ("127.0.0.1", stub.port))
|
|
sock.settimeout(0.2)
|
|
with pytest.raises(socket.timeout):
|
|
sock.recvfrom(4096)
|
|
# …and it is still answering the real questions.
|
|
assert read_ip(ask(stub, query("web.youtube.com"))) == "127.0.0.1"
|
|
|
|
|
|
def test_a_dead_upstream_becomes_servfail(tmp_path):
|
|
resolver = StubResolver(ports=(0,), upstream_path=str(tmp_path / "empty.conf"), timeout=0.3)
|
|
assert resolver.start()
|
|
try:
|
|
# The file does not exist and /etc/resolv.conf points at resolved, which
|
|
# is filtered out — so there is nowhere to forward to.
|
|
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
|
|
sock.settimeout(2)
|
|
sock.sendto(query("example.com"), ("127.0.0.1", resolver.port))
|
|
response, _ = sock.recvfrom(4096)
|
|
assert response[3] & 0x0F == 2 # SERVFAIL, not a lie about the address
|
|
finally:
|
|
resolver.stop()
|
|
|
|
|
|
def test_upstreams_are_read_from_resolv_conf_and_loops_are_dropped(tmp_path):
|
|
conf = tmp_path / "resolv.conf"
|
|
conf.write_text(
|
|
"# comment\n"
|
|
"search .\n"
|
|
"nameserver 127.0.0.53\n"
|
|
"nameserver 127.0.0.54\n"
|
|
"nameserver 192.168.3.1\n"
|
|
"nameserver 198.18.0.2:5353\n"
|
|
"nameserver [fe80::1]:53\n"
|
|
"nameserver not an address\n"
|
|
)
|
|
assert _parse_resolv_conf(str(conf)) == [
|
|
("127.0.0.53", 53),
|
|
("127.0.0.54", 53),
|
|
("192.168.3.1", 53),
|
|
("198.18.0.2", 5353),
|
|
("fe80::1", 53),
|
|
]
|
|
|
|
resolver = StubResolver(ports=(0,), upstream_path=str(conf))
|
|
resolver.set_rules([], [], "127.0.0.1", active=False)
|
|
servers = resolver._upstreams()
|
|
# resolved's own listeners are what would bounce a query straight back.
|
|
assert ("127.0.0.53", 53) not in servers
|
|
assert ("127.0.0.54", 53) not in servers
|
|
assert ("192.168.3.1", 53) in servers
|
|
|
|
|
|
# ==========================================================================
|
|
# The resolved route
|
|
# ==========================================================================
|
|
|
|
|
|
class FakeHelper:
|
|
"""Writes down its argv instead of asking polkit for anything."""
|
|
|
|
def __init__(self, path: Path) -> None:
|
|
self.log = path / "helper.log"
|
|
self.path = path / "kadence-helper"
|
|
self.path.write_text(f'#!/bin/sh\necho "$@" >> "{self.log}"\nexit 0\n')
|
|
self.path.chmod(0o755)
|
|
|
|
def calls(self) -> list[str]:
|
|
return self.log.read_text().splitlines() if self.log.exists() else []
|
|
|
|
|
|
class LocalRoute(ResolvedRoute):
|
|
"""The route, pointed straight at the fake helper.
|
|
|
|
How to become root is the environment's business (pkexec, sudo, whatever a
|
|
developer's machine offers); what these tests are about is *what gets asked*,
|
|
so the elevation layer is skipped rather than simulated.
|
|
"""
|
|
|
|
def _runner(self) -> list[str]:
|
|
return [self.helper_path]
|
|
|
|
|
|
def test_routing_zones_names_the_stub_and_the_helper(helper, tmp_path):
|
|
fake = FakeHelper(tmp_path)
|
|
route = LocalRoute(helper_path=str(fake.path), dropin_path=str(tmp_path / "kadence.conf"))
|
|
|
|
assert route.on(["T.Me", "youtube.com", "youtube.com"], 5353)
|
|
assert fake.calls() == ["resolved-on --port 5353 --domains t.me,youtube.com"]
|
|
|
|
# Same zones, same port: resolved is not restarted for nothing.
|
|
assert route.on(["youtube.com", "t.me"], 5353)
|
|
assert len(fake.calls()) == 1
|
|
|
|
# A different port means the drop-in has grown stale, so it goes again.
|
|
assert route.on(["youtube.com", "t.me"], 5453)
|
|
assert fake.calls()[1] == "resolved-on --port 5453 --domains t.me,youtube.com"
|
|
|
|
|
|
def test_stopping_the_route_is_a_no_op_when_nothing_was_routed(tmp_path):
|
|
fake = FakeHelper(tmp_path)
|
|
route = ResolvedRoute(helper_path=str(fake.path), dropin_path=str(tmp_path / "absent.conf"))
|
|
assert route.off()
|
|
assert fake.calls() == []
|
|
|
|
|
|
def test_the_dropin_written_by_the_helper_is_read_back(helper, tmp_path):
|
|
# The two halves have to agree on the file's shape, or every app start would
|
|
# restart resolved "because nothing changed".
|
|
dropin = tmp_path / "kadence.conf"
|
|
dropin.write_text(helper.resolved_dropin_text(5353, ["youtube.com", "t.me"]))
|
|
fake = FakeHelper(tmp_path)
|
|
route = LocalRoute(helper_path=str(fake.path), dropin_path=str(dropin))
|
|
|
|
assert route.applied == (("t.me", "youtube.com"), 5353)
|
|
assert route.on(["t.me", "youtube.com"], 5353)
|
|
assert fake.calls() == []
|
|
|
|
assert route.off()
|
|
assert fake.calls() == ["resolved-off"]
|
|
assert route.applied is None
|
|
|
|
|
|
def test_a_missing_helper_is_reported_not_ignored(tmp_path):
|
|
route = ResolvedRoute(helper_path=str(tmp_path / "gone"), dropin_path=str(tmp_path / "kadence.conf"))
|
|
assert not route.on(["youtube.com"], 5353)
|
|
assert "not installed" in (route.last_error or "")
|
|
|
|
|
|
# ==========================================================================
|
|
# The helper's own validation — the part that runs as root
|
|
# ==========================================================================
|
|
|
|
|
|
def test_zone_names_are_validated_before_they_go_into_the_file(helper):
|
|
assert helper.load_zones("youtube.com, t.me ,YouTube.com") == ["youtube.com", "t.me"]
|
|
for bad in ["", " ", "localhost", "com", "youtube.com\nDNS=6.6.6.6", "127.0.0.1", "*.youtube.com"]:
|
|
with pytest.raises(SystemExit):
|
|
helper.load_zones(bad)
|
|
|
|
|
|
def test_the_dropin_says_exactly_what_it_means(helper):
|
|
text = helper.resolved_dropin_text(5353, ["youtube.com"])
|
|
assert "[Resolve]" in text
|
|
assert "DNS=127.0.0.1:5353" in text
|
|
assert "Domains=~youtube.com" in text
|
|
assert "systemd-resolved" in text
|
|
|
|
|
|
def test_a_priviliged_port_is_refused(helper):
|
|
for port in (53, 80, 1023, 70000):
|
|
with pytest.raises(SystemExit):
|
|
helper.command_resolved_on(port, "youtube.com")
|