Sam Rolfe
1275b1baaa
Adds windows/, the Python client for the kids' Windows computers, and proves both halves of the Phase 4 definition of done on the real box (DESKTOP-TE2RHV2). The client: - subscribes, retained, to its own config on the 8883 TLS device plane with its own broker account (FR-3, FR-6); - keeps the last valid config on disk and loads it before connecting, so a broker outage changes nothing (FR-14); - terminates every process on the blocklist while the device is effectively locked, with the allowlist winning and the client's own tree plus critical system processes protected (FR-4, FR-16); - publishes presence and daily per-app usage; the day's totals are held on disk and republished when the link returns (FR-7); - shows an always-on-top warning window naming what is blocked, why and until when (FR-9). install-scheduled-task.ps1 registers the logon Scheduled Task for the kid's account with an interactive token, and grants that account Modify on the state directory: a file created from an elevated shell inherits Administrators only, and the limited task is otherwise refused with Access denied. Evidence (windows/EVIDENCE.md, from the machine's own log): notepad.exe killed inside a scheduled window (pid 16600, then 9724), the device online in the Control Service with locked/until/reason, a telemetry row for the device (app=690s), enforcement continuing with the broker unreachable (192.168.20.13:8884), the warning window on the interactive desktop, and the Scheduled Task registered at logon as New Comp. Domains are deliberately not reported: Windows has no cheap per-process DNS attribution, so the client sends "domains": [] and the gap is documented rather than guessed. 54 unit tests, run with `make windows-test` (no broker, no Windows, no psutil). AGENTS.md §11 is now 10 of 10.
103 lines
4.0 KiB
Python
103 lines
4.0 KiB
Python
"""Settings loading and the MQTT payload builders (no broker, no paho)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from webrestrict.mqtt_link import iso_now, status_payload, telemetry_payload
|
|
from webrestrict.settings import DEFAULT_PROTECTED, Settings
|
|
|
|
RAW = {
|
|
"broker_host": "192.168.20.13",
|
|
"broker_port": 8883,
|
|
"device_topic": "restrict/finn_s_computer_2/config",
|
|
"mqtt_username": "finn_s_computer_2",
|
|
"mqtt_password": "secret",
|
|
"ca_file": "C:/ProgramData/WebRestrict/ca.crt",
|
|
"state_dir": "C:/ProgramData/WebRestrict",
|
|
"unknown_future_field": "ignored",
|
|
}
|
|
|
|
|
|
class SettingsTests(unittest.TestCase):
|
|
def test_derives_topics_and_token(self):
|
|
settings = Settings.from_dict(RAW)
|
|
self.assertEqual(settings.device_token, "finn_s_computer_2")
|
|
self.assertEqual(settings.status_topic, "restrict/finn_s_computer_2/status")
|
|
self.assertEqual(settings.telemetry_topic, "restrict/finn_s_computer_2/telemetry")
|
|
self.assertEqual(settings.effective_client_id, "kid-finn_s_computer_2")
|
|
|
|
def test_unknown_fields_are_ignored(self):
|
|
settings = Settings.from_dict(RAW)
|
|
self.assertFalse(hasattr(settings, "unknown_future_field"))
|
|
|
|
def test_loads_from_a_file(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = Path(tmp) / "client.json"
|
|
path.write_text(json.dumps({**RAW, "state_dir": tmp}), encoding="utf-8")
|
|
settings = Settings.load(path)
|
|
self.assertEqual(settings.broker_port, 8883)
|
|
self.assertEqual(settings.log_path(), Path(tmp) / "client.log")
|
|
|
|
def test_validate_rejects_a_bad_topic(self):
|
|
settings = Settings.from_dict({**RAW, "device_topic": "restrict/finn"})
|
|
with self.assertRaises(ValueError):
|
|
settings.validate()
|
|
|
|
def test_validate_rejects_a_missing_account(self):
|
|
bad = {k: v for k, v in RAW.items() if k != "mqtt_username"}
|
|
with self.assertRaises(ValueError):
|
|
Settings.from_dict(bad).validate()
|
|
|
|
def test_validate_rejects_a_bad_port(self):
|
|
with self.assertRaises(ValueError):
|
|
Settings.from_dict({**RAW, "broker_port": 70000}).validate()
|
|
|
|
def test_protected_list_covers_the_system_and_the_interpreter(self):
|
|
for name in ("explorer", "svchost", "lsass", "python"):
|
|
self.assertIn(name, DEFAULT_PROTECTED)
|
|
|
|
def test_the_default_config_path_is_overridable(self):
|
|
from webrestrict.settings import default_config_path
|
|
|
|
previous = os.environ.get("WEBRESTRICT_CONFIG")
|
|
os.environ["WEBRESTRICT_CONFIG"] = "/tmp/webrestrict-test.json"
|
|
try:
|
|
self.assertEqual(str(default_config_path()), "/tmp/webrestrict-test.json")
|
|
finally:
|
|
if previous is None:
|
|
os.environ.pop("WEBRESTRICT_CONFIG", None)
|
|
else:
|
|
os.environ["WEBRESTRICT_CONFIG"] = previous
|
|
|
|
|
|
class PayloadTests(unittest.TestCase):
|
|
def test_status_reports_online_and_the_lock(self):
|
|
body = json.loads(status_payload(True, "notepad.exe"))
|
|
self.assertTrue(body["online"])
|
|
self.assertTrue(body["locked"])
|
|
self.assertEqual(body["app"], "notepad.exe")
|
|
self.assertTrue(body["at"].endswith("Z"))
|
|
|
|
def test_status_can_carry_a_tamper_report(self):
|
|
body = json.loads(status_payload(True, "", tamper={"kind": "client_stopped", "detail": "watchdog"}))
|
|
self.assertEqual(body["tamper"]["kind"], "client_stopped")
|
|
self.assertTrue(body["tamper"]["at"].endswith("Z"))
|
|
|
|
def test_telemetry_shape_is_the_list_form(self):
|
|
body = json.loads(telemetry_payload("2026-10-14", {"notepad.exe": 60, "steam.exe": 0}))
|
|
self.assertEqual(body["day"], "2026-10-14")
|
|
self.assertEqual(body["apps"], [{"name": "notepad.exe", "seconds": 60}])
|
|
self.assertEqual(body["domains"], [])
|
|
|
|
def test_iso_now_is_utc(self):
|
|
self.assertRegex(iso_now(), r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|