"""Turning the server's policy into what this machine actually does. Called on every poll with the policy the backend computed. It keeps the hosts file, the DNS routes and the process list in step with it, reports every turned-away attempt back to the API (that is what makes the daily report fun), and — importantly — puts everything back the moment the policy says blocking is off, including when the app exits or crashes mid-interval. Two mechanisms do the website blocking, because one cannot do it all: * `/etc/hosts` answers the concrete hostnames the server expanded — exact, cheap, and readable with `cat`; * the patterns a hosts file cannot express (`*.youtube.com`, `*gram.com`) are answered by the local DNS stub, with `systemd-resolved` sent the affected zones through the helper's drop-in. The stub and the drop-in are kept for as long as the app runs, not per interval: repointing resolved costs a restart and a dropped DNS cache, and turning the block on and off is instantaneous — the stub answers or forwards, per query. That is why `release()` does not touch either of them, and why `shutdown()` is its own method: the drop-in has to be gone *before* the stub stops, or the machine spends a moment asking a dead port. """ from __future__ import annotations import logging from typing import Callable from kadence_desktop.blocking import processes from kadence_desktop.blocking.hosts import HostsBlocker from kadence_desktop.blocking.resolver import ResolvedRoute, StubResolver, split_routable log = logging.getLogger(__name__) class BlockerManager: def __init__( self, *, helper_path: str, report: Callable[[str, str, str, str | None], None], notify: Callable[[str, str], None] | None = None, stub: StubResolver | None = None, route: ResolvedRoute | None = None, ) -> None: self._hosts = HostsBlocker(helper_path) # Injectable so tests can drive the ordering without binding ports or # asking a real systemd-resolved for anything. self._stub = stub if stub is not None else StubResolver() self._route = route if route is not None else ResolvedRoute(helper_path) self._report = report self._notify = notify self._active = False self._applied: list[str] = [] self._announced: set[int] = set() self._redirect_ip = "127.0.0.1" self.enforce_hosts = True self.enforce_apps = True self.last_error: str | None = None self.route_error: str | None = None #: Patterns no machine can enforce at DNS level (`*gram*`); the UI says so. self.unroutable: list[str] = [] #: Zones currently handed to `systemd-resolved`. self.routed_zones: list[str] = [] self.killed_today = 0 if self._route.applied is not None: # A drop-in survived a crash: the zones point at this stub, so it has # to be listening *now*, before the first policy even arrives. self._stub.start() # ---- state ------------------------------------------------------------- @property def active(self) -> bool: return self._active @property def helper_installed(self) -> bool: return self._hosts.helper_installed @property def blocked_domains(self) -> list[str]: """What is really in `/etc/hosts` right now — not what the policy wants.""" return list(self._applied) @property def error(self) -> str | None: """The first thing that went wrong, hosts or DNS, for the UI to show.""" return self.last_error or self.route_error # ---- the one entry point ---------------------------------------------- def sync(self, policy: dict) -> None: """Bring the machine in line with `policy` (from /api/blocker/policy).""" should_block = bool(policy.get("active")) and bool(policy.get("enabled")) domains = [str(item) for item in policy.get("domains", [])] apps = [str(item) for item in policy.get("apps", [])] self._redirect_ip = str(policy.get("redirect_ip") or "127.0.0.1") patterns = [str(item) for item in policy.get("unexpandable", [])] allow = [str(item) for item in policy.get("allowed", [])] routable, unroutable, zones = split_routable(patterns) self.unroutable = unroutable if not self.enforce_hosts: routable, zones = [], [] self._sync_dns(routable, allow, zones, active=should_block) if should_block: if self.enforce_hosts and domains != self._applied: if self._hosts.apply(domains, self._redirect_ip): self._applied = domains self.last_error = None log.info("blocking %s domain(s)", len(domains)) else: self.last_error = self._hosts.last_error if self.enforce_apps and apps: self._enforce_apps(apps, kill=bool(policy.get("kill_apps", True))) self._active = True else: self.release() def _sync_dns( self, patterns: list[str], allow: list[str], zones: list[str], *, active: bool, ) -> None: """Put the stub and the resolved route in step, in a safe order.""" if not zones: if self._route.applied is not None: self.route_error = None if self._route.off() else self._route.last_error else: self.route_error = None # Nothing points at the stub any more, so it can go. if self._stub.running: self._stub.stop() self.routed_zones = [] return # The stub first. Writing the drop-in before anything listens on its port # would break resolution for the very zones we are trying to protect. if not self._stub.running and not self._stub.start(): self.route_error = self._stub.error if self._route.applied is not None: self._route.off() self.routed_zones = [] return if self._route.on(zones, self._stub.port or 0): self.route_error = None else: self.route_error = self._route.last_error self._stub.set_rules(patterns, allow, self._redirect_ip, active=active) self.routed_zones = zones def release(self) -> None: """Unblock everything. Called when the interval ends, and on shutdown.""" if self._applied or self._hosts_dirty(): if self._hosts.clear(): self._applied = [] self.last_error = None else: self.last_error = self._hosts.last_error self._active = False self._announced.clear() # The stub keeps running and keeps its route: the zones are still pointed # at it, so it has to answer — it just forwards everything again. self._stub.set_rules([], [], self._redirect_ip, active=False) def shutdown(self) -> None: """Give the machine's DNS back, then everything else. Called on exit.""" if self._route.applied is not None and not self._route.off(): self.route_error = self._route.last_error if self._stub.running: self._stub.stop() self.routed_zones = [] self.release() def _hosts_dirty(self) -> bool: from kadence_desktop.blocking.hosts import current_block return bool(current_block()) # ---- apps -------------------------------------------------------------- def _enforce_apps(self, patterns: list[str], *, kill: bool) -> None: for app in processes.scan(patterns): if app.pid in self._announced: continue self._announced.add(app.pid) if kill and processes.terminate(app): self.killed_today += 1 self._report("app", app.matched, "killed", f"{app.name} (pid {app.pid})") if self._notify: self._notify("Closed a distraction", f"{app.name} is blocked during focus.") else: self._report("app", app.matched, "warned", f"{app.name} (pid {app.pid})") if self._notify: self._notify("That app is blocked", f"{app.name} is on your block list.")