internet_use_control/windows/webrestrict/config.py

"""The device config and its on-disk cache (AGENTS.md FR-14).

``restrict/<device>/config`` carries the desired state, exactly as
``internal/mqtt/payload.go`` publishes it. This module parses that payload and
keeps the last valid copy on disk, so a device that cannot reach the broker
still knows its rules. Only a valid, not-older config replaces the cache:
a malformed or stale message never leaves the device without rules.
"""

from __future__ import annotations

import json
import os
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable

# The schema version this client understands. A payload with a higher version is
# rejected rather than guessed at; the cached config stays in force.
CONFIG_VERSION = 1


def parse_time(value: str | None) -> datetime | None:
    """Parse an RFC 3339 timestamp as the Control Service writes it.

    Accepts a trailing ``Z`` and an optional fractional part. Returns ``None``
    for an empty or unparseable value, so a bad timestamp downgrades to "no end
    time" instead of raising.
    """
    if not value:
        return None
    text = value.strip()
    if text.endswith("Z") or text.endswith("z"):
        text = text[:-1] + "+00:00"
    try:
        moment = datetime.fromisoformat(text)
    except ValueError:
        return None
    if moment.tzinfo is None:
        moment = moment.replace(tzinfo=timezone.utc)
    return moment


def bare(name: str) -> str:
    """Normalise a process or list entry for comparison.

    ``notepad.exe``, ``Notepad.EXE`` and ``notepad`` are the same app. The
    Windows process list reports ``.exe``; a blocklist entry may or may not.
    """
    value = (name or "").strip().casefold()
    if value.endswith(".exe"):
        value = value[:-4]
    return value


@dataclass
class DeviceConfig:
    """The desired state for one device (TRANSPORT_DESIGN.md, Phase 3)."""

    v: int = CONFIG_VERSION
    device_id: int = 0
    name: str = ""
    platform: str = "windows"
    locked: bool = False
    until: str | None = None
    allow: list[str] = field(default_factory=list)
    block: list[str] = field(default_factory=list)
    reason: str = ""
    grant_until: str | None = None
    generated_at: str | None = None

    # -- parsing -----------------------------------------------------------
    @classmethod
    def from_dict(cls, raw: dict[str, Any]) -> "DeviceConfig":
        """Parse a payload, ignoring unknown fields (forward compatible).

        Raises ``ValueError`` on a payload this client cannot honour, so the
        caller keeps the last good config.
        """
        version = raw.get("v", CONFIG_VERSION)
        try:
            version = int(version)
        except (TypeError, ValueError) as exc:
            raise ValueError(f"config v is not an integer: {version!r}") from exc
        if version > CONFIG_VERSION:
            raise ValueError(f"config v={version} is newer than this client ({CONFIG_VERSION})")

        return cls(
            v=version,
            device_id=int(raw.get("device_id") or 0),
            name=str(raw.get("name") or ""),
            platform=str(raw.get("platform") or "windows"),
            locked=bool(raw.get("locked")),
            until=_str_or_none(raw.get("until")),
            allow=_string_list(raw.get("allow")),
            block=_string_list(raw.get("block")),
            reason=str(raw.get("reason") or ""),
            grant_until=_str_or_none(raw.get("grant_until")),
            generated_at=_str_or_none(raw.get("generated_at")),
        )

    @classmethod
    def from_json(cls, payload: str | bytes) -> "DeviceConfig":
        if isinstance(payload, (bytes, bytearray)):
            payload = payload.decode("utf-8", "replace")
        data = json.loads(payload)
        if not isinstance(data, dict):
            raise ValueError("config payload is not a JSON object")
        return cls.from_dict(data)

    def to_json(self) -> str:
        return json.dumps(self.to_dict(), sort_keys=True)

    def to_dict(self) -> dict[str, Any]:
        return {
            "v": self.v,
            "device_id": self.device_id,
            "name": self.name,
            "platform": self.platform,
            "locked": self.locked,
            "until": self.until,
            "allow": list(self.allow),
            "block": list(self.block),
            "reason": self.reason,
            "grant_until": self.grant_until,
            "generated_at": self.generated_at,
        }

    # -- effective state --------------------------------------------------
    def until_at(self) -> datetime | None:
        return parse_time(self.until)

    def grant_at(self) -> datetime | None:
        return parse_time(self.grant_until)

    def is_blocking(self, now: datetime | None = None) -> bool:
        """Whether the device must block right now.

        This mirrors the Control Service's *effective* state
        (``internal/store/schedules.go`` ``EffectiveState``): the payload already
        carries the resolved ``locked`` flag, so the client only has to apply
        the clock. An expired ``until`` lifts the block locally — the same
        instant the Control Service would clear it — a live ``grant_until``
        never blocks, and an indefinite lock (``until`` absent) never lifts.
        """
        if not self.locked:
            return False
        moment = now or datetime.now(timezone.utc)
        grant = self.grant_at()
        if grant is not None and grant > moment:
            return False
        until = self.until_at()
        if until is not None and moment >= until:
            return False
        return True

    def blocked_names(self) -> set[str]:
        """The blocklist as normalised names, minus the allowlist (FR-4)."""
        allowed = {bare(item) for item in self.allow}
        return {bare(item) for item in self.block if bare(item)} - allowed

    def allowed_names(self) -> set[str]:
        return {bare(item) for item in self.allow if bare(item)}

    def is_blocked_name(self, process_name: str) -> bool:
        """Whether one process name is blocked, with the allowlist winning."""
        value = bare(process_name)
        if not value:
            return False
        if value in self.allowed_names():
            return False
        return value in self.blocked_names()

    @property
    def reason_text(self) -> str:
        """A short human reason for the overlay (FR-9)."""
        return {
            "schedule": "it is inside a scheduled screen-time window",
            "manual": "a parent locked this computer",
            "grant": "",
        }.get(self.reason, self.reason or "screen-time rules are active")

    def generation_key(self) -> tuple[str, int]:
        """A sortable key for "is this config newer?".

        The Control Service republishes a config whenever the state changes, so
        ``generated_at`` is normally enough. The fallback keeps a payload with a
        missing timestamp acceptable rather than dropping a real update.
        """
        return (self.generated_at or "", self.device_id)


