timka123
a75a011edb
An expansion list of labels guesses which subdomains exist; a '*' entry makes the guess unnecessary. parse_expansions now reports (labels, wildcard, rejected) instead of silently keeping junk, expand_entry keeps the base domain plus the configured labels and leaves '*.base' as a pattern when the wildcard is on, and the policy grows the 'allowed' masks the desktop resolver needs to serve exceptions.
284 lines
10 KiB
Python
284 lines
10 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)
|
|
|
|
|
|
def is_allowed(host: str, allow_masks: list[str]) -> bool:
|
|
"""Whether an allow-list entry (plain or masked) covers this hostname."""
|
|
return any(fnmatch(host, mask) for mask in allow_masks)
|
|
|
|
|
|
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
|
|
|
|
|
|
def policy(session: Session, settings: AppSettings) -> dict[str, Any]:
|
|
"""What the desktop app polls: should I block, and what."""
|
|
blocker = settings.blocker
|
|
current = pomodoro.current_session(session)
|
|
active = False
|
|
reason = "disabled"
|
|
|
|
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,
|
|
"strict": blocker.strict,
|
|
"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"]
|