timka123
d280101bce
docker / image (push) Failing after 24s
lint / ruff (push) Failing after 28s
tests / backend (push) Successful in 37s
tests / desktop (push) Successful in 22s
tests / android-core (push) Successful in 49s
tests / web (push) Successful in 1m20s
android / apk (push) Successful in 8m49s
357 lines
13 KiB
Python
357 lines
13 KiB
Python
"""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"]
|