def _str_or_none(value: Any) -> str | None:
    return None if value is None else str(value)


def _string_list(value: Any) -> list[str]:
    if value is None:
        return []
    if isinstance(value, str):
        return [value]
    if isinstance(value, Iterable):
        return [str(item) for item in value if str(item).strip()]
    return []


class ConfigCache:
    """The last valid config on disk (FR-14 "keep last state").

    A device that cannot reach the broker keeps enforcing what it already has.
    The cache is written atomically, so a power cut cannot leave a half-written
    file, and a cache that cannot be read is treated as empty rather than fatal.
    """

    def __init__(self, path: str | os.PathLike[str]):
        self.path = Path(path)
        self._current: DeviceConfig | None = None

    @property
    def current(self) -> DeviceConfig | None:
        """The in-memory config, loading from disk on first use."""
        if self._current is None:
            self._current = self._read()
        return self._current

    def load(self) -> DeviceConfig | None:
        return self.current

    def save(self, config: DeviceConfig) -> None:
        """Persist a config and make it current."""
        self.path.parent.mkdir(parents=True, exist_ok=True)
        temp = self.path.with_suffix(self.path.suffix + ".tmp")
        with open(temp, "w", encoding="utf-8") as handle:
            handle.write(config.to_json())
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temp, self.path)
        self._current = config

    def accept(self, payload: str | bytes) -> DeviceConfig | None:
        """Validate and store a payload.

        Returns the stored config, or ``None`` when the payload is rejected (a
        malformed message, a newer schema, or an older ``generated_at`` than the
        cache). A rejection leaves the current config untouched.
        """
        try:
            candidate = DeviceConfig.from_json(payload)
        except (ValueError, TypeError):
            # json.JSONDecodeError is a ValueError. A broker bug or a hostile
            # payload must never leave the device without rules (FR-14).
            return None
        existing = self.current
        if existing is not None and candidate.generation_key() < existing.generation_key():
            return None
        self.save(candidate)
        return candidate

    # ---------------------------------------------------------------------
    def _read(self) -> DeviceConfig | None:
        try:
            with open(self.path, "r", encoding="utf-8") as handle:
                return DeviceConfig.from_json(handle.read())
        except FileNotFoundError:
            return None
        except (OSError, ValueError, json.JSONDecodeError):
            return None