"""Blocking policy. The backend decides *whether* blocking should be on right now and *what* is on the list; the desktop app is the thing that actually enforces it (it is the only piece running on your machine with the privileges to rewrite `/etc/hosts` or kill a process). Keeping the policy here means the web UI can edit the rules, and a second machine running the desktop app obeys the same list. """ from __future__ import annotations import re from dataclasses import dataclass, field from datetime import date, datetime from fnmatch import fnmatch from typing import Any from sqlalchemy import func, select from sqlalchemy.orm import Session from app.models import BlockEvent from app.services import pomodoro from app.settings_schema import AppSettings from app.timeutil import ensure_utc, in_window, local_date, local_time # One label of a hostname: what may legally sit in front of `.example.com`. _LABEL_RE = re.compile(r"^[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?$") @dataclass(frozen=True) class Expansion: """What `subdomain_expansions` asked for, split three ways. * `labels` — real DNS labels, usable as hostnames (`web`, `m`). * `wildcard` — a bare `*` was in the list, meaning *every* subdomain. That is the entry people type when they mean "all of them", and it is now supported rather than refused: a hosts file still cannot express it (that is what [expand_entry] reports as unexpandable), but the desktop app routes the suffix through a local resolver stub and the phone matches the mask in the URL bar, so it works on both. * `rejected` — anything that is neither a label nor `*` (`not a label`, `..`), reported so the UI can say why it was ignored. """ labels: list[str] = field(default_factory=list) wildcard: bool = False rejected: list[str] = field(default_factory=list) def parse_expansions(raw: list[str]) -> Expansion: """Split `subdomain_expansions` into labels, the `*` flag, and rejects. The trap this replaced is worth stating plainly: `*` used to slip through as a subdomain label, so `*.telegram.org` expanded to the literal `*.telegram.org` — not a hostname, which the desktop app's own filter dropped without a word. The mask looked configured, `unexpandable` stayed empty, and `web.telegram.org` was never blocked. Silent, and exactly the shape of "the blocker does not work". """ labels: list[str] = [] rejected: list[str] = [] wildcard = False for entry in raw: label = entry.strip().lower() if not label: continue if label == "*": wildcard = True continue if _LABEL_RE.match(label): if label not in labels: labels.append(label) elif label not in rejected: rejected.append(label) return Expansion(labels=labels, wildcard=wildcard, rejected=rejected) #: What makes an entry a mask rather than a name, `fnmatch`'s own metacharacters. _WILDCARDS = "*?[" def is_mask(entry: str) -> bool: return any(char in entry for char in _WILDCARDS) def is_allowed(host: str, allow_masks: list[str]) -> bool: """Whether an allow-list entry (plain or masked) covers this hostname. A plain entry covers itself *and its subdomains*, which is the same rule a plain entry follows on the block side — `youtube.com` blocked catches `music.youtube.com`. It did not used to: an exception was glob-matched, so a plain one meant exactly one hostname while a plain block meant a subtree, and the two halves of one list read the same and behaved differently. A masked entry is glob-matched as it always was, so `docs.*` and `*.docs.google.com` are unchanged — and still the way to allow something *narrower* than a subtree. """ for mask in allow_masks: if is_mask(mask): if fnmatch(host, mask): return True elif host == mask or host.endswith(f".{mask}"): return True return False def expand_entry(raw: str, expansion: Expansion) -> tuple[list[str], str | None]: """Turn one blocklist entry into concrete hostnames. A hosts file has no wildcards, so a mask has to be expanded into real names: * `youtube.com` → `youtube.com`, `www.youtube.com`, `m.youtube.com` * `*.youtube.com` → the bare domain plus every configured subdomain * `*gram*` → nothing expandable; returned as a leftover pattern so the clients that *can* match masks (the phone's URL bar, and the desktop's resolver stub) know to enforce it With `*` in `subdomain_expansions`, `*.youtube.com` means *every* subdomain. The concrete names are still expanded — they block without any resolver help — and the pattern comes back alongside them for the stub to catch the rest. Returns `(hostnames, unexpandable_pattern)`. """ entry = raw.strip().lower().lstrip(".") if not entry: return [], None if entry.startswith("*."): base = entry[2:] if not base or "*" in base: return [], entry hosts = [base, *[f"{label}.{base}" for label in expansion.labels]] return hosts, (entry if expansion.wildcard else None) if "*" in entry: return [], entry return [entry, f"www.{entry}", f"m.{entry}"], None def expand_domains(settings: AppSettings) -> list[str]: """Every hostname the desktop app should point at the redirect IP. `youtube.com` means "that site", so `www.` and `m.` come along; masks are expanded as described above; and anything the allow list covers — itself maskable — is taken back out, which is how `*.google.com` blocked plus `*.docs.google.com` allowed works. """ blocker = settings.blocker allow_masks = [ entry.strip().lower().lstrip(".") for entry in blocker.allowed_domains if entry.strip() ] expansion = parse_expansions(blocker.subdomain_expansions) hosts: list[str] = [] for raw in blocker.blocked_domains: expanded, _ = expand_entry(raw, expansion) for host in expanded: if host not in hosts and not is_allowed(host, allow_masks): hosts.append(host) return hosts def unexpandable_masks(settings: AppSettings) -> list[str]: """Blocklist entries a hosts file cannot express. These are what the desktop app's local resolver stub is handed: patterns with a domain anchor (`*.youtube.com`, `*gram.com`) it can answer at DNS level, and `*gram*`-style patterns it has to report as unenforceable on this machine because no zone can be routed for them. """ expansion = parse_expansions(settings.blocker.subdomain_expansions) leftovers: list[str] = [] for raw in settings.blocker.blocked_domains: _, pattern = expand_entry(raw, expansion) if pattern and pattern not in leftovers: leftovers.append(pattern) return leftovers def schedule_active(settings: AppSettings, moment: datetime | None = None) -> bool: now_time = local_time(settings.timezone, moment) weekday = (moment or datetime.now()).weekday() for window in settings.blocker.schedule: if weekday in window.days and in_window(now_time, window.start, window.end): return True return False GUARD_LOCK_NOTE = "Finish or drop the interval to switch the focus guard off." """Said in one voice: the 409 detail, and the `lock_reason` every client renders.""" class GuardLocked(RuntimeError): """Switching the guard off was refused because an interval is live.""" def guard_locked(session: Session, settings: AppSettings) -> bool: """The guard is on and an interval is live, so it may not be switched off. Any kind, running or paused — `pomodoro.current_session` is exactly that. A break is not an escape hatch and neither is pausing the clock. Deliberately keyed off `blocker.enabled` rather than off `policy()["active"]`: in `scheduled` mode outside every window nothing is being blocked right now, and the guard still must not be quietly switched off before the window opens. """ return settings.blocker.enabled and pomodoro.current_session(session) is not None def ensure_guard_not_switched_off( session: Session, before: AppSettings, after: AppSettings ) -> None: """Refuse a settings write that would turn the guard off mid-interval. Compares the two documents rather than reading the patch. The patch is a deep merge of a shared object and arrives loosely typed — `{"enabled": "no"}` coerces to `False`, so inspecting the raw keys would have to re-implement pydantic to be right — and "did the stored value go from on to off" is the only question that cannot be fooled. It also means an unrelated patch, a full-document echo and a re-sent `enabled: false` all pass by construction. Switching the guard *on* is never refused. """ if not after.blocker.enabled and guard_locked(session, before): raise GuardLocked(GUARD_LOCK_NOTE) def policy(session: Session, settings: AppSettings) -> dict[str, Any]: """What the desktop app polls: should I block, and what. `active` and `locked` answer two different questions and do disagree: in `scheduled` mode outside every window, nothing is blocked (`active` false) while the guard is still locked on (`locked` true) because an interval runs. """ blocker = settings.blocker current = pomodoro.current_session(session) active = False reason = "disabled" # The same rule as `guard_locked()`, reusing the row this function already # fetched rather than querying twice — `test_blocker.py` pins the two together. locked = blocker.enabled and current is not None if blocker.enabled: if blocker.mode == "always": active, reason = True, "always on" elif blocker.mode == "scheduled": active = schedule_active(settings) reason = "inside a blocking window" if active else "outside every window" else: # during_work if current is not None and current.kind == "work" and current.status in ("running", "paused"): active, reason = True, "a work interval is running" elif ( current is not None and current.kind != "work" and blocker.block_during_breaks and current.status in ("running", "paused") ): active, reason = True, "blocking stays on through breaks" else: reason = "no work interval is running" return { "active": active, "reason": reason, "mode": blocker.mode, "enabled": blocker.enabled, "locked": locked, "lock_reason": GUARD_LOCK_NOTE if locked else "", "redirect_ip": blocker.redirect_ip, "domains": expand_domains(settings) if active else [], "apps": [app.strip() for app in blocker.blocked_apps if app.strip()] if active else [], "unexpandable": unexpandable_masks(settings), "invalid_expansions": parse_expansions(blocker.subdomain_expansions).rejected, "allowed": [ entry.strip().lower().lstrip(".") for entry in blocker.allowed_domains if entry.strip() ], "kill_apps": blocker.kill_blocked_apps, "notify_on_block": blocker.notify_on_block, "poll_seconds": blocker.app_poll_seconds, "session_id": current.id if current else None, "all_domains": expand_domains(settings), "all_apps": [app.strip() for app in blocker.blocked_apps if app.strip()], } def record_event( session: Session, settings: AppSettings, *, kind: str, target: str, action: str = "blocked", details: str | None = None, ) -> BlockEvent: current = pomodoro.current_session(session) event = BlockEvent( kind=kind, target=target.strip().lower(), action=action, details=details, local_date=local_date(settings.timezone), session_id=current.id if current else None, ) session.add(event) session.commit() return event def events_for_day(session: Session, settings: AppSettings, day: date | None = None) -> dict[str, Any]: day = day or local_date(settings.timezone) rows = list(session.scalars(select(BlockEvent).where(BlockEvent.local_date == day))) by_target = dict( session.execute( select(BlockEvent.target, func.count(BlockEvent.id)) .where(BlockEvent.local_date == day) .group_by(BlockEvent.target) .order_by(func.count(BlockEvent.id).desc()) ).all() ) return { "date": day.isoformat(), "total": len(rows), "by_target": [{"target": target, "count": count} for target, count in by_target.items()], "events": [ { "id": row.id, "kind": row.kind, "target": row.target, "action": row.action, "at": ensure_utc(row.at).isoformat(), "details": row.details, } for row in sorted(rows, key=lambda row: ensure_utc(row.at), reverse=True)[:100] ], } DEFAULT_DOMAINS = [ "*.youtube.com", "twitter.com", "x.com", "*.instagram.com", "tiktok.com", "reddit.com", "facebook.com", "vk.com", "twitch.tv", "netflix.com", "9gag.com", "pikabu.ru", ] DEFAULT_APPS = ["steam*", "discord", "telegram-desktop", "spotify"]