timka123
36af1e9e83
A hosts file cannot express a wildcard, so '*.youtube.com' was only ever the
base domain plus the guessed labels. The patterns the server now hands over
as unexpandable are answered by a stub on 127.0.0.1, and the helper owns one
systemd-resolved drop-in that points just those zones at it:
* the stub is listening before the drop-in is written, and the drop-in is
removed before the stub stops, so a zone's queries never land on a dead
port;
* stub and route live as long as the app, not the interval: between
intervals they forward everything, instead of restarting resolved (and
dropping its cache) on every pomodoro;
* a crashed run leaves a world-readable drop-in the next start adopts, and
the stub comes back before anything else;
* masks with nothing routable after their last '*' ('*gram*') are reported
as unenforceable, not silently dropped.
The Blocker page now reports what is really in place — the hosts region, the
routed zones, and the masks this machine cannot do — rather than the whole
expansion. 39 desktop tests, including the ordering invariants.
189 lines
6.8 KiB
Python
189 lines
6.8 KiB
Python
"""The blocker's lifecycle: which mechanism comes up when, and in which order.
|
|
|
|
`BlockerManager` is the one place that knows both mechanisms exist, and the
|
|
invariants are all about *ordering*: the DNS stub has to be listening before the
|
|
drop-in points zones at it, and the drop-in has to be gone before the stub stops.
|
|
Both rules are cheap to state and easy to break in a refactor, so they are
|
|
pinned here with stand-ins that do nothing but write down what was asked of them.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from kadence_desktop.blocking.manager import BlockerManager
|
|
|
|
|
|
class RecordingStub:
|
|
"""Stands in for the DNS stub: a flag, a port, and a list of events."""
|
|
|
|
def __init__(self, events: list, *, start_ok: bool = True) -> None:
|
|
self.events = events
|
|
self.start_ok = start_ok
|
|
self.running = False
|
|
self.port = 5353
|
|
self.error = None if start_ok else "the local DNS stub could not bind 127.0.0.1"
|
|
self.rules: tuple[list[str], list[str], str, bool] | None = None
|
|
|
|
def start(self) -> bool:
|
|
self.events.append("stub.start")
|
|
if not self.start_ok:
|
|
return False
|
|
self.running = True
|
|
return True
|
|
|
|
def stop(self) -> None:
|
|
self.events.append("stub.stop")
|
|
self.running = False
|
|
|
|
def set_rules(self, patterns: list[str], allow: list[str], redirect_ip: str, *, active: bool) -> None:
|
|
self.events.append(("rules", tuple(patterns), active))
|
|
self.rules = (list(patterns), list(allow), redirect_ip, active)
|
|
|
|
|
|
class RecordingRoute:
|
|
"""Stands in for the resolved drop-in."""
|
|
|
|
def __init__(self, events: list, *, ok: bool = True, applied=None) -> None:
|
|
self.events = events
|
|
self.ok = ok
|
|
self.applied = applied
|
|
self.last_error = None if ok else "the authorisation prompt was dismissed"
|
|
|
|
def on(self, zones: list[str], port: int) -> bool:
|
|
self.events.append(("route.on", tuple(zones), port))
|
|
if not self.ok:
|
|
return False
|
|
self.applied = (tuple(zones), port)
|
|
return True
|
|
|
|
def off(self) -> bool:
|
|
self.events.append("route.off")
|
|
if not self.ok:
|
|
return False
|
|
self.applied = None
|
|
return True
|
|
|
|
|
|
def policy(*, active: bool = True, domains=("youtube.com",), unexpandable=(), allowed=()) -> dict:
|
|
return {
|
|
"active": active,
|
|
"enabled": True,
|
|
"domains": list(domains),
|
|
"apps": [],
|
|
"kill_apps": False,
|
|
"redirect_ip": "127.0.0.1",
|
|
"unexpandable": list(unexpandable),
|
|
"allowed": list(allowed),
|
|
}
|
|
|
|
|
|
def manager(events: list, *, stub: RecordingStub | None = None, route: RecordingRoute | None = None) -> BlockerManager:
|
|
# A helper path that leads nowhere: the hosts half then refuses every write
|
|
# without touching anything, which is exactly what these tests do not need.
|
|
return BlockerManager(
|
|
helper_path="/nonexistent/kadence-helper",
|
|
report=lambda *args: None,
|
|
stub=stub if stub is not None else RecordingStub(events),
|
|
route=route if route is not None else RecordingRoute(events),
|
|
)
|
|
|
|
|
|
def test_the_stub_comes_up_before_the_route_and_the_route_before_the_rules():
|
|
events: list = []
|
|
blocker = manager(events)
|
|
blocker.sync(policy(unexpandable=["*.youtube.com"]))
|
|
|
|
assert events.index("stub.start") < events.index(("route.on", ("youtube.com",), 5353))
|
|
assert events.index(("route.on", ("youtube.com",), 5353)) < events.index(
|
|
("rules", ("*.youtube.com",), True)
|
|
)
|
|
assert blocker.routed_zones == ["youtube.com"]
|
|
assert blocker.active
|
|
|
|
|
|
def test_the_route_stays_up_between_intervals():
|
|
# A restart of systemd-resolved drops the machine's DNS cache, so the route
|
|
# belongs to the app's lifetime, not to one interval: between intervals the
|
|
# stub keeps the port and simply forwards everything.
|
|
events: list = []
|
|
blocker = manager(events)
|
|
# The server keeps reporting `unexpandable` while nothing is blocking — it is
|
|
# a property of the rules, not of the interval — so the route survives.
|
|
blocker.sync(policy(unexpandable=["*.youtube.com"]))
|
|
events.clear()
|
|
|
|
blocker.sync(policy(active=False, unexpandable=["*.youtube.com"]))
|
|
|
|
assert "route.off" not in events
|
|
assert "stub.stop" not in events
|
|
assert ("rules", (), False) in events
|
|
assert not blocker.active
|
|
|
|
|
|
def test_shutdown_takes_the_dropin_down_before_the_stub():
|
|
events: list = []
|
|
blocker = manager(events)
|
|
blocker.sync(policy(unexpandable=["*.youtube.com"]))
|
|
events.clear()
|
|
|
|
blocker.shutdown()
|
|
|
|
assert events.index("route.off") < events.index("stub.stop")
|
|
assert ("rules", (), False) in events
|
|
assert not blocker.active
|
|
|
|
|
|
def test_a_mask_without_a_zone_never_touches_the_route():
|
|
# `*gram*` has nothing after its last `*`: no zone to route, so the route is
|
|
# not opened just to say nothing.
|
|
events: list = []
|
|
blocker = manager(events)
|
|
blocker.sync(policy(domains=("instagram.com",), unexpandable=["*gram*"]))
|
|
|
|
assert blocker.unroutable == ["*gram*"]
|
|
assert not [event for event in events if isinstance(event, tuple) and event[0] == "route.on"]
|
|
assert blocker.routed_zones == []
|
|
|
|
|
|
def test_switching_hosts_file_enforcement_off_takes_the_route_with_it():
|
|
events: list = []
|
|
blocker = manager(events)
|
|
blocker.sync(policy(unexpandable=["*.youtube.com"]))
|
|
events.clear()
|
|
|
|
blocker.enforce_hosts = False
|
|
blocker.sync(policy(unexpandable=["*.youtube.com"]))
|
|
|
|
assert events.index("route.off") < events.index("stub.stop")
|
|
assert blocker.routed_zones == []
|
|
|
|
|
|
def test_a_stub_that_cannot_bind_leaves_no_route_behind():
|
|
# The invariant, upside down: if nothing is listening, the drop-in must not
|
|
# be written — that is the state where the zones' DNS simply dies.
|
|
events: list = []
|
|
blocker = manager(events, stub=RecordingStub(events, start_ok=False))
|
|
blocker.sync(policy(unexpandable=["*.youtube.com"]))
|
|
|
|
assert not [event for event in events if isinstance(event, tuple) and event[0] == "route.on"]
|
|
assert "route.off" not in events
|
|
assert blocker.route_error and "could not bind" in blocker.route_error
|
|
|
|
|
|
def test_a_dropin_left_by_a_crash_gets_its_stub_back_at_startup():
|
|
events: list = []
|
|
route = RecordingRoute(events, applied=(("youtube.com",), 5353))
|
|
manager(events, route=route) # construction alone must do it
|
|
|
|
assert events == ["stub.start"]
|
|
|
|
|
|
def test_a_refused_authorisation_is_reported_alongside_the_hosts_error():
|
|
events: list = []
|
|
blocker = manager(events, route=RecordingRoute(events, ok=False))
|
|
blocker.sync(policy(unexpandable=["*.youtube.com"]))
|
|
|
|
assert blocker.route_error == "the authorisation prompt was dismissed"
|
|
# The hosts half failed too (there is no helper), and it speaks first.
|
|
assert blocker.error == blocker.last_error
|
|
assert blocker.error is not None
|