"""Client settings, read from one JSON file (AGENTS.md §11 Phase 4). Everything the client needs is local configuration: the broker address, the device's own topic and MQTT account, and where to keep its state. Nothing here is a second source of truth — the Control Service still owns the rules. The file is provisioned on the machine and is **never** committed. It holds the device's MQTT password, which must stay off the repository (AGENTS.md §10 rule 8). The path comes from ``--config``, then ``WEBRESTRICT_CONFIG``, then ``%PROGRAMDATA%\\WebRestrict\\client.json``. """ from __future__ import annotations import json import os from dataclasses import dataclass, field from pathlib import Path from typing import Any # Default location. ProgramData is machine-wide and readable by the New Comp # account, which is what the logon Scheduled Task runs as. DEFAULT_CONFIG_NAME = "client.json" DEFAULT_STATE_DIR_NAME = "WebRestrict" # Never kill these, whatever a blocklist says. The client must not be able to # take the machine down, and it must not kill itself (it runs as python.exe). # The client's own interpreter and its parent chain are added at runtime. DEFAULT_PROTECTED: tuple[str, ...] = ( "system", "system idle process", "registry", "smss", "csrss", "wininit", "winlogon", "services", "lsass", "svchost", "dwm", "explorer", "taskhostw", "sihost", "ctfmon", "conhost", "fontdrvhost", "audiodg", "msmpeng", "securityhealthservice", "searchindexer", "startmenuexperiencehost", "shellexperiencehost", "textinputhost", "applicationframehost", "python", "pythonw", ) def default_program_data() -> Path: """The machine-wide directory for client configuration and state.""" base = os.environ.get("PROGRAMDATA") or os.environ.get("ALLUSERSPROFILE") if base: return Path(base) / DEFAULT_STATE_DIR_NAME # Not Windows (unit tests, a lab box): keep it under the user's home. return Path.home() / ".webrestrict" def default_config_path() -> Path: """The config path: ``WEBRESTRICT_CONFIG`` wins, else ``%PROGRAMDATA%``.""" override = os.environ.get("WEBRESTRICT_CONFIG") if override: return Path(override) return default_program_data() / DEFAULT_CONFIG_NAME @dataclass class Settings: """Everything the client needs to run, with safe lab defaults. A missing field falls back to the default, so a hand-written config file only needs the values that differ. Unknown fields are ignored, so the file may grow without a client update. """ # --- the link --------------------------------------------------------- broker_host: str = "192.168.20.13" broker_port: int = 8883 ca_file: str = "" # empty = the platform trust store device_topic: str = "" mqtt_username: str = "" mqtt_password: str = "" client_id: str = "" # empty = derive from the topic token # --- local state ------------------------------------------------------ state_dir: str = "" # --- timings (seconds) ------------------------------------------------ poll_seconds: float = 5.0 telemetry_seconds: float = 60.0 status_seconds: float = 60.0 connect_timeout_seconds: float = 10.0 # --- behaviour -------------------------------------------------------- overlay: bool = True protected: list[str] = field(default_factory=lambda: list(DEFAULT_PROTECTED)) # --------------------------------------------------------------------- @property def state_path(self) -> Path: """The directory holding the cached config, the usage roll-up and the log.""" if self.state_dir: return Path(self.state_dir) if self.device_token: return default_program_data() / self.device_token return default_program_data() @property def device_token(self) -> str: """The middle topic level, which is also the broker account name.""" base = self.device_topic.removesuffix("/config") return base.rsplit("/", 1)[-1] if "/" in base else base @property def status_topic(self) -> str: return self.device_topic.removesuffix("/config") + "/status" @property def telemetry_topic(self) -> str: return self.device_topic.removesuffix("/config") + "/telemetry" @property def effective_client_id(self) -> str: return self.client_id or f"kid-{self.device_token or 'windows'}" def config_path(self) -> Path: return self.state_path / "config-cache.json" def telemetry_path(self) -> Path: return self.state_path / "telemetry.json" def log_path(self) -> Path: return self.state_path / "client.log" def validate(self) -> None: """Raise when the client cannot possibly work, instead of failing later.""" if not self.device_topic: raise ValueError("device_topic is not set (e.g. restrict/finn_s_computer_2/config)") if not self.device_topic.endswith("/config"): raise ValueError(f"device_topic must end with /config: {self.device_topic!r}") if not self.broker_host: raise ValueError("broker_host is not set") if not 1 <= self.broker_port <= 65535: raise ValueError(f"broker_port is out of range: {self.broker_port}") if not self.mqtt_username: raise ValueError("mqtt_username is not set (it is the topic token)") # --------------------------------------------------------------------- @classmethod def from_dict(cls, raw: dict[str, Any]) -> "Settings": """Build settings from a parsed JSON object, ignoring unknown keys.""" known = {f for f in cls.__dataclass_fields__} # type: ignore[attr-defined] data = {k: v for k, v in raw.items() if k in known} settings = cls(**data) settings.broker_port = int(settings.broker_port) return settings @classmethod def load(cls, path: str | os.PathLike[str] | None = None) -> "Settings": """Read the config file and validate it.""" target = Path(path) if path else default_config_path() with open(target, "r", encoding="utf-8") as handle: raw = json.load(handle) if not isinstance(raw, dict): raise ValueError(f"{target}: expected a JSON object") settings = cls.from_dict(raw) settings.validate() return settings