diff --git a/contracts/workout_poke_v1.json b/contracts/workout_poke_v1.json new file mode 100644 index 0000000..b6f3aff --- /dev/null +++ b/contracts/workout_poke_v1.json @@ -0,0 +1,6 @@ +{ + "_comment": "Shared literal fixture for the phone->PC workout poke (v1). Both the Python receiver and the Dart sender test against THESE bytes. body_utf8 is the exact request body; signature = hex(HMAC-SHA256(bytes.fromhex(key_hex), body_utf8.encode('utf-8'))). Never re-serialize before verifying.", + "key_hex": "7f3c9a1e5b2d4f6081a3c5e7092b4d6f8a1c3e5072946b8d0f2a4c6e8013579b", + "body_utf8": "{\"v\":1,\"sent_at_ms\":1791558449000,\"nonce\":\"0b1e6c2a-9f4d-4c1b-8e7a-3d5f6a7b8c9d\",\"sandbox\":false,\"record_id\":\"2026-10-09T15:10:40.953\",\"payload\":{\"date\":\"2026-10-09\",\"start_time\":\"2026-10-09T15:10:40.953\",\"workout_type\":\"A\",\"duration_seconds\":7020.5,\"succeeded\":true,\"exercises\":[{\"name\":\"Squat\",\"weight_kg\":100.0,\"sets\":[5,5,5,5,5]}],\"note\":\"zażółć gęślą jaźń\"}}", + "signature_hex": "98b0b80fe717079c800e79562f669fd0189d89c3578a5bbf973761a0e7a3a3f5" +} diff --git a/docs/DOCS-workout-poke-contract.md b/docs/DOCS-workout-poke-contract.md new file mode 100644 index 0000000..ecde2f5 --- /dev/null +++ b/docs/DOCS-workout-poke-contract.md @@ -0,0 +1,75 @@ +# Workout poke: phone -> PC wire contract (v1) + +The phone sends a finished StrongLifts session straight to the PC over the +home LAN, so the credit (shutdown time, gaming budget, notification) lands in +under a second instead of waiting for the GitHub/Firebase round trip. The +normal sync push still runs in parallel; the PC dedups by record id, so a +session that arrives both ways is credited once. + +## Transport + +- `POST http://:8773/v1/workout`, `Content-Type: application/json`. +- `` is a phone setting, default `192.168.1.43` (static on the PC). + LAN only: the PC firewall accepts tcp/8773 from `192.168.1.0/24` and nothing + else. No WireGuard. Off the home LAN the poke fails fast and the regular + sync is the only path. +- Header `X-Poke-Signature: `. + The PC verifies the MAC over the bytes **as received** and only then parses + them. Neither side may re-serialize JSON to compute or check the MAC. +- Key: 32 random bytes, hex, PC copy at `~/.config/workout_poke/key` + (mode 600); phone copy in the Android keystore, put there by + `scripts/pair_phone.sh` over adb. + +## Request body + +```json +{"v":1,"sent_at_ms":,"nonce":"","sandbox":, + "record_id":"","payload":{}} +``` + +- `payload` is the same map the phone puts in its sync log for the session; + the PC feeds `(record_id, payload)` to `ingest_session_records`, which + re-validates it (never trusts `succeeded`). +- `sandbox` is inside the signed body. A sandbox request is verified, timed + and answered, but **never written anywhere** and never credited. + +## PC checks, in order (each rejection is logged at warning) + +1. Signature matches -> else `401`. +2. `|now - sent_at_ms| <= 120 s` -> else `401` (replay window). +3. `nonce` not seen within the window -> else `409`. +4. `v == 1` and fields present -> else `400`. + +## Response (always JSON) + +```json +{"ok":,"credited":,"duplicate":,"sandbox":, + "shutdown":"HH:MM"|null,"gaming_budget_minutes":,"pc_ms":, + "reason":""} +``` + +The phone must treat `shutdown` and `gaming_budget_minutes` as nullable. + +Outcomes of a verified (200) request: + +| case | ok | credited | duplicate | +|---|---|---|---| +| newly credited | true | true | false | +| already in `log.json` (e.g. the sync got there first) | true | false | true | +| logged, but today's credit slot was already paid by another copy of the same workout | true | false | false | +| signed but does not count (too short, no date): `reason` says why | false | false | false | +| sandbox | true | false | false | + +Refusals are never 200. Besides the four checks above, the PC may answer +`404` (wrong path), `405` (not POST), `411` (no Content-Length), `413` (body +over 64 KiB) and `500` (internal error; the regular sync still credits). All +of them carry the same JSON shape with `ok:false` and `shutdown` and +`gaming_budget_minutes` both `null`. The PC answers a `413` without reading +the body, so the phone may see a connection reset instead of the reply. + +## Shared fixture + +`contracts/workout_poke_v1.json` holds a literal key, body and expected +signature (with a float and non-ASCII text, the usual cross-language traps). +The Python receiver and the Dart sender both test against it, so a drift on +either side fails that side's tests. diff --git a/install_systemd.sh b/install_systemd.sh index bf8e1e2..8facada 100755 --- a/install_systemd.sh +++ b/install_systemd.sh @@ -16,6 +16,7 @@ SYNC_TIMER_NAME="workout-sync.timer" BONUS_SERVICE_NAME="earner-bonus.service" BONUS_PATH_NAME="earner-bonus.path" RUNNERUP_WATCH_NAME="runnerup-watch.service" +WORKOUT_POKE_NAME="workout-poke.service" # Runtime dependencies. screen_lock.py imports tkinter at module scope. On Arch # the tkinter MODULE ships inside the `python` package, but the shared library @@ -140,6 +141,12 @@ PYTHONPATH="$SCRIPT_DIR" python3 -m screen_locker._earner_units "$USER_SERVICE_D # workout-sync at once instead of waiting for the next timer tick. cp "$SCRIPT_DIR/$RUNNERUP_WATCH_NAME" "$USER_SERVICE_DIR/$RUNNERUP_WATCH_NAME" +# The phone's LAN poke: a finished StrongLifts session is credited within a +# second instead of at the next sync. The key must exist before the unit +# starts, or it refuses to (docs/DOCS-workout-poke-contract.md). +bash "$SCRIPT_DIR/scripts/workout_poke_keygen.sh" +cp "$SCRIPT_DIR/$WORKOUT_POKE_NAME" "$USER_SERVICE_DIR/$WORKOUT_POKE_NAME" + # Update paths in the service file to use absolute paths REPO_ROOT="$SCRIPT_DIR" sed -i "s|WorkingDirectory=.*|WorkingDirectory=$REPO_ROOT|" "$USER_SERVICE_DIR/$SERVICE_NAME" @@ -168,6 +175,10 @@ systemctl --user enable --now "$BONUS_PATH_NAME" systemctl --user enable "$RUNNERUP_WATCH_NAME" systemctl --user restart "$RUNNERUP_WATCH_NAME" +# Enable the LAN workout poke listener (restart picks up new code) +systemctl --user enable "$WORKOUT_POKE_NAME" +systemctl --user restart "$WORKOUT_POKE_NAME" + # Verify enforcement is actually armed. enable can silently no-op when systemd # breaks an ordering cycle by deleting the timer's job -- exactly what happened # on 2026-08-04 -- so a successful `enable` is NOT proof the locker will run. diff --git a/remove_systemd.sh b/remove_systemd.sh index d1f6d4a..b1902ea 100755 --- a/remove_systemd.sh +++ b/remove_systemd.sh @@ -16,6 +16,7 @@ UNITS=( "earner-bonus.path" "earner-bonus.service" "runnerup-watch.service" + "workout-poke.service" ) for unit in "${UNITS[@]}"; do diff --git a/requirements.txt b/requirements.txt index be649e8..3d072b4 100644 --- a/requirements.txt +++ b/requirements.txt @@ -22,3 +22,7 @@ pytest-xdist==3.8.0 requests==2.34.2 ruff==0.16.10 types-requests==2.33.0.20261006 +# Direct dependency since the Firebase session stream (_session_stream) catches +# urllib3's own errors on the long-lived SSE read; requests pulls it in, but +# the module imports it itself, so it is declared rather than assumed. +urllib3==2.8.0 diff --git a/screen_locker/_poke_credit.py b/screen_locker/_poke_credit.py new file mode 100644 index 0000000..5ff0adf --- /dev/null +++ b/screen_locker/_poke_credit.py @@ -0,0 +1,213 @@ +"""Credit one poked StrongLifts session: the sync's ingest, minus the network. + +The phone POSTs a finished session straight to the PC +(``docs/DOCS-workout-poke-contract.md``); this applies it through exactly the +path a synced session takes -- :func:`ingest_session_records`, which +re-validates and dedups under ``log.lock``, then the shared +:meth:`~screen_locker._workout_credit.WorkoutCreditMixin._apply_credit_for_written_entry` +(slot-based credit, shutdown push, debt, the "Workout credited" notification). +Nothing here computes a reward of its own. + +Each call builds a FRESH headless locker, the same object +``--sync-only`` uses, so no state leaks from one request into the next. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import date +import logging +from typing import TYPE_CHECKING, Final, cast + +from screen_locker._cli import _headless_locker +from screen_locker._constants import ( + EXTRA_BENEFITS_FILE, + SHUTDOWN_BASE_FILE, + SHUTDOWN_CONFIG_FILE, + SICK_DAY_STATE_FILE, +) +from screen_locker._credit_notify import gaming_minutes_today +from screen_locker._day import today_str +from screen_locker._earned import hhmm +from screen_locker._extra_benefits import process_week_transition +from screen_locker._grace_floor import tonight_minutes +from screen_locker._session_sync import ingest_session_records, validate_session +from screen_locker._shutdown import read_shutdown_config +from screen_locker._shutdown_base import reset_to_base_if_new_day +from screen_locker.screen_lock import ScreenLocker + +if TYPE_CHECKING: + from collections.abc import Iterable, Mapping + from pathlib import Path + + from screen_locker._workout_credit import WorkoutCreditResult + + +__all__ = [ + "BatchOutcome", + "PokeOutcome", + "credit_session", + "credit_sessions", + "current_gaming_minutes", + "current_shutdown", +] + +_logger: Final = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class PokeOutcome: + """What crediting one poked session did; ``reason`` is for the phone's UI.""" + + ok: bool + credited: bool + duplicate: bool + reason: str + + +@dataclass(frozen=True) +class BatchOutcome: + """What one batch did: newly logged ids, and how many of those earned time.""" + + ingested: tuple[str, ...] + credited: int + + +class PokeLocker(ScreenLocker): + """The headless locker plus the two steps a poke needs, as public calls.""" + + def start_the_day(self) -> None: + """Run the new-day reset the locker's startup runs, in the same order. + + Mirrors ``StartupChecksMixin._check_non_verify_exits``, not + ``sync_now`` (which never resets): the week transition must be + processed BEFORE the reset, and the weekly bonus layered on only when + this call did the reset. If the poke stamped the day without them, the + locker's own reset later would be a no-op and that day's streak bonus + would be lost. + """ + for reward in process_week_transition(self.log_file, EXTRA_BENEFITS_FILE): + _logger.info("Weekly reward: %s", reward) + if reset_to_base_if_new_day( + SHUTDOWN_BASE_FILE, + self, + sick_day_state_file=SICK_DAY_STATE_FILE, + log_file=self.log_file, + ): + self._apply_weekly_shutdown_bonus() + + def credit_written( + self, entry: dict[str, str], prior_entries: list[dict[str, object]] + ) -> WorkoutCreditResult: + """``SyncMixin._credit_ingested_workout``, keeping the result.""" + self.workout_data = entry + return self._apply_credit_for_written_entry(prior_entries) + + +def fresh_locker() -> PokeLocker: + """The ``--sync-only`` headless locker: no Tk, no window, no lock.""" + return cast("PokeLocker", _headless_locker(PokeLocker)) + + +def current_shutdown() -> str | None: + """Tonight's shutdown from ``/etc/shutdown-schedule.conf`` as ``HH:MM``.""" + config = read_shutdown_config(SHUTDOWN_CONFIG_FILE) + if config is None: + _logger.warning( + "Workout poke: shutdown config unreadable, replying shutdown=null" + ) + return None + return hhmm(tonight_minutes(config, date.fromisoformat(today_str()))) + + +def current_gaming_minutes(log_file: Path) -> int | None: + """Today's gaming budget (``None`` and a warning when it cannot be summed).""" + return gaming_minutes_today(log_file) + + +def _precheck(record_id: str, payload: Mapping[str, object]) -> str | None: + """Why the session can never count, or ``None`` if it can. + + ``ingest_session_records`` applies the same rules but only returns what + it ingested, so a rejection would be indistinguishable from a duplicate. + """ + day = payload.get("date") + if not isinstance(day, str) or not day: + return f"session {record_id} has no usable date ({day!r}), so it cannot count" + valid, detail = validate_session(payload) + if not valid: + return f"session {record_id} does not count: {detail}" + return None + + +def credit_session(record_id: str, payload: Mapping[str, object]) -> PokeOutcome: + """Reset the day if new, then ingest and credit one session. + + The caller serialises calls; the ingest also holds ``log.lock`` and the + credit the shutdown-state lock, so a concurrent sync or locker pass is safe. + """ + locker = fresh_locker() + locker.start_the_day() + problem = _precheck(record_id, payload) + if problem is not None: + _logger.warning("Workout poke not credited: %s", problem) + return PokeOutcome(ok=False, credited=False, duplicate=False, reason=problem) + + results: list[WorkoutCreditResult] = [] + + def on_ingested( + entry: dict[str, str], prior_entries: list[dict[str, object]] + ) -> None: + results.append(locker.credit_written(entry, prior_entries)) + + ingested = ingest_session_records( + locker.log_file, [(record_id, payload)], on_ingested=on_ingested + ) + if not ingested or not results: + reason = ( + f"session {record_id} is already in the PC's log (the sync or an " + "earlier poke got there first); nothing new to credit" + ) + _logger.info("Workout poke duplicate: %s", reason) + return PokeOutcome(ok=True, credited=False, duplicate=True, reason=reason) + if results[0].already_counted_today: + reason = ( + f"session {record_id} was logged, but today's slot for it was already " + "paid by another copy of the same workout; no extra time" + ) + _logger.warning("Workout poke not credited: %s", reason) + return PokeOutcome(ok=True, credited=False, duplicate=False, reason=reason) + _logger.info("Workout poke credited session %s", record_id) + return PokeOutcome( + ok=True, credited=True, duplicate=False, reason="workout credited" + ) + + +def credit_sessions( + records: Iterable[tuple[str, Mapping[str, object]]], +) -> BatchOutcome: + """Reset the day if new, then ingest and credit a batch of synced sessions. + + ``SyncMixin._ingest_synced_sessions`` with one fresh locker for the whole + batch -- what the Firebase stream (:mod:`screen_locker._session_stream`) + runs per event. Already-logged ids cost one ``log.json`` read and write + nothing; invalid ones are warned about by the ingest itself. + """ + locker = fresh_locker() + locker.start_the_day() + results: list[WorkoutCreditResult] = [] + + def on_ingested( + entry: dict[str, str], prior_entries: list[dict[str, object]] + ) -> None: + results.append(locker.credit_written(entry, prior_entries)) + + ingested = ingest_session_records(locker.log_file, records, on_ingested=on_ingested) + credited = sum(not result.already_counted_today for result in results) + if len(results) > credited: + _logger.warning( + "%d streamed session(s) were logged but earned no time: today's slot " + "was already paid by another copy of the same workout", + len(results) - credited, + ) + return BatchOutcome(ingested=tuple(ingested), credited=credited) diff --git a/screen_locker/_poke_server.py b/screen_locker/_poke_server.py new file mode 100644 index 0000000..d444db4 --- /dev/null +++ b/screen_locker/_poke_server.py @@ -0,0 +1,218 @@ +"""The resident LAN listener that credits a phone workout within a second. + +``workout-poke.service`` runs ``python3 -m screen_locker.poke_server``: a +stdlib ``ThreadingHTTPServer`` on ``0.0.0.0:8773`` (the firewall admits only +192.168.1.0/24) answering ``POST /v1/workout`` per +``docs/DOCS-workout-poke-contract.md``. Every import -- the locker, the +earned-time registry, gatelock -- is paid once at startup, so a request costs +only the credit itself. + +The wire checks are pure (:mod:`screen_locker._poke_wire`); the crediting is +the sync's own ingest (:mod:`screen_locker._poke_credit`). One lock serialises +the nonce cache and every credit in this process; cross-process safety comes +from the locks the ingest and credit already hold. +""" + +from __future__ import annotations + +from http import HTTPStatus +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import json +import logging +import threading +import time +from typing import TYPE_CHECKING, Final + +from screen_locker import _poke_credit +from screen_locker._poke_wire import ( + NonceCache, + Rejection, + check_request, +) + +if TYPE_CHECKING: + from screen_locker._poke_wire import PokeRequest + +__all__ = ["MAX_BODY_BYTES", "POKE_PATH", "PokeServer"] + +_logger: Final = logging.getLogger(__name__) + +MAX_BODY_BYTES: Final = 64 * 1024 +POKE_PATH: Final = "/v1/workout" +_HANDLER_TIMEOUT_S: Final = 10 + + +def _reply( + outcome: _poke_credit.PokeOutcome, + *, + sandbox: bool = False, + snapshot: tuple[str | None, int | None] = (None, None), +) -> dict[str, object]: + """The contract's response shape; ``pc_ms`` is filled in by the handler. + + ``snapshot`` is (tonight's shutdown ``HH:MM``, gaming minutes); refusals + leave it null rather than tell an unverified caller anything. + """ + shutdown, gaming = snapshot + return { + "ok": outcome.ok, + "credited": outcome.credited, + "duplicate": outcome.duplicate, + "sandbox": sandbox, + "shutdown": shutdown, + "gaming_budget_minutes": gaming, + "pc_ms": 0, + "reason": outcome.reason, + } + + +def _refused(reason: str) -> dict[str, object]: + """A refusal's reply: not ok, nothing credited.""" + return _reply( + _poke_credit.PokeOutcome( + ok=False, credited=False, duplicate=False, reason=reason + ) + ) + + +def _process(request: PokeRequest) -> dict[str, object]: + """Credit (or, for a sandbox request, only answer) one verified poke.""" + locker = _poke_credit.fresh_locker() + if request.sandbox: + outcome = _poke_credit.PokeOutcome( + ok=True, + credited=False, + duplicate=False, + reason="sandbox request verified; nothing written or credited", + ) + else: + outcome = _poke_credit.credit_session(request.record_id, request.payload) + return _reply( + outcome, + sandbox=request.sandbox, + snapshot=( + _poke_credit.current_shutdown(), + _poke_credit.current_gaming_minutes(locker.log_file), + ), + ) + + +class PokeServer(ThreadingHTTPServer): + """Holds the key, the nonce cache and the one credit lock.""" + + daemon_threads = True + + def __init__(self, address: tuple[str, int], key: bytes) -> None: + """Bind ``address`` and serve :class:`PokeHandler` with ``key``.""" + super().__init__(address, PokeHandler) + self.key = key + self.nonces = NonceCache() + self.gate = threading.Lock() + + +class PokeHandler(BaseHTTPRequestHandler): + """``POST /v1/workout``; anything else is answered 404/405 in JSON.""" + + server: PokeServer + timeout = _HANDLER_TIMEOUT_S + server_version = "workout-poke/1" + + # The default access log (one stderr line per request) is kept: the + # journal is where it lands, and every refusal is also logged at warning. + + def do_GET(self) -> None: + """Only POST exists.""" + self._send( + *self._refusal( + Rejection( + HTTPStatus.METHOD_NOT_ALLOWED, f"only POST {POKE_PATH} exists" + ) + ) + ) + + def do_POST(self) -> None: + """Verify, credit and answer one poke; never let an error go unanswered.""" + started = time.perf_counter() + try: + status, reply = self._handle() + except Exception: + _logger.exception("Workout poke from %s crashed", self.client_address[0]) + status = HTTPStatus.INTERNAL_SERVER_ERROR + reply = _refused( + "the PC hit an internal error (see its journal); the " + "regular sync will still credit this session" + ) + reply["pc_ms"] = round((time.perf_counter() - started) * 1000) + self._send(status, reply) + + def _refusal(self, rejection: Rejection) -> tuple[HTTPStatus, dict[str, object]]: + """Log a refusal at warning; the reply carries the same reason.""" + _logger.warning( + "Workout poke from %s rejected (%d): %s", + self.client_address[0], + rejection.status, + rejection.reason, + ) + return rejection.status, _refused(rejection.reason) + + def _read_body(self) -> bytes | Rejection: + """Exactly Content-Length bytes of ``POST /v1/workout``, capped. + + Over :data:`MAX_BODY_BYTES` is refused before reading a byte of it. + """ + if self.path != POKE_PATH: + return Rejection( + HTTPStatus.NOT_FOUND, f"unknown path {self.path!r}; use {POKE_PATH}" + ) + raw = self.headers.get("Content-Length") + if raw is None: + return Rejection(HTTPStatus.LENGTH_REQUIRED, "Content-Length is required") + length = int(raw) if raw.strip().isdigit() else -1 + if length < 0: + return Rejection(HTTPStatus.BAD_REQUEST, f"bad Content-Length {raw!r}") + if length > MAX_BODY_BYTES: + return Rejection( + HTTPStatus.REQUEST_ENTITY_TOO_LARGE, + f"body is {length} bytes; the limit is {MAX_BODY_BYTES}", + ) + body = self.rfile.read(length) + if len(body) != length: + return Rejection( + HTTPStatus.BAD_REQUEST, + f"body ended after {len(body)} of {length} bytes", + ) + return body + + def _handle(self) -> tuple[HTTPStatus, dict[str, object]]: + """Run the contract's checks, then credit; returns (status, reply).""" + body = self._read_body() + if isinstance(body, Rejection): + return self._refusal(body) + with self.server.gate: + verdict = check_request( + self.server.key, + body, + self.headers.get("X-Poke-Signature"), + self.server.nonces, + now_ms=time.time_ns() // 1_000_000, + ) + if isinstance(verdict, Rejection): + return self._refusal(verdict) + reply = _process(verdict) + _logger.info( + "Workout poke %s (sandbox=%s): %s", + verdict.record_id, + verdict.sandbox, + reply["reason"], + ) + return HTTPStatus.OK, reply + + def _send(self, status: HTTPStatus, reply: dict[str, object]) -> None: + """Write ``reply`` as a JSON response and close the connection.""" + data = json.dumps(reply, ensure_ascii=False).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json; charset=utf-8") + self.send_header("Content-Length", str(len(data))) + self.send_header("Connection", "close") + self.end_headers() + self.wfile.write(data) diff --git a/screen_locker/_poke_wire.py b/screen_locker/_poke_wire.py new file mode 100644 index 0000000..2698bd7 --- /dev/null +++ b/screen_locker/_poke_wire.py @@ -0,0 +1,218 @@ +"""Phone -> PC workout poke: the wire checks, pure and stdlib-only. + +The binding contract is ``docs/DOCS-workout-poke-contract.md`` and its literal +fixture ``contracts/workout_poke_v1.json``. Everything here is a pure function +of its inputs (``now_ms`` is a parameter, never read from a clock), so the +shared-fixture test can feed it the fixture's bytes as-is. The HTTP plumbing +lives in :mod:`screen_locker._poke_server`, the crediting in +:mod:`screen_locker._poke_credit`. + +The MAC is computed over the raw body bytes *as received*; nothing here ever +re-serializes JSON, because a Dart and a Python encoder disagree on floats and +non-ASCII escaping. +""" + +from __future__ import annotations + +from dataclasses import dataclass +import hashlib +import hmac +from http import HTTPStatus +import json +import logging +from typing import Final + +__all__ = [ + "REPLAY_WINDOW_MS", + "NonceCache", + "PokeRequest", + "Rejection", + "check_request", + "parse_key", + "verify_signature", +] + +PROTOCOL_VERSION: Final = 1 +REPLAY_WINDOW_MS: Final = 120_000 +KEY_HEX_LENGTH: Final = 64 + +_logger: Final = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class PokeRequest: + """A request that passed every contract check. + + Attributes: + sent_at_ms: The phone's send time, epoch milliseconds. + nonce: The request's one-time id. + sandbox: Verify and answer, but write and credit nothing. + record_id: The session's sync record id. + payload: The session exactly as the phone synced it. + """ + + sent_at_ms: int + nonce: str + sandbox: bool + record_id: str + payload: dict[str, object] + + +@dataclass(frozen=True) +class Rejection: + """Why a request was refused, and with which HTTP status.""" + + status: HTTPStatus + reason: str + + +def parse_key(text: str) -> bytes: + """Decode the shared key file's content (64 hex chars, 32 bytes). + + Raises: + ValueError: With a sentence saying what is wrong with the key. + """ + stripped = text.strip() + if len(stripped) != KEY_HEX_LENGTH: + msg = f"key must be {KEY_HEX_LENGTH} hex characters, got {len(stripped)}" + raise ValueError(msg) + try: + return bytes.fromhex(stripped) + except ValueError as exc: + msg = f"key is not valid hex ({exc})" + raise ValueError(msg) from exc + + +def verify_signature(key: bytes, body: bytes, signature_header: str | None) -> bool: + """True iff ``signature_header`` is hex HMAC-SHA256(key, body). + + Compared as bytes: the header arrives latin-1 decoded, and + ``hmac.compare_digest`` raises on a non-ASCII ``str``. + """ + if not signature_header: + return False + expected = hmac.new(key, body, hashlib.sha256).hexdigest().encode("ascii") + given = signature_header.strip().lower().encode("latin-1", errors="replace") + return hmac.compare_digest(expected, given) + + +class NonceCache: + """Nonces seen inside the replay window, pruned by their ``sent_at``. + + Pruned by the request's own send time, not by when it arrived: a request + stamped up to 120 s in the future stays acceptable until ``sent_at + 120 + s``, so its nonce must be remembered at least that long. + """ + + def __init__(self) -> None: + """Start empty; the window is :data:`REPLAY_WINDOW_MS`.""" + self._seen: dict[str, int] = {} + + def check_and_add(self, nonce: str, sent_at_ms: int, now_ms: int) -> bool: + """Record ``nonce``; False if it was already seen inside the window.""" + self._seen = { + seen: sent + for seen, sent in self._seen.items() + if now_ms - sent <= REPLAY_WINDOW_MS + } + if nonce in self._seen: + return False + self._seen[nonce] = sent_at_ms + return True + + def __len__(self) -> int: + """How many nonces are remembered right now.""" + return len(self._seen) + + +def _is_int(value: object) -> bool: + """An ``int`` that is not a ``bool`` (JSON ``true`` decodes to one).""" + return isinstance(value, int) and not isinstance(value, bool) + + +def _schema_error(doc: dict[str, object]) -> str | None: + """The first missing or mistyped field after the replay checks, if any.""" + if not _is_int(doc.get("v")) or doc.get("v") != PROTOCOL_VERSION: + return f"unsupported protocol version {doc.get('v')!r}; expected 1" + if not isinstance(doc.get("sandbox"), bool): + return "field 'sandbox' must be true or false" + record_id = doc.get("record_id") + if not isinstance(record_id, str) or not record_id: + return "field 'record_id' must be a non-empty string" + if not isinstance(doc.get("payload"), dict): + return "field 'payload' must be a JSON object" + return None + + +def _decode(body: bytes) -> dict[str, object] | Rejection: + """The verified body as a JSON object, or a 400.""" + try: + doc = json.loads(body.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + # Signed by the key holder yet unparsable: a sender bug, not noise. + _logger.warning("Poke body is signed but not UTF-8 JSON: %s", exc) + return Rejection(HTTPStatus.BAD_REQUEST, f"body is not UTF-8 JSON ({exc})") + if not isinstance(doc, dict): + return Rejection(HTTPStatus.BAD_REQUEST, "body is not a JSON object") + return doc + + +def _replay_check( + doc: dict[str, object], nonces: NonceCache, now_ms: int +) -> tuple[int, str] | Rejection: + """Checks 2 and 3: fresh enough (401), nonce unseen (409). + + Returns ``(sent_at_ms, nonce)``; a missing or mistyped field either check + needs is a 400 at that point. + """ + sent_at = doc.get("sent_at_ms") + if not isinstance(sent_at, int) or isinstance(sent_at, bool): + return Rejection(HTTPStatus.BAD_REQUEST, "field 'sent_at_ms' must be an int") + if abs(now_ms - sent_at) > REPLAY_WINDOW_MS: + return Rejection( + HTTPStatus.UNAUTHORIZED, + f"request is stale: sent_at_ms is {(now_ms - sent_at) / 1000:+.0f} s " + "off the PC clock (+ = in the past), the window is 120 s; check " + "the phone's clock", + ) + nonce = doc.get("nonce") + if not isinstance(nonce, str) or not nonce: + return Rejection(HTTPStatus.BAD_REQUEST, "field 'nonce' must be a string") + if not nonces.check_and_add(nonce, sent_at, now_ms): + return Rejection(HTTPStatus.CONFLICT, f"nonce {nonce} was already used") + return sent_at, nonce + + +def check_request( + key: bytes, + body: bytes, + signature_header: str | None, + nonces: NonceCache, + now_ms: int, +) -> PokeRequest | Rejection: + """Run the contract's checks in order; the first failure wins. + + 1. signature -> 401, 2. ``|now - sent_at| <= 120 s`` -> 401, + 3. nonce unseen -> 409, 4. ``v == 1`` and fields present -> 400. + JSON is parsed only after the MAC over the raw bytes matched. + """ + if not verify_signature(key, body, signature_header): + return Rejection(HTTPStatus.UNAUTHORIZED, "signature does not match the key") + doc = _decode(body) + if isinstance(doc, Rejection): + return doc + replay = _replay_check(doc, nonces, now_ms) + if isinstance(replay, Rejection): + return replay + problem = _schema_error(doc) + if problem is not None: + return Rejection(HTTPStatus.BAD_REQUEST, problem) + sent_at, nonce = replay + payload = doc["payload"] + return PokeRequest( + sent_at_ms=sent_at, + nonce=nonce, + sandbox=doc["sandbox"] is True, + record_id=str(doc["record_id"]), + payload=dict(payload) if isinstance(payload, dict) else {}, + ) diff --git a/screen_locker/_rtdb_stream.py b/screen_locker/_rtdb_stream.py new file mode 100644 index 0000000..4a53ffe --- /dev/null +++ b/screen_locker/_rtdb_stream.py @@ -0,0 +1,119 @@ +"""Pure parsing for a Firebase RTDB REST stream of the device-log node. + +``GET /screen-locker-sync/devices.json`` with ``Accept: text/event-stream`` +answers with Server-Sent Events: ``put``/``patch`` carry +``{"path": ..., "data": ...}`` relative to the streamed node, plus +``keep-alive``, ``cancel`` and ``auth_revoked``. Everything here is network-free +so it can be unit-tested; the connection itself is +:mod:`screen_locker._session_stream`. + +Each device log is ONE JSON-string leaf at ``/log~2Ejson`` +(``crdt_sync._fbkeys`` escapes the dot), so an event normalises to "these +devices now hold this text (or nothing)". +""" + +from __future__ import annotations + +from dataclasses import dataclass +import json +from typing import TYPE_CHECKING, Final + +from crdt_sync._firebase import decode_key + +if TYPE_CHECKING: + from collections.abc import Iterable, Iterator + +__all__ = ["LOG_LEAF", "SseEvent", "device_updates", "parse_sse"] + +LOG_LEAF: Final = "log.json" + + +@dataclass(frozen=True) +class SseEvent: + """One Server-Sent Event: its name and its (joined) data lines.""" + + name: str + data: str + + +def parse_sse(lines: Iterable[str]) -> Iterator[SseEvent]: + """Yield events from decoded SSE lines; a blank line ends each event.""" + name = "" + data: list[str] = [] + for line in lines: + if not line: + if name or data: + yield SseEvent(name or "message", "\n".join(data)) + name, data = "", [] + elif line.startswith("event:"): + name = line[len("event:") :].strip() + elif line.startswith("data:"): + data.append(line[len("data:") :].lstrip()) + + +def _segments(path: str) -> list[str]: + """``/a/b~2Ec`` -> ``["a", "b.c"]`` (decoded, empty segments dropped).""" + return [decode_key(part) for part in path.split("/") if part] + + +def _texts_under(segments: list[str], data: object) -> dict[str, str | None]: + """Device -> log text for a ``put`` of ``data`` at ``segments``. + + ``None`` means that device's log is gone (deleted, or not a text blob -- + the latter is the caller's to warn about, see :func:`device_updates`). + + Raises: + TypeError: The data does not have the device-node shape at all. + """ + if not segments: # the whole node: {device: {leaf: text}} + if data is None: + return {} + if not isinstance(data, dict): + msg = f"devices node is {type(data).__name__}, not an object" + raise TypeError(msg) + return {decode_key(key): _leaf_text(value) for key, value in data.items()} + device = segments[0] + if len(segments) == 1: + return {device: _leaf_text(data)} + if segments[1:] == [LOG_LEAF]: + return {device: data if isinstance(data, str) else None} + return {} # something below a device that is not its log: irrelevant + + +def _leaf_text(device_node: object) -> str | None: + """The log text inside one ``{leaf: text}`` device node, if any.""" + if not isinstance(device_node, dict): + return None + for key, value in device_node.items(): + if decode_key(key) == LOG_LEAF and isinstance(value, str): + return value + return None + + +def device_updates(event: SseEvent) -> tuple[bool, dict[str, str | None]]: + """Normalise a ``put``/``patch`` into ``(is_full_snapshot, updates)``. + + ``updates`` maps a device id to its log's new full text, or ``None`` when + it no longer has one. A ``put`` at ``/`` is a full snapshot: the caller + must REPLACE its cache, since devices missing from it are gone. A + ``patch`` carries several children whose keys may be multi-segment paths. + + Raises: + ValueError: The data is not JSON. + TypeError: The JSON is not the documented ``{path, data}`` shape. + """ + body = json.loads(event.data) + if not isinstance(body, dict) or not isinstance(body.get("path"), str): + msg = f"{event.name} event without a path: {event.data[:120]!r}" + raise TypeError(msg) + base = _segments(body["path"]) + if event.name == "put": + return not base, _texts_under(base, body.get("data")) + children = body.get("data") + if not isinstance(children, dict): + msg = f"patch data is {type(children).__name__}, not an object" + raise TypeError(msg) + updates: dict[str, str | None] = {} + for key, value in children.items(): + updates.update(_texts_under([*base, *_segments(key)], value)) + return False, updates diff --git a/screen_locker/_session_stream.py b/screen_locker/_session_stream.py new file mode 100644 index 0000000..a6cb461 --- /dev/null +++ b/screen_locker/_session_stream.py @@ -0,0 +1,232 @@ +"""Off-LAN instant credit: hold a Firebase RTDB stream of the device logs. + +Off the home LAN the phone's only transport is Firebase, so without this a +finished session waits up to 15 minutes for ``workout-sync.timer``. This +thread, inside ``workout-poke.service``, keeps ``GET +/screen-locker-sync/devices.json`` open as Server-Sent Events and credits +each newly streamed session within about a second +(:class:`screen_locker._stream_cache.DeviceLogCache`). + +Free-tier budget (Spark: 10 GB/month download, no per-read quota). Every +(re)connect downloads the whole devices node once: 375,755 bytes measured on +2026-10-09 (18 device dirs; logged with each snapshot, after transfer +decoding, so an upper bound on the wire). The ID token lives one hour, so +that is ~24 snapshots a day, ~9 MB: less than the 15-min sync, which fetches +every device log on every tick. After the snapshot, a write streams only the +written device's whole log: ~83 KB for the current phone, ~43 KB for each PC +push. RTDB sends NO event for a byte-identical write (observed), and the +phone's home-screen ``syncNow`` skips the write when its log is unchanged, so +no-op syncs cost nothing. Nothing here issues a read beyond the stream. + +Every failure is logged at warning or above with what it means, the thread +backs off (1 s doubling to 60 s, jittered) and never touches the HTTP +listener. The sync stays the safety net: the stream only makes credit faster. +""" + +from __future__ import annotations + +from datetime import UTC, datetime +import logging +import random +import threading +import time +from typing import TYPE_CHECKING, Final + +import requests +import urllib3 + +from screen_locker._device import device_identity +from screen_locker._rtdb_stream import parse_sse +from screen_locker._stream_auth import StreamError, stream_auth +from screen_locker._stream_cache import DeviceLogCache +from screen_locker._workout_sync import _DEVICES_PREFIX + +if TYPE_CHECKING: + from collections.abc import Iterator + + from screen_locker._rtdb_stream import SseEvent + +__all__ = ["SessionStream"] + +_logger: Final = logging.getLogger(__name__) + +_FALLBACK: Final = ( + "off-LAN workouts will only be credited by the 15-min workout-sync until " + "this recovers" +) +_BACKOFF_MIN_S: Final = 1.0 +_BACKOFF_MAX_S: Final = 60.0 +# RTDB sends a keep-alive every ~30 s; silence past this means a dead socket. +_TIMEOUT_S: Final = (10, 90) +_LOGGED_KEEPALIVES: Final = 3 +_HTTP_AUTH_REJECTED: Final = (401, 403) +# A connection that ends sooner than this pays a backoff even when the end was +# planned (auth_revoked, renewal): no path may reconnect in a tight loop, and +# each reconnect re-downloads the snapshot. +_MIN_LIFETIME_S: Final = 300.0 +_STREAM_ERRORS: Final = ( + requests.RequestException, + urllib3.exceptions.HTTPError, + OSError, +) +_JITTER: Final = random.SystemRandom() + + +class SessionStream: + """A daemon thread holding the stream; one connection at a time.""" + + def __init__(self, gate: threading.Lock) -> None: + """``gate`` is the poke server's credit lock, so credits never race.""" + identity = device_identity() + self._cache = DeviceLogCache( + _DEVICES_PREFIX, + frozenset({identity.device_id, identity.legacy_id}), + gate, + ) + self._http = requests.Session() + self._stop = threading.Event() + self._got_snapshot = False + + def start(self) -> threading.Thread: + """Run :meth:`run_forever` on a daemon thread and return it.""" + thread = threading.Thread( + target=self.run_forever, name="firebase-session-stream", daemon=True + ) + thread.start() + return thread + + def stop(self) -> None: + """Ask the loop to exit after the current connection.""" + self._stop.set() + + def run_forever(self) -> None: + """Connect, stream, back off on failure; never raise out of the thread.""" + delay = _BACKOFF_MIN_S + while not self._stop.is_set(): + self._got_snapshot = False + opened = time.monotonic() + planned = self._run_one() + lived = time.monotonic() - opened >= _MIN_LIFETIME_S + if self._got_snapshot and lived: + delay = _BACKOFF_MIN_S + if planned and lived: + continue # a long-lived connection ending on schedule + pause = delay * _JITTER.uniform(0.5, 1.0) + delay = min(delay * 2, _BACKOFF_MAX_S) + _logger.warning("Reconnecting the session stream in %.1f s", pause) + self._stop.wait(pause) + + def _run_one(self) -> bool: + """One connection; True if it ended on purpose (renewal, auth_revoked).""" + try: + self._connect_once() + except StreamError as exc: + _logger.warning("Firebase session stream failed: %s; %s", exc, _FALLBACK) + return False + except Exception: + _logger.exception("Firebase session stream crashed; %s", _FALLBACK) + return False + return True + + def _connect_once(self) -> None: + """One connection; returns only when it ended on purpose. + + Raises: + StreamError: Auth, HTTP or protocol failure. + """ + auth = stream_auth(_DEVICES_PREFIX) + try: + response = self._http.get( + auth.url, + params={"auth": auth.token}, + headers={"Accept": "text/event-stream"}, + stream=True, + timeout=_TIMEOUT_S, + ) + except _STREAM_ERRORS as exc: + msg = f"network error opening the stream: {exc}" + raise StreamError(msg) from exc + with response: + if not response.ok: + kind = ( + "rejected (token or rules)" + if response.status_code in _HTTP_AUTH_REJECTED + else "failed" + ) + msg = ( + f"stream {kind}: HTTP {response.status_code} {response.text[:200]}" + ) + raise StreamError(msg) + _logger.info("Firebase session stream connected to %s", _DEVICES_PREFIX) + self._consume(response, auth.renew_at) + + def _consume(self, response: requests.Response, renew_at: datetime) -> None: + """Dispatch events until the server or the renew deadline ends it. + + Raises: + StreamError: The stream broke, was cancelled or closed unasked. + """ + meter = _ByteMeter() + keepalives = 0 + try: + for event in parse_sse(meter.lines(response)): + if event.name == "keep-alive": + keepalives += 1 + if keepalives <= _LOGGED_KEEPALIVES: + _logger.info("Session stream keep-alive #%d", keepalives) + elif event.name in {"put", "patch"}: + self._dispatch(event, meter.total) + elif event.name == "auth_revoked": + _logger.warning( + "Session stream token revoked/expired (%s); reconnecting", + event.data, + ) + return + elif event.name == "cancel": + msg = f"the database cancelled the stream ({event.data})" + raise StreamError(msg) + else: + _logger.warning("Session stream ignored event %r", event.name) + if datetime.now(UTC) >= renew_at: + _logger.info("Session stream renewing its ID token before expiry") + return + except _STREAM_ERRORS as exc: + msg = f"stream broke after {meter.total} bytes: {exc}" + raise StreamError(msg) from exc + msg = f"the server closed the stream after {meter.total} bytes" + raise StreamError(msg) + + def _dispatch(self, event: SseEvent, total_bytes: int) -> None: + """Hand one put/patch to the cache and log what it did.""" + try: + result = self._cache.handle(event) + except (ValueError, TypeError) as exc: + _logger.warning("Session stream event undecodable (%s); %s", exc, _FALLBACK) + return + self._got_snapshot |= result.snapshot + _logger.info( + "Session stream %s (%d bytes read on this connection): %s", + "snapshot" if result.snapshot else event.name, + total_bytes, + result.summary, + ) + + +class _ByteMeter: + """Splits the raw stream into lines as bytes arrive, counting them.""" + + def __init__(self) -> None: + self.total = 0 + + def lines(self, response: requests.Response) -> Iterator[str]: + """Yield decoded lines without waiting for a full read buffer. + + ``iter_lines`` reads fixed-size chunks and would sit on a 30-byte + keep-alive until more data came; ``read1`` returns what has arrived. + """ + pending = b"" + while chunk := response.raw.read1(65536): + self.total += len(chunk) + *complete, pending = (pending + chunk).split(b"\n") + for line in complete: + yield line.rstrip(b"\r").decode("utf-8") diff --git a/screen_locker/_stream_auth.py b/screen_locker/_stream_auth.py new file mode 100644 index 0000000..f1fb9bb --- /dev/null +++ b/screen_locker/_stream_auth.py @@ -0,0 +1,93 @@ +"""The ID token and URL for the Firebase session stream. + +Split out of :mod:`screen_locker._session_stream` for the 250-line cap. +Reuses crdt_sync's own token provider and the screen_locker credential cache +(``~/.config/screen_locker/firebase_auth.json``); it never signs in with the +password, which is stale -- only the cached refresh token authenticates. A +missing session is healed the way the sync heals it, +:func:`screen_locker._sync_client.try_recover_firebase_session`. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from typing import Final + +from crdt_sync import ( + ConfigError, + FirebaseAuthError, + FirebaseConfig, + FirebaseTokenProvider, + credential_store_for, +) +from crdt_sync._firebase import encode_path + +from screen_locker._sync_client import try_recover_firebase_session + +__all__ = ["StreamAuth", "StreamError", "stream_auth"] + +_APP: Final = "screen_locker" +# Inside crdt_sync's 5-minute refresh skew, so the reconnect mints a new token +# instead of being handed the dying one back. +_RENEW_BEFORE_EXPIRY: Final = timedelta(minutes=4) +_ASSUMED_LIFETIME: Final = timedelta(hours=1) + + +class StreamError(Exception): + """One stream connection failed in a way that warrants backing off.""" + + +@dataclass(frozen=True) +class StreamAuth: + """Where to stream from, with which token, and when to renew it.""" + + url: str + token: str + renew_at: datetime + + +def _provider(config: FirebaseConfig) -> FirebaseTokenProvider: + """A provider over the cached session, recovering one if there is none. + + Raises: + StreamError: No cached session, and recovery failed too. + """ + store = credential_store_for(_APP) + auth = FirebaseTokenProvider(config.api_key, store) + if auth.has_session(): + return auth + recovery = try_recover_firebase_session() + if not recovery.recovered: + msg = f"no cached Firebase session and recovery failed: {recovery.reason}" + raise StreamError(msg) + return FirebaseTokenProvider(config.api_key, store) + + +def stream_auth(path: str) -> StreamAuth: + """A fresh token for streaming ``path`` (a logical RTDB path). + + A new provider per call re-reads the cache, so a token the 15-min sync + refreshed in the meantime is picked up rather than refreshed again. + + Raises: + StreamError: The config, the session or the token is unusable. + """ + try: + config = FirebaseConfig.load() + except ConfigError as exc: + msg = f"Firebase config unusable: {exc}" + raise StreamError(msg) from exc + auth = _provider(config) + try: + token = auth.id_token() + except FirebaseAuthError as exc: + msg = f"could not get an ID token: {exc}" + raise StreamError(msg) from exc + saved = credential_store_for(_APP).load() + expires = saved.expires_at if saved else datetime.now(UTC) + _ASSUMED_LIFETIME + return StreamAuth( + url=f"{config.database_url.rstrip('/')}/{encode_path(path)}.json", + token=token, + renew_at=expires - _RENEW_BEFORE_EXPIRY, + ) diff --git a/screen_locker/_stream_cache.py b/screen_locker/_stream_cache.py new file mode 100644 index 0000000..622c28e --- /dev/null +++ b/screen_locker/_stream_cache.py @@ -0,0 +1,153 @@ +"""What the Firebase stream does with one ``put``/``patch``: cache, merge, credit. + +Holds every device's last-seen log text, so each event recomputes the +cross-device union with the sync's OWN decoder and tombstone rule +(:func:`screen_locker._sync_records.merge_device_texts`) without a single +extra Firebase read. The PC's own log stays in the union -- it carries the +tombstones ``_manual_push`` re-publishes every tick, and dropping it would +let a retracted session be credited again -- but an event that touched only +this PC's log never triggers a pass. + +Crediting is :func:`screen_locker._poke_credit.credit_sessions` under the +same in-process lock as the LAN poke handler. ``write_signed_entry`` already +dedups by workout id; the ``seen`` set here only spares the file I/O for +records this daemon has already handed to the ingest. +""" + +from __future__ import annotations + +from dataclasses import dataclass +import logging +import time +from typing import TYPE_CHECKING, Final + +from screen_locker import _poke_credit +from screen_locker._rtdb_stream import LOG_LEAF, device_updates +from screen_locker._sync_records import _session_records, merge_device_texts + +if TYPE_CHECKING: + import threading + + from screen_locker._rtdb_stream import SseEvent + +__all__ = ["DeviceLogCache", "EventResult"] + +_logger: Final = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class EventResult: + """What one event did, as a sentence for the journal.""" + + snapshot: bool + summary: str + + +class DeviceLogCache: + """Every device's last-streamed log text, plus records already ingested.""" + + def __init__( + self, prefix: str, own_ids: frozenset[str], gate: threading.Lock + ) -> None: + """Cache logs under ``prefix``; ``own_ids`` are this PC's device dirs.""" + self._prefix = prefix + self._own_ids = own_ids + self._gate = gate + self._texts: dict[str, str] = {} + self._seen: set[tuple[str, str]] = set() + + def _apply(self, updates: dict[str, str | None], *, full: bool) -> set[str]: + """Fold ``updates`` into the cache; return the devices that changed.""" + if full: + fresh = {dev: text for dev, text in updates.items() if text is not None} + changed = { + dev + for dev in fresh.keys() | self._texts.keys() + if fresh.get(dev) != self._texts.get(dev) + } + self._texts = fresh + return changed + changed = set() + for dev, text in updates.items(): + if text == self._texts.get(dev): + continue # byte-identical rewrite: nothing to read or credit + changed.add(dev) + if text is None: + self._texts.pop(dev, None) + else: + self._texts[dev] = text + return changed + + def _pending(self) -> list[tuple[str, dict[str, object], str]]: + """Union records this daemon has not yet handed to the ingest.""" + union = merge_device_texts( + ( + (f"{self._prefix}/{dev}/{LOG_LEAF}", text) + for dev, text in sorted(self._texts.items()) + ), + _session_records, + "sessions", + ) + return [ + (rid, payload, str(hlc)) + for rid, (payload, hlc) in sorted(union.items()) + if (rid, str(hlc)) not in self._seen + ] + + def handle(self, event: SseEvent) -> EventResult: + """Apply one ``put``/``patch`` and credit whatever it newly brought. + + Raises: + ValueError: The event's data is not JSON. + TypeError: The JSON is not the documented ``{path, data}`` shape. + """ + started = time.perf_counter() + full, updates = device_updates(event) + changed = self._apply(updates, full=full) + devices = ", ".join(sorted(changed)) or "none" + if not full and not changed: + return EventResult( + snapshot=False, summary="byte-identical rewrite; nothing to do" + ) + foreign = changed - self._own_ids + if not full and not foreign: + return EventResult( + snapshot=False, + summary=f"only this PC's own log changed ({devices}); no credit", + ) + pending = self._pending() + if not pending: + return EventResult( + snapshot=full, + summary=f"{len(self._texts)} device log(s) cached, changed: {devices}; " + f"every session already handed to the ingest -- no-op, no file I/O " + f"({_ms(started)} ms)", + ) + try: + with self._gate: + outcome = _poke_credit.credit_sessions( + [(rid, payload) for rid, payload, _hlc in pending] + ) + except Exception: + # Not a stream problem: keep the connection (a reconnect would + # re-download the snapshot and fail the same way). Nothing is + # marked seen, so the next event retries. + _logger.exception( + "Crediting %d streamed session(s) failed; the next stream event " + "retries, and the 15-min workout-sync still covers them", + len(pending), + ) + return EventResult(snapshot=full, summary="crediting failed (see above)") + self._seen.update((rid, hlc) for rid, _payload, hlc in pending) + return EventResult( + snapshot=full, + summary=f"{len(self._texts)} device log(s) cached, changed: {devices}; " + f"{len(pending)} session(s) checked, {len(outcome.ingested)} newly " + f"logged {list(outcome.ingested)}, {outcome.credited} credited " + f"({_ms(started)} ms)", + ) + + +def _ms(started: float) -> int: + """Milliseconds since ``started`` (a ``perf_counter`` reading).""" + return round((time.perf_counter() - started) * 1000) diff --git a/screen_locker/_sync_records.py b/screen_locker/_sync_records.py index 7c965f1..e9471a4 100644 --- a/screen_locker/_sync_records.py +++ b/screen_locker/_sync_records.py @@ -18,7 +18,7 @@ from crdt_sync import Hlc, Record from screen_locker._manual_workout import MANUAL_WORKOUT_SYNC_KIND if TYPE_CHECKING: - from collections.abc import Callable + from collections.abc import Callable, Iterable _logger = logging.getLogger(__name__) @@ -123,3 +123,48 @@ def _manual_records(log_json: str) -> dict[str, tuple[dict, Hlc]]: corrupt device log as "no manual records from that device". """ return _records_matching(log_json, _is_manual_payload) + + +def merge_device_texts( + texts: Iterable[tuple[str, str]], + extract: Callable[[str], dict[str, tuple[dict, Hlc]]], + what: str, +) -> dict[str, tuple[dict, Hlc]]: + """Merge ``extract``-selected records across already-fetched device logs. + + ``texts`` is ``(path, log_json)`` per device. Keeps the highest-HLC copy of + each record id (records are id-stable, so one workout mirrored into two + device logs dedups to one), then drops every id ANY device tombstoned. + + The pure half of ``_workout_sync._merge_device_records``, shared with the + Firebase stream (:mod:`screen_locker._session_stream`) so the two readers + cannot disagree on what counts. A corrupt log is skipped with a warning + naming it, never raised. + """ + merged: dict[str, tuple[dict, Hlc]] = {} + tombstoned: set[str] = set() + for path, text in texts: + try: + records = extract(text) + tombstoned |= _tombstoned_ids(text) + except (ValueError, KeyError, TypeError) as exc: + _logger.warning("Corrupt sync data at %s: %s", path, exc) + continue + for rid, (payload, hlc) in records.items(): + existing = merged.get(rid) + if existing is None or existing[1] < hlc: + merged[rid] = (payload, hlc) + + # Applied after the union, not per device: one device deleting a record + # while another still holds it live must delete it. crdt-sync's own merge + # makes deletion monotonic for exactly this reason, and this reader rolls + # its own merge, so it has to honour the same rule itself. + for rid in tombstoned & merged.keys(): + _logger.info( + "Synced record %s is tombstoned on another device — dropping it, " + "so a deleted %s stops counting here too", + rid, + what, + ) + del merged[rid] + return merged diff --git a/screen_locker/_workout_sync.py b/screen_locker/_workout_sync.py index 3055edd..709d35b 100644 --- a/screen_locker/_workout_sync.py +++ b/screen_locker/_workout_sync.py @@ -36,7 +36,7 @@ from screen_locker._sync_records import ( _manual_records, _records_matching, _session_records, - _tombstoned_ids, + merge_device_texts, ) from screen_locker._sync_retry import with_sync_retry @@ -116,8 +116,7 @@ def _merge_device_records( raise return {} - merged: dict[str, tuple[dict, Hlc]] = {} - tombstoned: set[str] = set() + texts: list[tuple[str, str]] = [] for device in devices: path = f"{_DEVICES_PREFIX}/{device}/log.json" try: @@ -131,33 +130,9 @@ def _merge_device_records( what, ) continue - if text is None: - continue - try: - records = extract(text) - tombstoned |= _tombstoned_ids(text) - except (ValueError, KeyError, TypeError) as exc: - _logger.warning("Corrupt sync data at %s: %s", path, exc) - continue - for rid, (payload, hlc) in records.items(): - existing = merged.get(rid) - if existing is None or existing[1] < hlc: - merged[rid] = (payload, hlc) - - # Applied after the union, not per device: one device deleting a record - # while another still holds it live must delete it. crdt-sync's own merge - # makes deletion monotonic for exactly this reason, and this reader rolls - # its own merge, so it has to honour the same rule itself. - for rid in tombstoned & merged.keys(): - _logger.info( - "Synced record %s is tombstoned on another device — dropping it, " - "so a deleted %s stops counting here too", - rid, - what, - ) - del merged[rid] - - return merged + if text is not None: + texts.append((path, text)) + return merge_device_texts(texts, extract, what) def pull_synced_workout() -> tuple[dict | None, str | None]: diff --git a/screen_locker/poke_server.py b/screen_locker/poke_server.py new file mode 100644 index 0000000..4aeeb85 --- /dev/null +++ b/screen_locker/poke_server.py @@ -0,0 +1,118 @@ +"""``python3 -m screen_locker.poke_server``: the phone -> PC workout poke daemon. + +What ``workout-poke.service`` runs: load the shared key, warm the imports and +the budget sum, start the Firebase session stream (off-LAN credit, +:mod:`screen_locker._session_stream`) on a daemon thread, then serve ``POST +/v1/workout`` on port 8773 for good. The request handling lives in +:mod:`screen_locker._poke_server`; the contract is +``docs/DOCS-workout-poke-contract.md``. + +A missing, malformed or group/world-readable key is a refusal to start, +logged at critical, so ``systemctl --user status workout-poke`` says why. +""" + +from __future__ import annotations + +import argparse +import logging +from pathlib import Path +import stat +import sys +from typing import Final + +from screen_locker import _poke_credit +from screen_locker._poke_server import POKE_PATH, PokeServer +from screen_locker._poke_wire import parse_key +from screen_locker._session_stream import SessionStream + +__all__ = ["DEFAULT_KEY_FILE", "DEFAULT_PORT", "load_key", "main"] + +_logger: Final = logging.getLogger(__name__) + +DEFAULT_PORT: Final = 8773 +DEFAULT_KEY_FILE: Final = Path.home() / ".config" / "workout_poke" / "key" +_KEYGEN_HINT: Final = "run scripts/workout_poke_keygen.sh" + + +def load_key(path: Path) -> bytes: + """Read the shared key; refuse a world/group-readable or malformed one. + + Raises: + ValueError: With a sentence naming the problem and the fix. + OSError: The file cannot be read (usually: it does not exist). + """ + mode = stat.S_IMODE(path.stat().st_mode) + if mode & 0o077: + msg = f"{path} has mode {mode:o}; it must be 600 (chmod 600 {path})" + raise ValueError(msg) + return parse_key(path.read_text(encoding="ascii")) + + +def _parse_args(argv: list[str] | None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + # "" binds every interface; the firewall admits tcp/8773 from the LAN only. + parser.add_argument("--bind", default="", help="address (default: all)") + parser.add_argument("--port", type=int, default=DEFAULT_PORT) + parser.add_argument("--key-file", type=Path, default=DEFAULT_KEY_FILE) + parser.add_argument( + "--no-stream", + action="store_true", + help="skip the Firebase stream (off-LAN credit falls back to the sync)", + ) + return parser.parse_args(argv) + + +def _start_stream(server: PokeServer) -> None: + """Start the Firebase session stream; a failure here never stops the LAN poke.""" + try: + # Same lock as the HTTP handler: a poke and a streamed copy of one + # session must never be credited concurrently. + SessionStream(server.gate).start() + except Exception: + _logger.exception( + "Could not start the Firebase session stream; off-LAN workouts will " + "only be credited by the 15-min workout-sync until the daemon restarts" + ) + + +def main(argv: list[str] | None = None) -> int: + """Load the key, warm up, serve forever; 1 (logged critical) if it cannot.""" + logging.basicConfig( + level=logging.INFO, format="%(levelname)s %(name)s: %(message)s" + ) + args = _parse_args(argv) + try: + key = load_key(args.key_file) + except (OSError, ValueError) as exc: + _logger.critical("Refusing to start: bad poke key (%s); %s", exc, _KEYGEN_HINT) + return 1 + # Warm the budget sum (registry, ledgers) and the conf read once, so the + # first poke costs what every later one does. + _poke_credit.current_shutdown() + _poke_credit.current_gaming_minutes(_poke_credit.fresh_locker().log_file) + try: + server = PokeServer((args.bind, args.port), key) + except OSError as exc: + _logger.critical( + "Refusing to start: cannot bind %s:%d (%s)", args.bind, args.port, exc + ) + return 1 + _logger.info( + "Workout poke listening on %s:%d%s", args.bind or "*", args.port, POKE_PATH + ) + if args.no_stream: + _logger.warning( + "Firebase session stream disabled by --no-stream; off-LAN workouts " + "will only be credited by the 15-min workout-sync" + ) + else: + _start_stream(server) + try: + server.serve_forever() + finally: + server.server_close() + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/screen_locker/tests/test_poke_contract.py b/screen_locker/tests/test_poke_contract.py new file mode 100644 index 0000000..a6add0f --- /dev/null +++ b/screen_locker/tests/test_poke_contract.py @@ -0,0 +1,61 @@ +"""The PC receiver against the shared phone<->PC poke fixture. + +``contracts/workout_poke_v1.json`` is the literal both sides test against +(the Dart sender has the twin of this test). A drift in key decoding, body +encoding or MAC computation on this side fails here. +""" + +from __future__ import annotations + +from http import HTTPStatus +import json +from pathlib import Path + +from screen_locker._poke_wire import ( + NonceCache, + PokeRequest, + Rejection, + check_request, + parse_key, + verify_signature, +) + +_FIXTURE = Path(__file__).resolve().parents[2] / "contracts" / "workout_poke_v1.json" + + +def _fixture() -> tuple[bytes, bytes, str]: + data = json.loads(_FIXTURE.read_text(encoding="utf-8")) + key = parse_key(data["key_hex"]) + body = data["body_utf8"].encode("utf-8") + return key, body, data["signature_hex"] + + +def test_fixture_signature_verifies_over_raw_bytes() -> None: + key, body, signature = _fixture() + assert verify_signature(key, body, signature) + + +def test_one_altered_byte_is_rejected() -> None: + key, body, signature = _fixture() + altered = bytearray(body) + # The '1' in "v":1 -> '2': still valid JSON, so only the MAC can catch it. + altered[body.index(b'"v":1') + 4] = ord("2") + assert not verify_signature(key, bytes(altered), signature) + + +def test_fixture_passes_every_contract_check() -> None: + key, body, signature = _fixture() + sent_at = json.loads(body)["sent_at_ms"] + result = check_request(key, body, signature, NonceCache(), now_ms=sent_at) + assert isinstance(result, PokeRequest) + assert result.record_id == "2026-10-09T15:10:40.953" + assert result.payload["duration_seconds"] == 7020.5 + assert result.payload["note"] == "zażółć gęślą jaźń" + + +def test_altered_body_is_a_401_through_check_request() -> None: + key, body, signature = _fixture() + altered = body.replace(b'"sandbox":false', b'"sandbox":true ') + result = check_request(key, altered, signature, NonceCache(), now_ms=0) + assert isinstance(result, Rejection) + assert result.status is HTTPStatus.UNAUTHORIZED diff --git a/screen_locker/tests/test_poke_credit.py b/screen_locker/tests/test_poke_credit.py new file mode 100644 index 0000000..e907a28 --- /dev/null +++ b/screen_locker/tests/test_poke_credit.py @@ -0,0 +1,204 @@ +"""Crediting poked sessions: precheck, ingest, slot outcomes and the day reset.""" + +from __future__ import annotations + +import json +import logging +from typing import TYPE_CHECKING, Any +from unittest.mock import MagicMock + +import pytest + +from screen_locker import _poke_credit +from screen_locker._poke_credit import ( + BatchOutcome, + PokeLocker, + PokeOutcome, + credit_session, + credit_sessions, +) +from screen_locker._workout_credit import WorkoutCreditResult + +if TYPE_CHECKING: + from pathlib import Path + +_DAY = "2026-10-09" + + +def _session(**overrides: object) -> dict[str, Any]: + payload: dict[str, Any] = { + "workout_type": "B", + "date": _DAY, + "start_time": f"{_DAY}T15:10:40.953", + "duration_seconds": 7020, + "succeeded": True, + "exercises": [{"name": "Row", "succeeded": True}], + } + payload.update(overrides) + return payload + + +def _result(*, counted: bool = False) -> WorkoutCreditResult: + return WorkoutCreditResult( + shutdown_adjusted=not counted, + new_debt=None, + extra_bonus_delta=0, + weekly_count=1, + already_counted_today=counted, + ) + + +@pytest.fixture +def log_file(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + """A fresh PokeLocker on a tmp log; the day reset and the credit stubbed.""" + log = tmp_path / "poke_log.json" + + def headless(locker_cls: type[PokeLocker]) -> PokeLocker: + locker = object.__new__(locker_cls) + locker.log_file = log + locker.workout_data = {} + return locker + + monkeypatch.setattr(_poke_credit, "_headless_locker", headless) + monkeypatch.setattr(_poke_credit, "process_week_transition", lambda *_: []) + monkeypatch.setattr(_poke_credit, "reset_to_base_if_new_day", lambda *_a, **_k: 0) + return log + + +@pytest.fixture +def credit(monkeypatch: pytest.MonkeyPatch) -> MagicMock: + """Replace the shared credit step; the ingest itself stays real.""" + stub = MagicMock(return_value=_result()) + monkeypatch.setattr(PokeLocker, "_apply_credit_for_written_entry", stub) + return stub + + +def _logged_ids(log: Path) -> list[str]: + data = json.loads(log.read_text()) + return [e["workout_data"]["sync_record_id"] for es in data.values() for e in es] + + +class TestCreditSession: + def test_happy_path_logs_and_credits_once( + self, log_file: Path, credit: MagicMock + ) -> None: + outcome = credit_session("r1", _session()) + assert outcome == PokeOutcome( + ok=True, credited=True, duplicate=False, reason="workout credited" + ) + assert _logged_ids(log_file) == [f"{_DAY}T15:10:40.953"] + credit.assert_called_once() + + def test_second_poke_of_the_same_session_is_a_duplicate( + self, log_file: Path, credit: MagicMock + ) -> None: + credit_session("r1", _session()) + outcome = credit_session("r1", _session()) + assert (outcome.ok, outcome.credited, outcome.duplicate) == (True, False, True) + assert "already in the PC's log" in outcome.reason + assert credit.call_count == 1 + assert len(_logged_ids(log_file)) == 1 + + def test_slot_already_paid_logs_but_credits_nothing( + self, + log_file: Path, + credit: MagicMock, + caplog: pytest.LogCaptureFixture, + ) -> None: + credit.return_value = _result(counted=True) + with caplog.at_level(logging.WARNING): + outcome = credit_session("r1", _session()) + assert (outcome.ok, outcome.credited, outcome.duplicate) == (True, False, False) + assert "slot" in outcome.reason + assert "not credited" in caplog.text + assert len(_logged_ids(log_file)) == 1 + + @pytest.mark.parametrize("day", [None, "", 20261009]) + def test_session_without_a_usable_date_is_refused( + self, log_file: Path, credit: MagicMock, day: object + ) -> None: + outcome = credit_session("r1", _session(date=day)) + assert (outcome.ok, outcome.credited, outcome.duplicate) == ( + False, + False, + False, + ) + assert "no usable date" in outcome.reason + assert not log_file.exists() + credit.assert_not_called() + + @pytest.mark.parametrize( + ("overrides", "fragment"), + [ + ({"exercises": []}, "no exercises"), + ({"duration_seconds": 600}, "10 min"), + ({"duration_seconds": "long"}, "not a number"), + ], + ) + def test_invalid_session_is_refused_with_the_reason( + self, + log_file: Path, + credit: MagicMock, + overrides: dict[str, object], + fragment: str, + ) -> None: + outcome = credit_session("r1", _session(**overrides)) + assert outcome.ok is False + assert "does not count" in outcome.reason + assert fragment in outcome.reason + assert not log_file.exists() + credit.assert_not_called() + + def test_the_day_is_reset_before_the_credit( + self, log_file: Path, credit: MagicMock, monkeypatch: pytest.MonkeyPatch + ) -> None: + order: list[str] = [] + + def reset(_self: PokeLocker) -> None: + order.append("reset") + + def paid(*_args: object) -> WorkoutCreditResult: + order.append("credit") + return _result() + + monkeypatch.setattr(PokeLocker, "start_the_day", reset) + credit.side_effect = paid + credit_session("r1", _session()) + assert order == ["reset", "credit"] + + +class TestCreditSessions: + def test_batch_counts_ingested_and_credited( + self, log_file: Path, credit: MagicMock + ) -> None: + other = _session(date="2026-10-08", start_time="2026-10-08T09:00:00.000") + outcome = credit_sessions([("r1", _session()), ("r2", other)]) + assert outcome == BatchOutcome(ingested=("r1", "r2"), credited=2) + assert credit.call_count == 2 + + def test_already_logged_ids_cost_nothing( + self, log_file: Path, credit: MagicMock + ) -> None: + credit_sessions([("r1", _session())]) + outcome = credit_sessions([("r1", _session())]) + assert outcome == BatchOutcome(ingested=(), credited=0) + assert credit.call_count == 1 + + def test_invalid_session_in_a_batch_is_skipped( + self, log_file: Path, credit: MagicMock + ) -> None: + outcome = credit_sessions([("bad", _session(exercises=[]))]) + assert outcome == BatchOutcome(ingested=(), credited=0) + assert not log_file.exists() + + def test_sessions_that_shared_a_paid_slot_are_warned_about( + self, + log_file: Path, + credit: MagicMock, + caplog: pytest.LogCaptureFixture, + ) -> None: + credit.return_value = _result(counted=True) + with caplog.at_level(logging.WARNING): + outcome = credit_sessions([("r1", _session())]) + assert outcome == BatchOutcome(ingested=("r1",), credited=0) + assert "1 streamed session(s)" in caplog.text diff --git a/screen_locker/tests/test_poke_credit_day.py b/screen_locker/tests/test_poke_credit_day.py new file mode 100644 index 0000000..83a8e7b --- /dev/null +++ b/screen_locker/tests/test_poke_credit_day.py @@ -0,0 +1,119 @@ +"""The poke credit's day reset, workout hand-off and the two reply snapshots.""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING +from unittest.mock import MagicMock + +import pytest + +from screen_locker import _poke_credit +from screen_locker._poke_credit import PokeLocker +from screen_locker._workout_credit import WorkoutCreditResult + +if TYPE_CHECKING: + from pathlib import Path + + +@pytest.fixture +def log_file(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + """A fresh PokeLocker on a tmp log; the day reset stubbed.""" + log = tmp_path / "poke_log.json" + + def headless(locker_cls: type[PokeLocker]) -> PokeLocker: + locker = object.__new__(locker_cls) + locker.log_file = log + locker.workout_data = {} + return locker + + monkeypatch.setattr(_poke_credit, "_headless_locker", headless) + monkeypatch.setattr(_poke_credit, "process_week_transition", lambda *_: []) + monkeypatch.setattr(_poke_credit, "reset_to_base_if_new_day", lambda *_a, **_k: 0) + return log + + +@pytest.fixture +def credit(monkeypatch: pytest.MonkeyPatch) -> MagicMock: + """Replace the shared credit step.""" + stub = MagicMock( + return_value=WorkoutCreditResult( + shutdown_adjusted=True, + new_debt=None, + extra_bonus_delta=0, + weekly_count=1, + already_counted_today=False, + ) + ) + monkeypatch.setattr(PokeLocker, "_apply_credit_for_written_entry", stub) + return stub + + +class TestStartTheDay: + def test_new_day_applies_the_weekly_bonus_after_the_reset( + self, log_file: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + calls: list[str] = [] + + def week(*_args: object) -> list[str]: + calls.append("week") + return ["+1h streak"] + + def reset(*_args: object, **_kwargs: object) -> bool: + calls.append("reset") + return True + + def bonus(_self: PokeLocker) -> None: + calls.append("bonus") + + monkeypatch.setattr(_poke_credit, "process_week_transition", week) + monkeypatch.setattr(_poke_credit, "reset_to_base_if_new_day", reset) + monkeypatch.setattr(PokeLocker, "_apply_weekly_shutdown_bonus", bonus) + _poke_credit.fresh_locker().start_the_day() + assert calls == ["week", "reset", "bonus"] + + def test_same_day_does_not_apply_the_weekly_bonus( + self, log_file: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + bonus = MagicMock() + monkeypatch.setattr(PokeLocker, "_apply_weekly_shutdown_bonus", bonus) + _poke_credit.fresh_locker().start_the_day() + bonus.assert_not_called() + + def test_credit_written_sets_the_entry_and_returns_the_result( + self, log_file: Path, credit: MagicMock + ) -> None: + locker = _poke_credit.fresh_locker() + assert locker.credit_written({"type": "x"}, [{"a": 1}]) == credit.return_value + assert locker.workout_data == {"type": "x"} + credit.assert_called_once_with([{"a": 1}]) + + +class TestSnapshots: + def test_current_shutdown_renders_tonights_time( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr(_poke_credit, "read_shutdown_config", lambda _p: (1, 2, 3)) + monkeypatch.setattr(_poke_credit, "tonight_minutes", lambda _c, _d: 20 * 60 + 5) + assert _poke_credit.current_shutdown() == "20:05" + + def test_unreadable_config_is_none_and_warned( + self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture + ) -> None: + monkeypatch.setattr(_poke_credit, "read_shutdown_config", lambda _p: None) + with caplog.at_level(logging.WARNING): + assert _poke_credit.current_shutdown() is None + assert "shutdown=null" in caplog.text + + def test_current_gaming_minutes_delegates( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + seen: list[Path] = [] + + def minutes(log: Path) -> int: + seen.append(log) + return 90 + + monkeypatch.setattr(_poke_credit, "gaming_minutes_today", minutes) + assert _poke_credit.current_gaming_minutes(tmp_path / "log.json") == 90 + assert seen == [tmp_path / "log.json"] diff --git a/screen_locker/tests/test_poke_main.py b/screen_locker/tests/test_poke_main.py new file mode 100644 index 0000000..dab8952 --- /dev/null +++ b/screen_locker/tests/test_poke_main.py @@ -0,0 +1,186 @@ +"""The poke daemon CLI: key loading, startup refusals, stream wiring.""" + +from __future__ import annotations + +import logging +import threading +from typing import TYPE_CHECKING, Any, ClassVar +from unittest.mock import MagicMock + +import pytest + +from screen_locker import _poke_credit, poke_server +from screen_locker._poke_server import POKE_PATH + +if TYPE_CHECKING: + from pathlib import Path + +_KEY_HEX = "ab" * 32 + + +def _key_file(tmp_path: Path, text: str = _KEY_HEX, mode: int = 0o600) -> Path: + path = tmp_path / "key" + path.write_text(text, encoding="ascii") + path.chmod(mode) + return path + + +class FakeServer: + """Stands in for PokeServer: records how the daemon drove it.""" + + instances: ClassVar[list[FakeServer]] = [] + + def __init__(self, address: tuple[str, int], key: bytes) -> None: + self.address = address + self.key = key + self.gate = threading.Lock() + self.closed = False + self.instances.append(self) + + def serve_forever(self) -> None: + return + + def server_close(self) -> None: + self.closed = True + + +@pytest.fixture(autouse=True) +def _daemon_stubs(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + """No real budget sum or server socket in any test here.""" + FakeServer.instances.clear() + locker = MagicMock() + locker.log_file = tmp_path / "log.json" + monkeypatch.setattr(_poke_credit, "fresh_locker", lambda: locker) + monkeypatch.setattr(_poke_credit, "current_shutdown", MagicMock()) + monkeypatch.setattr(_poke_credit, "current_gaming_minutes", MagicMock()) + monkeypatch.setattr(poke_server, "PokeServer", FakeServer) + + +@pytest.fixture(autouse=True) +def stream(monkeypatch: pytest.MonkeyPatch) -> MagicMock: + """The Firebase stream class; never a real connection.""" + stub = MagicMock() + monkeypatch.setattr(poke_server, "SessionStream", stub) + return stub + + +class TestLoadKey: + def test_reads_a_600_key(self, tmp_path: Path) -> None: + assert poke_server.load_key(_key_file(tmp_path)) == bytes.fromhex(_KEY_HEX) + + @pytest.mark.parametrize("mode", [0o640, 0o604, 0o660, 0o644]) + def test_group_or_world_access_is_refused_with_the_fix( + self, tmp_path: Path, mode: int + ) -> None: + with pytest.raises(ValueError, match="chmod 600"): + poke_server.load_key(_key_file(tmp_path, mode=mode)) + + def test_malformed_key_is_refused(self, tmp_path: Path) -> None: + with pytest.raises(ValueError, match="hex"): + poke_server.load_key(_key_file(tmp_path, text="short")) + + def test_missing_file_raises_oserror(self, tmp_path: Path) -> None: + with pytest.raises(OSError, match="key"): + poke_server.load_key(tmp_path / "key") + + +class TestMain: + def test_missing_key_refuses_to_start_at_critical( + self, tmp_path: Path, caplog: pytest.LogCaptureFixture + ) -> None: + with caplog.at_level(logging.CRITICAL): + code = poke_server.main(["--key-file", str(tmp_path / "nope")]) + assert code == 1 + assert "bad poke key" in caplog.text + assert "workout_poke_keygen.sh" in caplog.text + assert not FakeServer.instances + + def test_loose_key_mode_refuses_to_start(self, tmp_path: Path) -> None: + key = _key_file(tmp_path, mode=0o644) + assert poke_server.main(["--key-file", str(key)]) == 1 + assert not FakeServer.instances + + def test_bind_failure_refuses_to_start_at_critical( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + ) -> None: + def busy(_address: tuple[str, int], _key: bytes) -> Any: + msg = "address in use" + raise OSError(msg) + + monkeypatch.setattr(poke_server, "PokeServer", busy) + with caplog.at_level(logging.CRITICAL): + code = poke_server.main( + ["--key-file", str(_key_file(tmp_path)), "--port", "9"] + ) + assert code == 1 + assert "cannot bind :9 (address in use)" in caplog.text + + def test_serves_with_the_key_and_starts_the_stream( + self, + tmp_path: Path, + stream: MagicMock, + caplog: pytest.LogCaptureFixture, + ) -> None: + argv = ["--key-file", str(_key_file(tmp_path)), "--port", "1234"] + with caplog.at_level(logging.INFO): + code = poke_server.main([*argv, "--bind", "127.0.0.1"]) + (server,) = FakeServer.instances + assert code == 0 + assert server.address == ("127.0.0.1", 1234) + assert server.key == bytes.fromhex(_KEY_HEX) + assert server.closed + stream.assert_called_once_with(server.gate) + stream.return_value.start.assert_called_once_with() + assert f"127.0.0.1:1234{POKE_PATH}" in caplog.text + assert isinstance(_poke_credit.current_shutdown, MagicMock) + _poke_credit.current_shutdown.assert_called_once_with() + + def test_default_bind_is_all_interfaces_on_the_default_port( + self, tmp_path: Path, caplog: pytest.LogCaptureFixture + ) -> None: + with caplog.at_level(logging.INFO): + poke_server.main(["--key-file", str(_key_file(tmp_path))]) + assert FakeServer.instances[0].address == ("", poke_server.DEFAULT_PORT) + assert f"*:{poke_server.DEFAULT_PORT}" in caplog.text + + def test_no_stream_skips_firebase_and_says_so( + self, + tmp_path: Path, + stream: MagicMock, + caplog: pytest.LogCaptureFixture, + ) -> None: + with caplog.at_level(logging.WARNING): + code = poke_server.main( + ["--key-file", str(_key_file(tmp_path)), "--no-stream"] + ) + assert code == 0 + stream.assert_not_called() + assert "disabled by --no-stream" in caplog.text + assert FakeServer.instances[0].closed + + def test_stream_failure_never_stops_the_lan_poke( + self, + tmp_path: Path, + stream: MagicMock, + caplog: pytest.LogCaptureFixture, + ) -> None: + stream.return_value.start.side_effect = RuntimeError("no network") + with caplog.at_level(logging.ERROR): + code = poke_server.main(["--key-file", str(_key_file(tmp_path))]) + assert code == 0 + assert "Could not start the Firebase session stream" in caplog.text + assert FakeServer.instances[0].closed + + def test_server_is_closed_even_when_serving_raises( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + def explode(_self: FakeServer) -> None: + raise KeyboardInterrupt + + monkeypatch.setattr(FakeServer, "serve_forever", explode) + with pytest.raises(KeyboardInterrupt): + poke_server.main(["--key-file", str(_key_file(tmp_path))]) + assert FakeServer.instances[0].closed diff --git a/screen_locker/tests/test_poke_sandbox.py b/screen_locker/tests/test_poke_sandbox.py new file mode 100644 index 0000000..346a21b --- /dev/null +++ b/screen_locker/tests/test_poke_sandbox.py @@ -0,0 +1,87 @@ +"""A sandbox poke is verified and answered, but writes and credits nothing.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING +from unittest.mock import MagicMock + +import pytest + +from screen_locker import _poke_credit, _poke_server +from screen_locker._poke_credit import PokeLocker +from screen_locker._poke_wire import PokeRequest + +if TYPE_CHECKING: + from pathlib import Path + +_LOCKER_STEPS = ( + "start_the_day", + "credit_written", + "_write_shutdown_config", + "_adjust_shutdown_time_later", + "_apply_credit_for_written_entry", +) + + +def _request(*, sandbox: bool) -> PokeRequest: + return PokeRequest( + sent_at_ms=1, + nonce="n", + sandbox=sandbox, + record_id="rec-1", + payload={"date": "2026-10-09"}, + ) + + +@pytest.fixture +def pc(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> dict[str, MagicMock]: + """A real PokeLocker on a tmp log with every state-changing step spied.""" + + def headless(locker_cls: type[PokeLocker]) -> PokeLocker: + locker = object.__new__(locker_cls) + locker.log_file = tmp_path / "log.json" + return locker + + spies = {name: MagicMock() for name in (*_LOCKER_STEPS, "ingest", "credit")} + for name in _LOCKER_STEPS: + monkeypatch.setattr(PokeLocker, name, spies[name]) + monkeypatch.setattr(_poke_credit, "_headless_locker", headless) + monkeypatch.setattr(_poke_credit, "ingest_session_records", spies["ingest"]) + monkeypatch.setattr(_poke_credit, "credit_session", spies["credit"]) + monkeypatch.setattr(_poke_credit, "current_shutdown", lambda: "20:30") + monkeypatch.setattr(_poke_credit, "current_gaming_minutes", lambda _p: 75) + return spies + + +def test_sandbox_replies_ok_with_the_snapshot_and_nothing_credited( + pc: dict[str, MagicMock], +) -> None: + reply = _poke_server._process(_request(sandbox=True)) + assert reply["ok"] is True + assert reply["sandbox"] is True + assert reply["credited"] is False + assert reply["duplicate"] is False + assert reply["shutdown"] == "20:30" + assert reply["gaming_budget_minutes"] == 75 + assert "nothing written or credited" in str(reply["reason"]) + + +def test_sandbox_never_logs_credits_or_touches_the_shutdown_config( + pc: dict[str, MagicMock], tmp_path: Path +) -> None: + before = sorted(p.name for p in tmp_path.iterdir()) + _poke_server._process(_request(sandbox=True)) + for name, spy in pc.items(): + assert not spy.called, name + assert sorted(p.name for p in tmp_path.iterdir()) == before + assert not (tmp_path / "log.json").exists() + + +def test_a_real_poke_does_reach_the_credit_step(pc: dict[str, MagicMock]) -> None: + """Guards the guard: the spies above are wired to the path a real poke takes.""" + pc["credit"].return_value = _poke_credit.PokeOutcome( + ok=True, credited=True, duplicate=False, reason="workout credited" + ) + reply = _poke_server._process(_request(sandbox=False)) + assert reply["credited"] is True + pc["credit"].assert_called_once_with("rec-1", {"date": "2026-10-09"}) diff --git a/screen_locker/tests/test_poke_server.py b/screen_locker/tests/test_poke_server.py new file mode 100644 index 0000000..06da4c1 --- /dev/null +++ b/screen_locker/tests/test_poke_server.py @@ -0,0 +1,248 @@ +"""The poke HTTP front end, driven over real loopback HTTP on an ephemeral port.""" + +from __future__ import annotations + +import hashlib +import hmac +from http.client import HTTPConnection +import json +import socket +import threading +import time +from typing import TYPE_CHECKING, Any +from unittest.mock import MagicMock + +import pytest + +from screen_locker import _poke_credit, _poke_server +from screen_locker._poke_server import MAX_BODY_BYTES, POKE_PATH, PokeServer + +if TYPE_CHECKING: + from collections.abc import Iterator + from pathlib import Path + +# conftest stubs socket.create_connection process-wide (see test_web_server). +_REAL_CREATE_CONNECTION = socket.create_connection +_KEY = bytes(range(32)) + + +def _now_ms() -> int: + return time.time_ns() // 1_000_000 + + +def _body(nonce: str = "n-1", **overrides: object) -> bytes: + doc: dict[str, Any] = { + "v": 1, + "sent_at_ms": _now_ms(), + "nonce": nonce, + "sandbox": False, + "record_id": "rec-1", + "payload": {"date": "2026-10-09", "note": "zażółć"}, + } + doc.update(overrides) + return json.dumps(doc, ensure_ascii=False).encode("utf-8") + + +def _sign(body: bytes) -> str: + return hmac.new(_KEY, body, hashlib.sha256).hexdigest() + + +@pytest.fixture +def port(monkeypatch: pytest.MonkeyPatch) -> Iterator[int]: + """A running PokeServer on 127.0.0.1:0; shut down afterwards.""" + monkeypatch.setattr(socket, "create_connection", _REAL_CREATE_CONNECTION) + server = PokeServer(("127.0.0.1", 0), _KEY) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield server.server_address[1] + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + + +@pytest.fixture +def credit(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> MagicMock: + """Stub the PC side: tmp locker, fixed snapshot, a spy for the credit.""" + locker = MagicMock() + locker.log_file = tmp_path / "log.json" + monkeypatch.setattr(_poke_credit, "fresh_locker", lambda: locker) + monkeypatch.setattr(_poke_credit, "current_shutdown", lambda: "20:30") + monkeypatch.setattr(_poke_credit, "current_gaming_minutes", lambda _p: 75) + spy = MagicMock( + return_value=_poke_credit.PokeOutcome( + ok=True, credited=True, duplicate=False, reason="workout credited" + ) + ) + monkeypatch.setattr(_poke_credit, "credit_session", spy) + return spy + + +def _post( + port: int, + body: bytes, + signature: str | None, + path: str = POKE_PATH, +) -> tuple[int, dict[str, Any]]: + conn = HTTPConnection("127.0.0.1", port, timeout=10) + headers = {"Content-Type": "application/json"} + if signature is not None: + headers["X-Poke-Signature"] = signature + conn.request("POST", path, body=body, headers=headers) + resp = conn.getresponse() + data = json.loads(resp.read().decode("utf-8")) + conn.close() + return resp.status, data + + +def _raw(port: int, lines: list[str], send: bytes = b"") -> tuple[int, dict[str, Any]]: + """A hand-written POST (so headers http.client would add can be omitted).""" + conn = HTTPConnection("127.0.0.1", port, timeout=10) + conn.putrequest("POST", POKE_PATH) + for line in lines: + name, value = line.split(": ", 1) + conn.putheader(name, value) + conn.endheaders() + if send and conn.sock is not None: + conn.sock.sendall(send) + if conn.sock is not None: + conn.sock.shutdown(socket.SHUT_WR) + resp = conn.getresponse() + data = json.loads(resp.read().decode("utf-8")) + conn.close() + return resp.status, data + + +class TestCredit: + def test_valid_poke_credits_and_answers_the_contract_shape( + self, port: int, credit: MagicMock + ) -> None: + body = _body() + status, reply = _post(port, body, _sign(body)) + assert status == 200 + assert reply["ok"] is True + assert reply["credited"] is True + assert reply["duplicate"] is False + assert reply["sandbox"] is False + assert reply["shutdown"] == "20:30" + assert reply["gaming_budget_minutes"] == 75 + assert isinstance(reply["pc_ms"], int) + credit.assert_called_once_with( + "rec-1", {"date": "2026-10-09", "note": "zażółć"} + ) + + def test_replayed_nonce_is_409_and_credits_once( + self, port: int, credit: MagicMock + ) -> None: + body = _body() + assert _post(port, body, _sign(body))[0] == 200 + status, reply = _post(port, body, _sign(body)) + assert status == 409 + assert reply["ok"] is False + assert reply["shutdown"] is None + assert credit.call_count == 1 + + def test_stale_timestamp_is_401(self, port: int, credit: MagicMock) -> None: + body = _body(sent_at_ms=_now_ms() - 600_000) + status, reply = _post(port, body, _sign(body)) + assert status == 401 + assert "stale" in reply["reason"] + credit.assert_not_called() + + def test_bad_or_missing_signature_is_401( + self, port: int, credit: MagicMock + ) -> None: + body = _body() + assert _post(port, body, "00" * 32)[0] == 401 + assert _post(port, body, None)[0] == 401 + credit.assert_not_called() + + def test_schema_violation_is_400(self, port: int, credit: MagicMock) -> None: + body = _body(v=2) + status, reply = _post(port, body, _sign(body)) + assert status == 400 + assert "protocol version" in reply["reason"] + credit.assert_not_called() + + def test_internal_error_is_500_and_says_the_sync_will_cover_it( + self, + port: int, + credit: MagicMock, + caplog: pytest.LogCaptureFixture, + ) -> None: + credit.side_effect = RuntimeError("boom") + body = _body() + status, reply = _post(port, body, _sign(body)) + assert status == 500 + assert reply["ok"] is False + assert "regular sync will still credit" in reply["reason"] + assert "crashed" in caplog.text + + +class TestSandbox: + def test_sandbox_poke_over_http_is_answered_but_not_credited( + self, port: int, credit: MagicMock + ) -> None: + body = _body(sandbox=True) + status, reply = _post(port, body, _sign(body)) + assert (status, reply["ok"], reply["sandbox"]) == (200, True, True) + assert reply["credited"] is False + credit.assert_not_called() + + def test_sandbox_with_a_bad_signature_is_still_refused( + self, port: int, credit: MagicMock + ) -> None: + status, reply = _post(port, _body(sandbox=True), "00" * 32) + assert status == 401 + assert reply["sandbox"] is False + + +class TestRouting: + def test_unknown_path_is_404(self, port: int, credit: MagicMock) -> None: + body = _body() + status, reply = _post(port, body, _sign(body), path="/v2/other") + assert status == 404 + assert "unknown path" in reply["reason"] + + def test_get_is_405(self, port: int) -> None: + conn = HTTPConnection("127.0.0.1", port, timeout=10) + conn.request("GET", POKE_PATH) + resp = conn.getresponse() + data = json.loads(resp.read()) + conn.close() + assert resp.status == 405 + assert "only POST" in data["reason"] + + def test_missing_content_length_is_411(self, port: int) -> None: + status, _ = _raw(port, ["Host: x"]) + assert status == 411 + + @pytest.mark.parametrize("value", ["abc", "-5", ""]) + def test_malformed_content_length_is_400(self, port: int, value: str) -> None: + status, reply = _raw(port, [f"Content-Length: {value}"]) + assert status == 400 + assert "bad Content-Length" in reply["reason"] + + def test_oversized_body_is_413_without_reading_it(self, port: int) -> None: + status, reply = _raw(port, [f"Content-Length: {MAX_BODY_BYTES + 1}"]) + assert status == 413 + assert str(MAX_BODY_BYTES) in reply["reason"] + + def test_body_at_the_cap_is_read_and_judged_on_its_signature( + self, port: int + ) -> None: + body = b"x" * MAX_BODY_BYTES + assert _post(port, body, _sign(body))[0] == 400 + + def test_truncated_body_is_400(self, port: int) -> None: + status, reply = _raw(port, ["Content-Length: 100"], send=b"0123456789") + assert status == 400 + assert "10 of 100 bytes" in reply["reason"] + + +def test_reply_for_a_refusal_leaves_the_snapshot_null() -> None: + reply = _poke_server._refused("nope") + assert reply["ok"] is False + assert (reply["shutdown"], reply["gaming_budget_minutes"]) == (None, None) + assert reply["reason"] == "nope" diff --git a/screen_locker/tests/test_poke_wire.py b/screen_locker/tests/test_poke_wire.py new file mode 100644 index 0000000..5280e7b --- /dev/null +++ b/screen_locker/tests/test_poke_wire.py @@ -0,0 +1,204 @@ +"""The poke wire checks: signature, freshness, replay and schema, in order.""" + +from __future__ import annotations + +import hashlib +import hmac +from http import HTTPStatus +import json +from typing import Any + +import pytest + +from screen_locker._poke_wire import ( + REPLAY_WINDOW_MS, + NonceCache, + PokeRequest, + Rejection, + check_request, + parse_key, + verify_signature, +) + +_KEY = bytes(range(32)) +_NOW = 1_800_000_000_000 + + +def _doc(**overrides: object) -> dict[str, Any]: + doc: dict[str, Any] = { + "v": 1, + "sent_at_ms": _NOW, + "nonce": "n-1", + "sandbox": False, + "record_id": "rec-1", + "payload": {"date": "2026-10-09"}, + } + doc.update(overrides) + return doc + + +def _sign(body: bytes, key: bytes = _KEY) -> str: + return hmac.new(key, body, hashlib.sha256).hexdigest() + + +def _check( + doc: dict[str, Any] | bytes, + *, + nonces: NonceCache | None = None, + now_ms: int = _NOW, +) -> PokeRequest | Rejection: + body = doc if isinstance(doc, bytes) else json.dumps(doc).encode("utf-8") + return check_request( + _KEY, body, _sign(body), NonceCache() if nonces is None else nonces, now_ms + ) + + +def _rejected(result: PokeRequest | Rejection, status: HTTPStatus) -> Rejection: + assert isinstance(result, Rejection) + assert result.status is status + return result + + +class TestParseKey: + def test_decodes_64_hex_chars_with_surrounding_whitespace(self) -> None: + assert parse_key(f" {_KEY.hex()}\n") == _KEY + + def test_wrong_length_says_how_long_it_was(self) -> None: + with pytest.raises(ValueError, match="got 4"): + parse_key("abcd") + + def test_non_hex_of_the_right_length_is_refused(self) -> None: + with pytest.raises(ValueError, match="not valid hex"): + parse_key("zz" * 32) + + +class TestVerifySignature: + def test_matching_mac_verifies(self) -> None: + assert verify_signature(_KEY, b"body", _sign(b"body")) + + def test_uppercase_and_padded_header_is_accepted(self) -> None: + assert verify_signature(_KEY, b"body", f" {_sign(b'body').upper()} ") + + @pytest.mark.parametrize("header", [None, ""]) + def test_missing_header_is_false(self, header: str | None) -> None: + assert not verify_signature(_KEY, b"body", header) + + def test_wrong_key_is_false(self) -> None: + assert not verify_signature(_KEY, b"body", _sign(b"body", b"k" * 32)) + + def test_non_ascii_header_is_false_not_an_exception(self) -> None: + assert not verify_signature(_KEY, b"body", "é" * 64) + assert not verify_signature(_KEY, b"body", "☃" * 64) + + +class TestNonceCache: + def test_second_use_inside_window_is_refused(self) -> None: + cache = NonceCache() + assert cache.check_and_add("a", _NOW, _NOW) + assert not cache.check_and_add("a", _NOW, _NOW + 1) + assert len(cache) == 1 + + def test_pruned_by_sent_at_after_the_window(self) -> None: + cache = NonceCache() + cache.check_and_add("a", _NOW, _NOW) + assert cache.check_and_add("b", _NOW, _NOW + REPLAY_WINDOW_MS) + assert len(cache) == 2 + assert cache.check_and_add("a", _NOW, _NOW + REPLAY_WINDOW_MS + 1) + assert len(cache) == 1 # both old entries pruned, only the new "a" stays + + def test_future_stamped_nonce_is_remembered_until_sent_at_plus_window( + self, + ) -> None: + cache = NonceCache() + future = _NOW + 100_000 + cache.check_and_add("a", future, _NOW) + assert not cache.check_and_add("a", future, _NOW + 150_000) + + +class TestCheckRequest: + def test_valid_request_is_parsed(self) -> None: + result = _check(_doc(sandbox=True)) + assert isinstance(result, PokeRequest) + assert (result.nonce, result.record_id) == ("n-1", "rec-1") + assert result.sandbox is True + assert result.payload == {"date": "2026-10-09"} + + def test_bad_signature_is_401_and_parses_nothing(self) -> None: + body = b"not json at all" + result = check_request(_KEY, body, "00" * 32, NonceCache(), _NOW) + assert "signature" in _rejected(result, HTTPStatus.UNAUTHORIZED).reason + + def test_bad_signature_beats_every_later_check(self) -> None: + body = json.dumps(_doc(sent_at_ms=0)).encode() + result = check_request(_KEY, body, None, NonceCache(), _NOW) + assert "signature" in _rejected(result, HTTPStatus.UNAUTHORIZED).reason + + def test_signed_non_utf8_is_400(self) -> None: + reason = _rejected(_check(b"\xff\xfe"), HTTPStatus.BAD_REQUEST).reason + assert "UTF-8 JSON" in reason + + def test_signed_invalid_json_is_400(self) -> None: + _rejected(_check(b"{nope"), HTTPStatus.BAD_REQUEST) + + def test_signed_json_array_is_400(self) -> None: + reason = _rejected(_check(b"[1]"), HTTPStatus.BAD_REQUEST).reason + assert "not a JSON object" in reason + + @pytest.mark.parametrize("value", [None, "1", True, 1.5]) + def test_sent_at_must_be_a_real_int(self, value: object) -> None: + reason = _rejected( + _check(_doc(sent_at_ms=value)), HTTPStatus.BAD_REQUEST + ).reason + assert "sent_at_ms" in reason + + @pytest.mark.parametrize("skew", [REPLAY_WINDOW_MS + 1, -REPLAY_WINDOW_MS - 1]) + def test_stale_or_future_stamp_is_401_naming_the_clock(self, skew: int) -> None: + result = _check(_doc(sent_at_ms=_NOW - skew)) + assert "stale" in _rejected(result, HTTPStatus.UNAUTHORIZED).reason + + @pytest.mark.parametrize("skew", [REPLAY_WINDOW_MS, -REPLAY_WINDOW_MS]) + def test_exactly_the_window_edge_is_still_fresh(self, skew: int) -> None: + assert isinstance(_check(_doc(sent_at_ms=_NOW - skew)), PokeRequest) + + @pytest.mark.parametrize("nonce", [None, "", 7]) + def test_nonce_must_be_a_non_empty_string(self, nonce: object) -> None: + _rejected(_check(_doc(nonce=nonce)), HTTPStatus.BAD_REQUEST) + + def test_replayed_nonce_is_409(self) -> None: + nonces = NonceCache() + assert isinstance(_check(_doc(), nonces=nonces), PokeRequest) + result = _check(_doc(), nonces=nonces) + assert "already used" in _rejected(result, HTTPStatus.CONFLICT).reason + + def test_staleness_is_checked_before_replay(self) -> None: + nonces = NonceCache() + _check(_doc(), nonces=nonces) + stale = _doc(sent_at_ms=_NOW - REPLAY_WINDOW_MS - 1) + _rejected(_check(stale, nonces=nonces), HTTPStatus.UNAUTHORIZED) + + @pytest.mark.parametrize("version", [None, 2, "1", True]) + def test_unsupported_version_is_400(self, version: object) -> None: + reason = _rejected(_check(_doc(v=version)), HTTPStatus.BAD_REQUEST).reason + assert "protocol version" in reason + + @pytest.mark.parametrize("sandbox", [None, "false", 0]) + def test_sandbox_must_be_a_bool(self, sandbox: object) -> None: + reason = _rejected(_check(_doc(sandbox=sandbox)), HTTPStatus.BAD_REQUEST).reason + assert "'sandbox'" in reason + + @pytest.mark.parametrize("record_id", [None, "", 5]) + def test_record_id_must_be_a_non_empty_string(self, record_id: object) -> None: + reason = _rejected( + _check(_doc(record_id=record_id)), HTTPStatus.BAD_REQUEST + ).reason + assert "'record_id'" in reason + + @pytest.mark.parametrize("payload", [None, [], "x"]) + def test_payload_must_be_an_object(self, payload: object) -> None: + reason = _rejected(_check(_doc(payload=payload)), HTTPStatus.BAD_REQUEST).reason + assert "'payload'" in reason + + def test_schema_failure_still_burns_the_nonce(self) -> None: + nonces = NonceCache() + _rejected(_check(_doc(v=2), nonces=nonces), HTTPStatus.BAD_REQUEST) + _rejected(_check(_doc(), nonces=nonces), HTTPStatus.CONFLICT) diff --git a/screen_locker/tests/test_rtdb_stream.py b/screen_locker/tests/test_rtdb_stream.py new file mode 100644 index 0000000..40d596a --- /dev/null +++ b/screen_locker/tests/test_rtdb_stream.py @@ -0,0 +1,168 @@ +"""Tests for _rtdb_stream: the SSE line parser and the put/patch normaliser.""" + +from __future__ import annotations + +import json + +import pytest + +from screen_locker._rtdb_stream import SseEvent, device_updates, parse_sse + +_LEAF = "log~2Ejson" + + +def _event(name: str, path: str, data: object) -> SseEvent: + """A ``put``/``patch`` event carrying ``{path, data}`` as RTDB sends it.""" + return SseEvent(name, json.dumps({"path": path, "data": data})) + + +class TestParseSse: + """Server-Sent Events framing: a blank line ends each event.""" + + def test_named_event_with_data(self) -> None: + """``event:`` + ``data:`` + blank line yields one event.""" + lines = ["event: put", 'data: {"path":"/"}', ""] + assert list(parse_sse(lines)) == [SseEvent("put", '{"path":"/"}')] + + def test_multiple_data_lines_are_joined_with_newlines(self) -> None: + """Per the SSE spec, repeated ``data:`` lines join with ``\\n``.""" + events = list(parse_sse(["event: patch", "data: a", "data: b", ""])) + assert events == [SseEvent("patch", "a\nb")] + + def test_no_space_after_colon_is_accepted(self) -> None: + """``event:put`` is as valid as ``event: put``.""" + assert list(parse_sse(["event:keep-alive", "data:null", ""])) == [ + SseEvent("keep-alive", "null") + ] + + def test_data_without_event_is_a_message(self) -> None: + """An unnamed event defaults to ``message``.""" + assert list(parse_sse(["data: hi", ""])) == [SseEvent("message", "hi")] + + def test_event_without_data_still_yields(self) -> None: + """A name alone is an event (``cancel`` may carry no data).""" + assert list(parse_sse(["event: cancel", ""])) == [SseEvent("cancel", "")] + + def test_stray_blank_lines_and_comments_yield_nothing(self) -> None: + """Blank separators and ``:`` comments / unknown fields are ignored.""" + assert list(parse_sse(["", "", ": comment", "id: 7", "retry: 5", ""])) == [] + + def test_back_to_back_events_do_not_leak_state(self) -> None: + """Each event starts with a clean name and data.""" + lines = ["event: put", "data: 1", "", "data: 2", ""] + assert list(parse_sse(lines)) == [ + SseEvent("put", "1"), + SseEvent("message", "2"), + ] + + def test_unterminated_event_is_not_emitted(self) -> None: + """Without the closing blank line the event is incomplete: dropped.""" + assert list(parse_sse(["event: put", "data: 1"])) == [] + + +class TestDeviceUpdatesPut: + """``put`` events: the data replaces whatever sits at the path.""" + + def test_root_put_is_a_full_snapshot(self) -> None: + """``put /`` carries every device; the caller must REPLACE its cache.""" + data = {"phone": {_LEAF: "TEXT-P"}, "pc": {_LEAF: "TEXT-C"}} + full, updates = device_updates(_event("put", "/", data)) + assert full is True + assert updates == {"phone": "TEXT-P", "pc": "TEXT-C"} + + def test_root_put_with_null_is_an_empty_snapshot(self) -> None: + """An empty devices node arrives as ``data: null``.""" + assert device_updates(_event("put", "/", None)) == (True, {}) + + def test_root_put_decodes_plain_and_escaped_leaf_keys(self) -> None: + """The leaf may arrive as ``log.json`` or the escaped ``log~2Ejson``.""" + data = {"a": {"log.json": "X"}, "b": {_LEAF: "Y"}} + assert device_updates(_event("put", "/", data))[1] == {"a": "X", "b": "Y"} + + def test_root_put_device_without_log_is_none(self) -> None: + """A device node that is not a dict, lacks the leaf or holds a non-string.""" + data = {"a": "scalar", "b": {"other": "x"}, "c": {_LEAF: 5}, "d": None} + updates = device_updates(_event("put", "/", data))[1] + assert updates == {"a": None, "b": None, "c": None, "d": None} + + def test_root_put_with_non_object_is_a_type_error(self) -> None: + """A devices node that is not an object is a protocol violation.""" + with pytest.raises(TypeError, match="not an object"): + device_updates(_event("put", "/", ["x"])) + + def test_device_level_put(self) -> None: + """``put /phone`` replaces one device's node.""" + full, updates = device_updates(_event("put", "/phone", {_LEAF: "T"})) + assert (full, updates) == (False, {"phone": "T"}) + + def test_device_level_delete(self) -> None: + """``put /phone`` with null removes the log.""" + assert device_updates(_event("put", "/phone", None)) == (False, {"phone": None}) + + def test_leaf_level_put_and_delete(self) -> None: + """``put /phone/log~2Ejson`` sets or clears the text directly.""" + path = f"/phone/{_LEAF}" + assert device_updates(_event("put", path, "T")) == (False, {"phone": "T"}) + assert device_updates(_event("put", path, None)) == (False, {"phone": None}) + + def test_leaf_level_put_of_non_string_is_none(self) -> None: + """A non-text leaf is not a log: the device has none.""" + assert device_updates(_event("put", f"/phone/{_LEAF}", 7)) == ( + False, + {"phone": None}, + ) + + def test_put_below_the_log_leaf_is_irrelevant(self) -> None: + """Anything under a device that is not its log changes nothing.""" + assert device_updates(_event("put", "/phone/other", "x")) == (False, {}) + + def test_device_ids_are_decoded(self) -> None: + """Escaped device names (``~2E``) come back as the real id.""" + assert device_updates(_event("put", "/pc~2E1", {_LEAF: "T"}))[1] == { + "pc.1": "T" + } + + +class TestDeviceUpdatesPatch: + """``patch`` events: several children, keys may be multi-segment paths.""" + + def test_patch_with_leaf_paths(self) -> None: + """Each key is resolved relative to the event path.""" + data = {f"phone/{_LEAF}": "T1", f"pc/{_LEAF}": None} + full, updates = device_updates(_event("patch", "/", data)) + assert full is False + assert updates == {"phone": "T1", "pc": None} + + def test_patch_with_device_keys_and_base_path(self) -> None: + """A patch at ``/phone`` with a bare leaf key targets that device.""" + data = {_LEAF: "T"} + assert device_updates(_event("patch", "/phone", data)) == ( + False, + {"phone": "T"}, + ) + + def test_patch_at_root_is_never_a_snapshot(self) -> None: + """Even at ``/``, a patch merges; it must not replace the cache.""" + assert device_updates(_event("patch", "/", {"phone": {_LEAF: "T"}}))[0] is False + + def test_patch_with_non_object_data_is_a_type_error(self) -> None: + """``patch`` data must be an object.""" + with pytest.raises(TypeError, match="patch data is str"): + device_updates(_event("patch", "/", "oops")) + + +class TestDeviceUpdatesMalformed: + """Garbage must raise, so the caller can warn instead of guessing.""" + + def test_non_json_data_is_a_value_error(self) -> None: + """Unparsable data raises ``ValueError``.""" + with pytest.raises(json.JSONDecodeError): + device_updates(SseEvent("put", "{not json")) + + @pytest.mark.parametrize( + "data", ["[]", "null", '"x"', '{"data": 1}', '{"path": 3, "data": 1}'] + ) + def test_missing_or_bad_path_is_a_type_error(self, data: str) -> None: + """Anything but ``{path: str, data: ...}`` is rejected.""" + with pytest.raises(TypeError, match="without a path"): + device_updates(SseEvent("put", data)) diff --git a/screen_locker/tests/test_session_stream.py b/screen_locker/tests/test_session_stream.py new file mode 100644 index 0000000..248dad7 --- /dev/null +++ b/screen_locker/tests/test_session_stream.py @@ -0,0 +1,207 @@ +"""Tests for _session_stream: byte metering, event dispatch and token renewal.""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +import json +import logging +import threading +from types import SimpleNamespace +from typing import TYPE_CHECKING + +from crdt_sync import Hlc, Record +import pytest +import requests +import urllib3 + +from screen_locker import _session_stream, _stream_cache +from screen_locker._poke_credit import BatchOutcome +from screen_locker._session_stream import SessionStream +from screen_locker._stream_auth import StreamError + +if TYPE_CHECKING: + from collections.abc import Iterable, Mapping + +_FUTURE = datetime(2999, 1, 1, tzinfo=UTC) +_PAST = datetime.now(UTC) - timedelta(hours=1) +_LEAF = "log~2Ejson" +KEEPALIVE = ("keep-alive", "null") + + +class FakeRaw: + """``response.raw``: yields queued chunks, raises queued exceptions.""" + + def __init__(self, chunks: Iterable[bytes | Exception]) -> None: + self.chunks = list(chunks) + self.closed = False + + def read1(self, _size: int) -> bytes: + """The next chunk (``b""`` at EOF), or raise the queued error.""" + if not self.chunks: + return b"" + item = self.chunks.pop(0) + if isinstance(item, Exception): + raise item + return item + + def close(self) -> None: + """Mark the connection released.""" + self.closed = True + + +def fake_response( + chunks: Iterable[bytes | Exception] = (), + status_code: int = 200, + text: str = "", +) -> requests.Response: + """A real ``Response`` whose body is the queued chunks (no sockets).""" + response = requests.Response() + response.status_code = status_code + response.encoding = "utf-8" + response._content = text.encode() + response.raw = FakeRaw(chunks) + return response + + +def sse(*events: tuple[str, str]) -> bytes: + """Encode ``(name, data)`` pairs as an SSE byte stream.""" + return b"".join(f"event: {n}\ndata: {d}\n\n".encode() for n, d in events) + + +def make_stream(monkeypatch: pytest.MonkeyPatch) -> SessionStream: + """A stream with a fake device identity (the real one mints a file).""" + identity = SimpleNamespace(device_id="pc-new", legacy_id="pc-old") + monkeypatch.setattr(_session_stream, "device_identity", lambda: identity) + return SessionStream(threading.Lock()) + + +@pytest.fixture +def stream(monkeypatch: pytest.MonkeyPatch) -> SessionStream: + """A ``SessionStream`` that never reads or writes the real device id.""" + return make_stream(monkeypatch) + + +def _put(data: object, path: str = "/") -> tuple[str, str]: + return "put", json.dumps({"path": path, "data": data}) + + +class TestConsume: + """Dispatching events until the server or the renew deadline ends it.""" + + def test_snapshot_and_patch_are_credited( + self, stream: SessionStream, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Real cache + fake credit: snapshot then patch each credit once.""" + batches: list[list[str]] = [] + + def credit(records: Iterable[tuple[str, Mapping[str, object]]]) -> BatchOutcome: + ids = [rid for rid, _p in records] + batches.append(ids) + return BatchOutcome(tuple(ids), len(ids)) + + monkeypatch.setattr(_stream_cache._poke_credit, "credit_sessions", credit) + log = _session_log("s1") + log2 = _session_log("s2") + body = sse( + _put({"phone": {_LEAF: log}}), + ("patch", json.dumps({"path": "/", "data": {f"phone/{_LEAF}": log2}})), + ("auth_revoked", '"expired"'), + ) + stream._consume(fake_response([body]), _FUTURE) + assert batches == [["s1"], ["s2"]] + assert stream._got_snapshot is True + + def test_keepalives_are_logged_only_a_few_times( + self, + stream: SessionStream, + caplog: pytest.LogCaptureFixture, + ) -> None: + """Keep-alive events 1..3 are logged; later ones are silent.""" + body = sse(*[("keep-alive", "null")] * 5, ("auth_revoked", "x")) + with caplog.at_level(logging.INFO): + stream._consume(fake_response([body]), _FUTURE) + seen = [ + r.getMessage() for r in caplog.records if "keep-alive #" in r.getMessage() + ] + assert seen == [f"Session stream keep-alive #{n}" for n in (1, 2, 3)] + + def test_auth_revoked_returns_with_a_warning( + self, stream: SessionStream, caplog: pytest.LogCaptureFixture + ) -> None: + """A revoked token ends the connection on purpose, loudly.""" + with caplog.at_level(logging.WARNING): + stream._consume( + fake_response([sse(("auth_revoked", "stale"))]), + _FUTURE, + ) + assert any( + "token revoked/expired (stale)" in r.getMessage() + and r.levelno == logging.WARNING + for r in caplog.records + ) + + def test_cancel_is_a_stream_error(self, stream: SessionStream) -> None: + """``cancel`` means the rules deny us: back off.""" + response = fake_response([sse(("cancel", "permission_denied"))]) + with pytest.raises( + StreamError, match=r"cancelled the stream \(permission_denied\)" + ): + stream._consume(response, _FUTURE) + + def test_unknown_event_is_warned_about_and_ignored( + self, stream: SessionStream, caplog: pytest.LogCaptureFixture + ) -> None: + """Unexpected events are logged at WARNING and the stream carries on.""" + response = fake_response([sse(("mystery", "?"), ("auth_revoked", "x"))]) + with caplog.at_level(logging.WARNING): + stream._consume(response, _FUTURE) + assert any("ignored event 'mystery'" in r.getMessage() for r in caplog.records) + + def test_token_is_renewed_before_expiry( + self, stream: SessionStream, caplog: pytest.LogCaptureFixture + ) -> None: + """Once the renew deadline passes, the next event ends it cleanly.""" + response = fake_response([sse(("keep-alive", "null"), ("keep-alive", "null"))]) + with caplog.at_level(logging.INFO): + stream._consume(response, _PAST) + assert any("renewing its ID token" in r.getMessage() for r in caplog.records) + assert response.raw.chunks == [] + keepalives = [r for r in caplog.records if "keep-alive #" in r.getMessage()] + assert len(keepalives) == 1 # stopped after the first event + + def test_server_closing_unasked_is_a_stream_error( + self, stream: SessionStream + ) -> None: + """EOF without auth_revoked/renewal is a failure, with a byte count.""" + response = fake_response([sse(("keep-alive", "null"))]) + with pytest.raises( + StreamError, + match=f"server closed the stream after {len(sse(KEEPALIVE))} bytes", + ): + stream._consume(response, _FUTURE) + + @pytest.mark.parametrize( + "error", + [ + requests.exceptions.ReadTimeout("read timed out"), + requests.exceptions.ChunkedEncodingError("cut"), + urllib3.exceptions.ProtocolError("connection broken"), + TimeoutError("socket timeout"), + ], + ) + def test_read_errors_and_timeouts_become_stream_errors( + self, stream: SessionStream, error: Exception + ) -> None: + """A silent socket (read timeout) or broken read ends in StreamError.""" + response = fake_response([sse(("keep-alive", "null")), error]) + with pytest.raises( + StreamError, match=f"stream broke after {len(sse(KEEPALIVE))} bytes" + ) as ei: + stream._consume(response, _FUTURE) + assert ei.value.__cause__ is error + + +def _session_log(rid: str) -> str: + """A device-log blob with one completed-session record.""" + record = Record(rid, {"payload": ({"exercises": []}, Hlc(1, 0, "n"))}) + return json.dumps({rid: record.to_dict()}) diff --git a/screen_locker/tests/test_session_stream_connect.py b/screen_locker/tests/test_session_stream_connect.py new file mode 100644 index 0000000..01ad7eb --- /dev/null +++ b/screen_locker/tests/test_session_stream_connect.py @@ -0,0 +1,213 @@ +"""Tests for _session_stream: connecting, backoff, reconnect and thread control.""" + +from __future__ import annotations + +from datetime import UTC, datetime +import json +import logging + +import pytest +import requests + +from screen_locker import _session_stream +from screen_locker._rtdb_stream import SseEvent +from screen_locker._session_stream import SessionStream, _ByteMeter +from screen_locker._stream_auth import StreamAuth, StreamError +from screen_locker.tests.test_session_stream import fake_response, make_stream + +_RENEW = datetime(2999, 1, 1, tzinfo=UTC) +_LONG = 400.0 # longer than the 300 s minimum lifetime + + +@pytest.fixture +def stream(monkeypatch: pytest.MonkeyPatch) -> SessionStream: + """A ``SessionStream`` that never reads or writes the real device id.""" + return make_stream(monkeypatch) + + +class TestByteMeter: + """Line splitting without waiting for a full buffer.""" + + def test_lines_split_across_chunks_and_counted(self) -> None: + """A line broken over two reads is rejoined; CRLF is stripped.""" + meter = _ByteMeter() + chunks = [b"ev", b"ent: a\r\ndata: b\n\nx", b"y\n"] + lines = list(meter.lines(fake_response(chunks))) + assert lines == ["event: a", "data: b", "", "xy"] + assert meter.total == sum(map(len, chunks)) + + def test_trailing_partial_line_is_dropped_at_eof(self) -> None: + """Bytes after the last newline never form a line.""" + meter = _ByteMeter() + assert list(meter.lines(fake_response([b"a\nb"]))) == ["a"] + + +class TestDispatch: + """One put/patch handed to the cache.""" + + def test_undecodable_event_is_warned_and_does_not_end_the_stream( + self, stream: SessionStream, caplog: pytest.LogCaptureFixture + ) -> None: + """Malformed data logs a WARNING naming the fallback, never raises.""" + with caplog.at_level(logging.WARNING): + stream._dispatch(SseEvent("put", "{garbage"), 10) + stream._dispatch(SseEvent("patch", "[]"), 10) + messages = [r.getMessage() for r in caplog.records] + assert len(messages) == 2 + assert all("undecodable" in m and "15-min workout-sync" in m for m in messages) + assert stream._got_snapshot is False + + def test_snapshot_flag_only_set_by_snapshots( + self, stream: SessionStream, caplog: pytest.LogCaptureFixture + ) -> None: + """A patch does not mark the connection as having a snapshot.""" + patch_event = SseEvent("patch", json.dumps({"path": "/", "data": {}})) + with caplog.at_level(logging.INFO): + stream._dispatch(patch_event, 7) + assert stream._got_snapshot is False + assert any( + "Session stream patch (7 bytes" in r.getMessage() for r in caplog.records + ) + snapshot = SseEvent("put", json.dumps({"path": "/", "data": None})) + with caplog.at_level(logging.INFO): + stream._dispatch(snapshot, 9) + assert stream._got_snapshot is True + assert any( + "Session stream snapshot (9 bytes" in r.getMessage() for r in caplog.records + ) + + +class _FakeHttp: + """``requests.Session`` stand-in recording the one GET it is given.""" + + def __init__(self, outcome: requests.Response | Exception) -> None: + self.outcome = outcome + self.calls: list[tuple[str, dict[str, object]]] = [] + + def get(self, url: str, **kwargs: object) -> requests.Response: + """Record the request; return or raise the scripted outcome.""" + self.calls.append((url, kwargs)) + if isinstance(self.outcome, Exception): + raise self.outcome + return self.outcome + + +def _arm( + stream: SessionStream, + monkeypatch: pytest.MonkeyPatch, + outcome: requests.Response | Exception, +) -> _FakeHttp: + """Give ``stream`` a canned auth and HTTP layer.""" + auth = StreamAuth("https://db/x.json", "TOK", _RENEW) + monkeypatch.setattr(_session_stream, "stream_auth", lambda _path: auth) + http = _FakeHttp(outcome) + monkeypatch.setattr(stream, "_http", http) + return http + + +class TestConnectOnce: + """Opening the connection and classifying HTTP failures.""" + + def test_streams_with_token_and_sse_headers( + self, stream: SessionStream, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A 200 is consumed with the auth's renew deadline, then released.""" + response = fake_response() + http = _arm(stream, monkeypatch, response) + consumed: list[tuple[requests.Response, datetime]] = [] + monkeypatch.setattr( + stream, "_consume", lambda resp, renew: consumed.append((resp, renew)) + ) + stream._connect_once() + url, kwargs = http.calls[0] + assert url == "https://db/x.json" + assert kwargs["params"] == {"auth": "TOK"} + assert kwargs["headers"] == {"Accept": "text/event-stream"} + assert kwargs["stream"] is True + assert kwargs["timeout"] == (10, 90) + assert consumed == [(response, _RENEW)] + assert response.raw.closed is True + + @pytest.mark.parametrize( + "error", + [requests.ConnectionError("refused"), TimeoutError("connect timed out")], + ) + def test_network_error_opening_is_a_stream_error( + self, + stream: SessionStream, + monkeypatch: pytest.MonkeyPatch, + error: Exception, + ) -> None: + """Connect failures and connect timeouts back off.""" + _arm(stream, monkeypatch, error) + with pytest.raises(StreamError, match="network error opening") as ei: + stream._connect_once() + assert ei.value.__cause__ is error + + @pytest.mark.parametrize( + ("status", "kind"), + [(401, "rejected"), (403, "rejected"), (500, "failed")], + ) + def test_http_errors_are_classified( + self, + stream: SessionStream, + monkeypatch: pytest.MonkeyPatch, + status: int, + kind: str, + ) -> None: + """401/403 name the token or rules; others are plain failures.""" + response = fake_response(status_code=status, text="denied") + _arm(stream, monkeypatch, response) + with pytest.raises(StreamError, match=f"stream {kind}.*HTTP {status} denied"): + stream._connect_once() + + +class TestRunOne: + """One connection, its outcome and what gets logged.""" + + def test_clean_end_is_planned( + self, stream: SessionStream, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Returning normally means renewal or auth_revoked: planned.""" + monkeypatch.setattr(stream, "_connect_once", lambda: None) + assert stream._run_one() is True + + def test_stream_error_is_a_warning( + self, + stream: SessionStream, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + ) -> None: + """A StreamError is unplanned and logged at WARNING with the fallback.""" + + def boom() -> None: + msg = "HTTP 500" + raise StreamError(msg) + + monkeypatch.setattr(stream, "_connect_once", boom) + with caplog.at_level(logging.WARNING): + assert stream._run_one() is False + (record,) = caplog.records + assert record.levelno == logging.WARNING + assert "HTTP 500" in record.getMessage() + assert "15-min workout-sync" in record.getMessage() + + def test_unexpected_exception_is_logged_with_traceback( + self, + stream: SessionStream, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + ) -> None: + """A bug must not kill the thread -- and must not be silent.""" + + def boom() -> None: + msg = "bug" + raise RuntimeError(msg) + + monkeypatch.setattr(stream, "_connect_once", boom) + with caplog.at_level(logging.WARNING): + assert stream._run_one() is False + (record,) = caplog.records + assert record.levelno == logging.ERROR + assert record.exc_info is not None + assert "crashed" in record.getMessage() diff --git a/screen_locker/tests/test_session_stream_loop.py b/screen_locker/tests/test_session_stream_loop.py new file mode 100644 index 0000000..1123ada --- /dev/null +++ b/screen_locker/tests/test_session_stream_loop.py @@ -0,0 +1,171 @@ +"""Tests for _session_stream.run_forever: reconnect backoff and thread control.""" + +from __future__ import annotations + +import logging +from types import SimpleNamespace +from typing import TYPE_CHECKING + +import pytest + +from screen_locker import _session_stream +from screen_locker.tests.test_session_stream import make_stream + +if TYPE_CHECKING: + from screen_locker._session_stream import SessionStream + +_LONG = 400.0 # longer than the 300 s minimum lifetime + + +@pytest.fixture +def stream(monkeypatch: pytest.MonkeyPatch) -> SessionStream: + """A ``SessionStream`` that never reads or writes the real device id.""" + return make_stream(monkeypatch) + + +class _Loop: + """Scripted ``_run_one`` outcomes with a fake clock, jitter and stop event.""" + + def __init__( + self, + stream: SessionStream, + monkeypatch: pytest.MonkeyPatch, + script: list[tuple[bool, float, bool]], + factor: float = 1.0, + ) -> None: + self.script = script + self.index = 0 + self.now = 0.0 + self.waits: list[float] = [] + self.uniform_args: list[tuple[float, float]] = [] + self.factor = factor + self.stream = stream + monkeypatch.setattr(stream, "_run_one", self._run_one) + monkeypatch.setattr(stream, "_stop", self) + monkeypatch.setattr( + _session_stream, "time", SimpleNamespace(monotonic=lambda: self.now) + ) + monkeypatch.setattr(_session_stream, "_JITTER", self) + + def _run_one(self) -> bool: + planned, duration, snapshot = self.script[self.index] + self.index += 1 + self.now += duration + self.stream._got_snapshot = snapshot + return planned + + def is_set(self) -> bool: + """Stop once the script is exhausted.""" + return self.index >= len(self.script) + + def wait(self, pause: float) -> None: + """Record the backoff instead of sleeping.""" + self.waits.append(pause) + + def uniform(self, low: float, high: float) -> float: + """Record the jitter range; scale between the bounds by ``factor``.""" + self.uniform_args.append((low, high)) + return low + (high - low) * self.factor + + +class TestRunForever: + """Reconnect backoff: growth, cap, reset and the tight-loop guard.""" + + @staticmethod + def _run(loop: _Loop) -> list[float]: + loop.stream.run_forever() + return loop.waits + + def test_backoff_doubles_and_is_capped( + self, stream: SessionStream, monkeypatch: pytest.MonkeyPatch + ) -> None: + """1, 2, 4 ... capped at 60 s while connections keep failing fast.""" + loop = _Loop(stream, monkeypatch, [(False, 0.0, False)] * 8) + assert self._run(loop) == [1, 2, 4, 8, 16, 32, 60, 60] + + def test_pause_is_jittered_between_half_and_full( + self, stream: SessionStream, monkeypatch: pytest.MonkeyPatch + ) -> None: + """The pause is the delay times a factor drawn from 0.5..1.0.""" + loop = _Loop(stream, monkeypatch, [(False, 0.0, False)] * 2, factor=0.0) + assert self._run(loop) == [0.5, 1.0] + assert set(loop.uniform_args) == {(0.5, 1.0)} + + def test_long_lived_connection_with_snapshot_resets_the_backoff( + self, stream: SessionStream, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A healthy connection (snapshot + 5 min) restarts the ladder at 1 s.""" + script = [(False, 0.0, False)] * 3 + [(False, _LONG, True), (False, 0.0, False)] + assert self._run(_Loop(stream, monkeypatch, script)) == [1, 2, 4, 1, 2] + + def test_long_connection_without_snapshot_does_not_reset( + self, stream: SessionStream, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Surviving 5 min without ever getting data proves nothing.""" + script = [(False, 0.0, False)] * 3 + [(False, _LONG, False)] + assert self._run(_Loop(stream, monkeypatch, script)) == [1, 2, 4, 8] + + def test_short_snapshot_connection_does_not_reset( + self, stream: SessionStream, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A snapshot that dies within 5 min is a flapping link: keep backing off.""" + script = [(False, 0.0, False)] * 2 + [(False, 5.0, True)] + assert self._run(_Loop(stream, monkeypatch, script)) == [1, 2, 4] + + def test_planned_end_after_a_long_life_reconnects_at_once( + self, stream: SessionStream, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Token renewal after a healthy hour is not a failure: no pause.""" + script: list[tuple[bool, float, bool]] = [(True, 3600.0, True)] * 2 + assert self._run(_Loop(stream, monkeypatch, script)) == [] + + def test_planned_end_after_a_short_life_still_backs_off( + self, stream: SessionStream, monkeypatch: pytest.MonkeyPatch + ) -> None: + """auth_revoked straight after connecting must not spin in a tight loop.""" + script: list[tuple[bool, float, bool]] = [(True, 1.0, True)] * 3 + assert self._run(_Loop(stream, monkeypatch, script)) == [1, 2, 4] + + def test_each_pause_is_logged_as_a_warning( + self, + stream: SessionStream, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + ) -> None: + """Reconnect pauses are visible in the journal.""" + loop = _Loop(stream, monkeypatch, [(False, 0.0, False)]) + with caplog.at_level(logging.WARNING): + self._run(loop) + (record,) = caplog.records + assert record.levelno == logging.WARNING + assert record.getMessage() == "Reconnecting the session stream in 1.0 s" + + def test_stop_before_start_runs_nothing( + self, stream: SessionStream, monkeypatch: pytest.MonkeyPatch + ) -> None: + """A set stop flag exits without connecting.""" + loop = _Loop(stream, monkeypatch, []) + stream.run_forever() + assert loop.index == 0 + + +class TestThread: + """start() and stop().""" + + def test_start_runs_the_loop_on_a_named_daemon_thread( + self, stream: SessionStream, monkeypatch: pytest.MonkeyPatch + ) -> None: + """The loop runs off the caller's thread, as a daemon.""" + ran: list[str] = [] + monkeypatch.setattr(stream, "run_forever", lambda: ran.append("ran")) + thread = stream.start() + thread.join(timeout=5) + assert ran == ["ran"] + assert thread.daemon is True + assert thread.name == "firebase-session-stream" + + def test_stop_sets_the_flag(self, stream: SessionStream) -> None: + """stop() asks the loop to exit after the current connection.""" + assert stream._stop.is_set() is False + stream.stop() + assert stream._stop.is_set() is True diff --git a/screen_locker/tests/test_stream_auth.py b/screen_locker/tests/test_stream_auth.py new file mode 100644 index 0000000..322cb1a --- /dev/null +++ b/screen_locker/tests/test_stream_auth.py @@ -0,0 +1,133 @@ +"""Tests for _stream_auth: token, URL and renew deadline for the stream.""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from unittest.mock import MagicMock + +from crdt_sync import ( + ConfigError, + FirebaseAuthError, + FirebaseConfig, + FirebaseCredentials, +) +import pytest + +from screen_locker import _stream_auth +from screen_locker._credential_recovery import RecoveryResult +from screen_locker._stream_auth import StreamAuth, StreamError, stream_auth + +_CONFIG = FirebaseConfig( + api_key="key", + database_url="https://db.example/", + project_id="proj", + uid="uid", + email="me@example.com", +) +_EXPIRY = datetime(2030, 1, 1, 12, 0, tzinfo=UTC) + + +def _provider(*, has_session: bool = True, minted: str = "TOKEN") -> MagicMock: + """A token provider fake with a cached (or missing) session.""" + provider = MagicMock() + provider.has_session.return_value = has_session + provider.id_token.return_value = minted + return provider + + +class _Fakes: + """The patched collaborators of ``stream_auth``.""" + + def __init__(self) -> None: + self.store = MagicMock() + self.store.load.return_value = FirebaseCredentials("id", "refresh", _EXPIRY) + self.providers: list[MagicMock] = [_provider()] + self.recovery = RecoveryResult(recovered=True, reason="ok", donor="todo") + self.recover = MagicMock(side_effect=lambda: self.recovery) + + +@pytest.fixture +def fakes(monkeypatch: pytest.MonkeyPatch) -> _Fakes: + """Patch config loading, the credential store, providers and recovery.""" + fake = _Fakes() + monkeypatch.setattr( + _stream_auth.FirebaseConfig, "load", classmethod(lambda _cls: _CONFIG) + ) + monkeypatch.setattr(_stream_auth, "credential_store_for", lambda _app: fake.store) + monkeypatch.setattr( + _stream_auth, + "FirebaseTokenProvider", + MagicMock(side_effect=lambda *_a, **_k: fake.providers.pop(0)), + ) + monkeypatch.setattr(_stream_auth, "try_recover_firebase_session", fake.recover) + return fake + + +class TestCachedSession: + """The common path: the cache already holds a session.""" + + def test_url_token_and_renew_at_from_saved_expiry(self, fakes: _Fakes) -> None: + """The URL is the encoded path on the DB host; renew 4 min before expiry.""" + auth = stream_auth("screen-locker-sync/devices") + assert auth == StreamAuth( + "https://db.example/screen-locker-sync/devices.json", + "TOKEN", + _EXPIRY - timedelta(minutes=4), + ) + fakes.recover.assert_not_called() + + def test_assumed_lifetime_when_nothing_is_saved(self, fakes: _Fakes) -> None: + """No stored expiry: assume an hour, still renewing 4 minutes early.""" + fakes.store.load.return_value = None + before = datetime.now(UTC) + auth = stream_auth("a/b") + after = datetime.now(UTC) + lifetime = timedelta(hours=1) - timedelta(minutes=4) + assert before + lifetime <= auth.renew_at <= after + lifetime + + +class TestRecovery: + """A missing session is healed the way the sync heals it.""" + + def test_recovered_session_is_used(self, fakes: _Fakes) -> None: + """After recovery a fresh provider over the same store mints the token.""" + fakes.providers = [ + _provider(has_session=False), + _provider(minted="RECOVERED"), + ] + auth = stream_auth("a/b") + assert auth.url.endswith("/a/b.json") + assert auth == StreamAuth(auth.url, "RECOVERED", auth.renew_at) + fakes.recover.assert_called_once_with() + + def test_failed_recovery_is_a_stream_error_with_the_reason( + self, fakes: _Fakes + ) -> None: + """No donor session: back off, and say why.""" + fakes.providers = [_provider(has_session=False)] + fakes.recovery = RecoveryResult(recovered=False, reason="no donor found") + with pytest.raises(StreamError, match="recovery failed: no donor found"): + stream_auth("a/b") + + +class TestUnusableAuth: + """Library errors become StreamError, keeping the cause.""" + + def test_config_error(self, fakes: _Fakes, monkeypatch: pytest.MonkeyPatch) -> None: + """A broken Firebase config file is a StreamError.""" + + def boom(_cls: type[FirebaseConfig]) -> FirebaseConfig: + msg = "placeholder key" + raise ConfigError(msg) + + monkeypatch.setattr(_stream_auth.FirebaseConfig, "load", classmethod(boom)) + with pytest.raises(StreamError, match="config unusable: placeholder key") as ei: + stream_auth("a/b") + assert isinstance(ei.value.__cause__, ConfigError) + + def test_firebase_auth_error(self, fakes: _Fakes) -> None: + """A refresh the server rejects is a StreamError.""" + fakes.providers[0].id_token.side_effect = FirebaseAuthError("revoked") + with pytest.raises(StreamError, match="ID token: revoked") as ei: + stream_auth("a/b") + assert isinstance(ei.value.__cause__, FirebaseAuthError) diff --git a/screen_locker/tests/test_stream_cache.py b/screen_locker/tests/test_stream_cache.py new file mode 100644 index 0000000..0cf379e --- /dev/null +++ b/screen_locker/tests/test_stream_cache.py @@ -0,0 +1,169 @@ +"""Tests for _stream_cache: merge, tombstones, own-device rule and crediting.""" + +from __future__ import annotations + +import json +import logging +import threading +from typing import TYPE_CHECKING + +from crdt_sync import Hlc, Record +import pytest + +from screen_locker import _stream_cache +from screen_locker._poke_credit import BatchOutcome +from screen_locker._rtdb_stream import SseEvent +from screen_locker._stream_cache import DeviceLogCache + +if TYPE_CHECKING: + from collections.abc import Iterable, Mapping + +_LEAF = "log~2Ejson" +OWN = "pc-own" +_PAYLOAD = {"exercises": [], "date": "2026-10-09"} + + +def log_blob( + rid: str, *, wall: int = 1, deleted: bool = False, payload: object = None +) -> str: + """One device-log blob holding a single (possibly tombstoned) record.""" + hlc = Hlc(wall, 0, "n") + record = Record( + rid, + {"payload": (_PAYLOAD if payload is None else payload, hlc)}, + deleted=deleted, + deleted_hlc=hlc if deleted else None, + ) + return json.dumps({rid: record.to_dict()}) + + +def snapshot(logs: Mapping[str, str]) -> SseEvent: + """A ``put /`` carrying every device log.""" + data = {dev: {_LEAF: text} for dev, text in logs.items()} + return SseEvent("put", json.dumps({"path": "/", "data": data})) + + +def patch_event(logs: Mapping[str, str | None]) -> SseEvent: + """A ``patch /`` rewriting some device logs (``None`` deletes one).""" + data = {f"{dev}/{_LEAF}": text for dev, text in logs.items()} + return SseEvent("patch", json.dumps({"path": "/", "data": data})) + + +class Credit: + """Stand-in for ``credit_sessions`` that records every batch it is handed.""" + + def __init__(self) -> None: + self.batches: list[list[str]] = [] + self.error: Exception | None = None + self.gate_held: list[bool] = [] + self.gate: threading.Lock | None = None + + def __call__( + self, records: Iterable[tuple[str, Mapping[str, object]]] + ) -> BatchOutcome: + if self.gate is not None: + self.gate_held.append(self.gate.locked()) + if self.error is not None: + raise self.error + ids = [rid for rid, _payload in records] + self.batches.append(ids) + return BatchOutcome(ingested=tuple(ids), credited=len(ids)) + + +@pytest.fixture +def credit(monkeypatch: pytest.MonkeyPatch) -> Credit: + """Replace the real (locker-building) credit step with a recorder.""" + fake = Credit() + monkeypatch.setattr(_stream_cache._poke_credit, "credit_sessions", fake) + return fake + + +@pytest.fixture +def cache(credit: Credit) -> DeviceLogCache: + """A cache whose own device dir is ``pc-own``.""" + gate = threading.Lock() + credit.gate = gate + return DeviceLogCache("sync/devices", frozenset({OWN}), gate) + + +class TestCredit: + """Which events credit, and how.""" + + def test_snapshot_credits_foreign_sessions_under_the_gate( + self, cache: DeviceLogCache, credit: Credit + ) -> None: + """A phone session in the snapshot is credited while the lock is held.""" + result = cache.handle(snapshot({"phone": log_blob("s1")})) + assert result.snapshot is True + assert credit.batches == [["s1"]] + assert credit.gate_held == [True] + assert ( + "1 session(s) checked, 1 newly logged ['s1'], 1 credited" in result.summary + ) + + def test_patch_credits_a_new_session( + self, cache: DeviceLogCache, credit: Credit + ) -> None: + """A later patch for the phone credits only what is new.""" + cache.handle(snapshot({"phone": log_blob("s1")})) + result = cache.handle(patch_event({"phone": log_blob("s2")})) + assert result.snapshot is False + assert credit.batches == [["s1"], ["s2"]] + + def test_already_handed_sessions_are_a_no_op( + self, cache: DeviceLogCache, credit: Credit + ) -> None: + """Re-streaming the same snapshot costs no second credit call.""" + cache.handle(snapshot({"phone": log_blob("s1")})) + result = cache.handle(snapshot({"phone": log_blob("s1")})) + assert credit.batches == [["s1"]] + assert "no-op, no file I/O" in result.summary + assert result.snapshot is True + + def test_a_newer_hlc_of_the_same_record_is_credited_again( + self, cache: DeviceLogCache, credit: Credit + ) -> None: + """The seen-set is keyed on (id, hlc): an edit is a new thing to check.""" + cache.handle(snapshot({"phone": log_blob("s1", wall=1)})) + cache.handle(patch_event({"phone": log_blob("s1", wall=2)})) + assert credit.batches == [["s1"], ["s1"]] + + def test_non_session_records_are_not_credited( + self, cache: DeviceLogCache, credit: Credit + ) -> None: + """A record without ``exercises`` (e.g. runnerup_verified) is skipped.""" + cache.handle(snapshot({"phone": log_blob("v1", payload={"kind": "x"})})) + assert credit.batches == [] + + +class TestFailures: + """Nothing is swallowed silently; nothing is marked seen on failure.""" + + def test_credit_failure_is_logged_and_retried_next_event( + self, + cache: DeviceLogCache, + credit: Credit, + caplog: pytest.LogCaptureFixture, + ) -> None: + """A crediting error logs at ERROR, keeps the stream, retries later.""" + credit.error = RuntimeError("disk full") + with caplog.at_level(logging.WARNING): + result = cache.handle(snapshot({"phone": log_blob("s1")})) + assert result.summary == "crediting failed (see above)" + assert result.snapshot is True + failures = [r for r in caplog.records if r.levelno >= logging.ERROR] + assert failures + assert failures[0].exc_info is not None + assert "Crediting 1 streamed session(s) failed" in failures[0].getMessage() + credit.error = None + cache.handle(snapshot({"phone": log_blob("s1")})) + assert credit.batches == [["s1"]] + + def test_bad_json_propagates_for_the_caller_to_warn( + self, cache: DeviceLogCache + ) -> None: + """Undecodable events are the connection's to report, not ours to eat.""" + with pytest.raises(json.JSONDecodeError): + cache.handle(SseEvent("put", "{nope")) + with pytest.raises(TypeError, match="without a path"): + cache.handle(SseEvent("put", "[]")) diff --git a/screen_locker/tests/test_stream_cache_merge.py b/screen_locker/tests/test_stream_cache_merge.py new file mode 100644 index 0000000..bab172d --- /dev/null +++ b/screen_locker/tests/test_stream_cache_merge.py @@ -0,0 +1,137 @@ +"""Tests for _stream_cache: union merge, tombstones and the own-device rule.""" + +from __future__ import annotations + +import json +import logging +import threading + +import pytest + +from screen_locker import _stream_cache +from screen_locker._rtdb_stream import SseEvent +from screen_locker._stream_cache import DeviceLogCache +from screen_locker.tests.test_stream_cache import ( + OWN, + Credit, + log_blob, + patch_event, + snapshot, +) + + +@pytest.fixture +def credit(monkeypatch: pytest.MonkeyPatch) -> Credit: + """Replace the real (locker-building) credit step with a recorder.""" + fake = Credit() + monkeypatch.setattr(_stream_cache._poke_credit, "credit_sessions", fake) + return fake + + +@pytest.fixture +def cache(credit: Credit) -> DeviceLogCache: + """A cache whose own device dir is ``pc-own``.""" + gate = threading.Lock() + credit.gate = gate + return DeviceLogCache("sync/devices", frozenset({OWN}), gate) + + +class TestMerge: + """Union across devices, tombstones and the own-device rule.""" + + def test_highest_hlc_copy_wins_across_devices( + self, cache: DeviceLogCache, credit: Credit + ) -> None: + """One record mirrored in two device logs is credited once.""" + both = {"phone": log_blob("s1", wall=1), "tab": log_blob("s1", wall=5)} + result = cache.handle(snapshot(both)) + assert credit.batches == [["s1"]] + assert "1 session(s) checked" in result.summary + + def test_tombstone_on_another_device_drops_the_record( + self, cache: DeviceLogCache, credit: Credit + ) -> None: + """A delete anywhere beats a live copy elsewhere: nothing is credited.""" + logs = {"phone": log_blob("s1"), "tab": log_blob("s1", deleted=True)} + result = cache.handle(snapshot(logs)) + assert credit.batches == [] + assert "no-op" in result.summary + + def test_own_log_tombstone_still_suppresses_a_phone_session( + self, cache: DeviceLogCache, credit: Credit + ) -> None: + """The PC's own log stays in the union: its tombstones must be honoured.""" + logs = {OWN: log_blob("s1", deleted=True), "phone": log_blob("s1")} + cache.handle(snapshot(logs)) + assert credit.batches == [] + + def test_event_touching_only_the_own_log_never_credits( + self, cache: DeviceLogCache, credit: Credit + ) -> None: + """Our own push echoing back must not trigger a pass.""" + result = cache.handle(patch_event({OWN: log_blob("s9")})) + assert credit.batches == [] + assert "only this PC's own log changed (pc-own); no credit" in result.summary + assert result.snapshot is False + + def test_own_snapshot_is_still_processed( + self, cache: DeviceLogCache, credit: Credit + ) -> None: + """A full snapshot always runs the union, even if only own logs differ.""" + cache.handle(snapshot({OWN: log_blob("s9")})) + assert credit.batches == [["s9"]] + + def test_byte_identical_rewrite_does_nothing( + self, cache: DeviceLogCache, credit: Credit + ) -> None: + """RTDB-style repeat of the same text: nothing to read or credit.""" + cache.handle(snapshot({"phone": log_blob("s1")})) + result = cache.handle(patch_event({"phone": log_blob("s1")})) + assert "byte-identical rewrite" in result.summary + assert credit.batches == [["s1"]] + + def test_deleting_an_unknown_device_is_a_no_op(self, cache: DeviceLogCache) -> None: + """A ``null`` for a device never cached changes nothing.""" + result = cache.handle(patch_event({"ghost": None})) + assert "byte-identical rewrite" in result.summary + + def test_patch_null_removes_a_cached_device( + self, cache: DeviceLogCache, credit: Credit + ) -> None: + """Deleting a device log drops it from the cache.""" + cache.handle(snapshot({"phone": log_blob("s1")})) + result = cache.handle(patch_event({"phone": None})) + assert "0 device log(s) cached, changed: phone" in result.summary + assert credit.batches == [["s1"]] + + def test_snapshot_replaces_the_cache(self, cache: DeviceLogCache) -> None: + """Devices missing from a new snapshot are gone, and reported changed.""" + cache.handle(snapshot({"a": log_blob("s1"), "b": log_blob("s2")})) + result = cache.handle(snapshot({"a": log_blob("s1")})) + assert "1 device log(s) cached, changed: b;" in result.summary + + def test_snapshot_ignores_devices_without_alog_blob( + self, cache: DeviceLogCache + ) -> None: + """A device node with no text is not cached.""" + event = SseEvent( + "put", json.dumps({"path": "/", "data": {"a": {"other": "x"}}}) + ) + result = cache.handle(event) + assert "0 device log(s) cached" in result.summary + + def test_corrupt_log_is_skipped_with_a_warning( + self, + cache: DeviceLogCache, + credit: Credit, + caplog: pytest.LogCaptureFixture, + ) -> None: + """One broken device log does not hide the others' sessions.""" + with caplog.at_level(logging.WARNING): + cache.handle(snapshot({"bad": "not json", "phone": log_blob("s1")})) + assert credit.batches == [["s1"]] + assert any( + "Corrupt sync data at sync/devices/bad/log.json" in r.getMessage() + and r.levelno >= logging.WARNING + for r in caplog.records + ) diff --git a/scripts/pair_phone.sh b/scripts/pair_phone.sh new file mode 100755 index 0000000..6fda203 --- /dev/null +++ b/scripts/pair_phone.sh @@ -0,0 +1,102 @@ +#!/bin/bash + +# ============================================================================ +# Pair the phone's workout_app with this PC's workout poke listener. +# +# Creates ~/.config/workout_poke/key if missing (workout_poke_keygen.sh), then +# hands it and the PC's LAN address to the app over adb via its +# PairPcReceiver broadcast, which stores both in the Android keystore. Success +# is decided by the broadcast's own reply -- `adb shell` exits 0 even when +# the receiver is missing -- so it needs both `result=-1` and `data="paired"`. +# +# The key is on adb's command line for the broadcast's lifetime (that is the +# receiver's interface) but is never printed. +# +# Usage: pair_phone.sh [--sandbox] [--host IP] [-s SERIAL] +# --sandbox pair the sandbox flavor (com.kuhy.workout_app.sandbox) +# --host IP the PC address the phone should poke (default 192.168.1.43) +# -s SERIAL passed to adb when several devices are attached +# ============================================================================ + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +readonly SCRIPT_DIR +readonly SCRIPT_NAME="${0##*/}" +readonly KEY_FILE="${HOME}/.config/workout_poke/key" +readonly ACTION="com.kuhy.workout_app.PAIR_PC" +readonly RECEIVER="com.kuhy.workout_app.PairPcReceiver" + +PACKAGE="com.kuhy.workout_app" +HOST="192.168.1.43" +ADB_ARGS=() + +usage() { + echo "Usage: $SCRIPT_NAME [--sandbox] [--host IP] [-s SERIAL]" + exit 0 +} + +require_adb() { + if ! command -v adb >/dev/null 2>&1; then + echo "Installing missing dependency: android-tools (adb)" + sudo pacman -S --needed --noconfirm android-tools + fi +} + +# The broadcast's reply minus the key, should anything ever echo it back. +redact() { + local text="$1" key="$2" + printf '%s\n' "${text//"$key"/}" +} + +main() { + require_adb + bash "$SCRIPT_DIR/workout_poke_keygen.sh" + local key output + key="$(<"$KEY_FILE")" + echo "Pairing ${PACKAGE} with host ${HOST} ..." + if ! output="$(adb "${ADB_ARGS[@]}" shell am broadcast \ + -a "$ACTION" \ + -n "${PACKAGE}/${RECEIVER}" \ + --es key "$key" \ + --es host "$HOST" 2>&1)"; then + echo "Error: adb failed (is the phone connected and authorized?):" >&2 + redact "$output" "$key" >&2 + exit 1 + fi + if [[ $output == *"result=-1"* && $output == *'data="paired"'* ]]; then + echo "paired" + return 0 + fi + echo "Error: ${PACKAGE} did not confirm the pairing. A missing" \ + "'result=-1' usually means the app (or this flavor) is not installed" \ + "or predates PairPcReceiver. adb said:" >&2 + redact "$output" "$key" >&2 + exit 1 +} + +while [[ $# -gt 0 ]]; do + case $1 in + --sandbox) + PACKAGE="com.kuhy.workout_app.sandbox" + shift + ;; + --host) + HOST="${2:?--host needs an address}" + shift 2 + ;; + -s) + ADB_ARGS=(-s "${2:?-s needs a device serial}") + shift 2 + ;; + -h | --help) + usage + ;; + *) + echo "Unknown option: $1" >&2 + exit 1 + ;; + esac +done + +main diff --git a/scripts/poke_selftest.py b/scripts/poke_selftest.py new file mode 100755 index 0000000..b134124 --- /dev/null +++ b/scripts/poke_selftest.py @@ -0,0 +1,137 @@ +#!/usr/bin/env python3 +"""Drive the running workout poke listener from this PC, signed with the real key. + +Each case POSTs to ``http://127.0.0.1:8773/v1/workout`` exactly as the phone +would (``docs/DOCS-workout-poke-contract.md``) and prints the status and reply: + +* ``sandbox`` -- a signed sandbox request: 200, ok, credited=false; +* ``badsig`` -- a signature for different bytes: 401; +* ``replay`` -- the same nonce twice: 200 (sandbox), then 409; +* ``stale`` -- ``sent_at_ms`` 10 minutes old: 401; +* ``duplicate`` -- a NON-sandbox poke of today's already-logged session + (record ``2026-10-09T15:10:40.953``): duplicate=true, credited=false; +* ``latency`` -- N sandbox round trips, p50/p95 in ms. + +Usage: ``python3 scripts/poke_selftest.py CASE [--count N] [--port P]``. +""" + +from __future__ import annotations + +import argparse +import hashlib +import hmac +import http.client +import json +from pathlib import Path +import statistics +import sys +import time +import uuid + +KEY_FILE = Path.home() / ".config" / "workout_poke" / "key" +STALE_MS = 600_000 +DUPLICATE_RECORD_ID = "2026-10-09T15:10:40.953" +# The session as the phone syncs it (same shape as the shared fixture). +DUPLICATE_PAYLOAD: dict[str, object] = { + "date": "2026-10-09", + "start_time": DUPLICATE_RECORD_ID, + "workout_type": "A", + "duration_seconds": 7020.5, + "succeeded": True, + "exercises": [{"name": "Squat", "weight_kg": 100.0, "sets": [5, 5, 5, 5, 5]}], +} + + +def _body( + *, sandbox: bool, sent_at_ms: int | None = None, nonce: str | None = None +) -> bytes: + """One request body; compact separators like the phone's encoder.""" + doc = { + "v": 1, + "sent_at_ms": sent_at_ms + if sent_at_ms is not None + else time.time_ns() // 1_000_000, + "nonce": nonce or str(uuid.uuid4()), + "sandbox": sandbox, + "record_id": DUPLICATE_RECORD_ID, + "payload": DUPLICATE_PAYLOAD, + } + return json.dumps(doc, separators=(",", ":"), ensure_ascii=False).encode("utf-8") + + +def _post( + port: int, key: bytes, body: bytes, *, sign_over: bytes | None = None +) -> tuple[int, dict[str, object], float]: + """POST ``body``, signed over ``sign_over`` (default: itself). + + Returns (status, reply, round-trip ms). + """ + signature = hmac.new( + key, sign_over if sign_over is not None else body, hashlib.sha256 + ).hexdigest() + started = time.perf_counter() + conn = http.client.HTTPConnection("127.0.0.1", port, timeout=10) + try: + conn.request( + "POST", + "/v1/workout", + body=body, + headers={"Content-Type": "application/json", "X-Poke-Signature": signature}, + ) + response = conn.getresponse() + reply = json.loads(response.read()) + finally: + conn.close() + return response.status, reply, (time.perf_counter() - started) * 1000 + + +def _show(label: str, result: tuple[int, dict[str, object], float]) -> int: + status, reply, ms = result + shown = json.dumps(reply, ensure_ascii=False) + sys.stdout.write(f"{label}: HTTP {status} rtt={ms:.1f}ms {shown}\n") + return status + + +def _latency(port: int, key: bytes, count: int) -> None: + rtts = sorted(_post(port, key, _body(sandbox=True))[2] for _ in range(count)) + p95 = rtts[max(0, round(0.95 * count) - 1)] + p50 = statistics.median(rtts) + sys.stdout.write( + f"latency over {count} sandbox pokes: p50={p50:.1f}ms " + f"p95={p95:.1f}ms max={rtts[-1]:.1f}ms\n" + ) + + +def main() -> int: + """Run one case against the local listener.""" + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument( + "case", choices=["sandbox", "badsig", "replay", "stale", "duplicate", "latency"] + ) + parser.add_argument("--count", type=int, default=20) + parser.add_argument("--port", type=int, default=8773) + args = parser.parse_args() + key = bytes.fromhex(KEY_FILE.read_text(encoding="ascii").strip()) + if args.case == "sandbox": + _show("sandbox", _post(args.port, key, _body(sandbox=True))) + elif args.case == "badsig": + _show( + "badsig", + _post(args.port, key, _body(sandbox=True), sign_over=b"something else"), + ) + elif args.case == "replay": + body = _body(sandbox=True) + _show("replay first", _post(args.port, key, body)) + _show("replay again", _post(args.port, key, body)) + elif args.case == "stale": + stale = time.time_ns() // 1_000_000 - STALE_MS + _show("stale", _post(args.port, key, _body(sandbox=True, sent_at_ms=stale))) + elif args.case == "duplicate": + _show("duplicate (NON-sandbox)", _post(args.port, key, _body(sandbox=False))) + else: + _latency(args.port, key, args.count) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/workout_poke_keygen.sh b/scripts/workout_poke_keygen.sh new file mode 100755 index 0000000..74511b2 --- /dev/null +++ b/scripts/workout_poke_keygen.sh @@ -0,0 +1,87 @@ +#!/bin/bash + +# ============================================================================ +# Create the phone -> PC workout poke key if it does not exist yet. +# +# The key is 32 random bytes as 64 hex characters in +# ~/.config/workout_poke/key (mode 600, dir 700). workout-poke.service verifies +# each poke's HMAC with it and refuses to start without it; pair_phone.sh +# copies it to the phone. An existing key is NEVER overwritten: a new key +# would silently unpair the phone. Prints the path, never the key. +# +# Usage: workout_poke_keygen.sh [--key-file PATH] +# ============================================================================ + +set -euo pipefail + +KEY_FILE="${HOME}/.config/workout_poke/key" +readonly SCRIPT_NAME="${0##*/}" +readonly KEY_HEX_LENGTH=64 + +usage() { + echo "Usage: $SCRIPT_NAME [--key-file PATH]" + exit 0 +} + +# An existing key must still be well-formed and private, or the daemon will +# refuse it; say so here rather than let the unit fail later. +check_existing_key() { + local key mode + mode="$(stat -c '%a' "$KEY_FILE")" + if [[ $mode != "600" ]]; then + echo "Error: $KEY_FILE has mode $mode; run: chmod 600 $KEY_FILE" >&2 + exit 1 + fi + key="$(<"$KEY_FILE")" + if [[ ! $key =~ ^[0-9a-f]{$KEY_HEX_LENGTH}$ ]]; then + echo "Error: $KEY_FILE is not $KEY_HEX_LENGTH lowercase hex characters;" \ + "move it aside and re-run (the phone must then be re-paired)" >&2 + exit 1 + fi + echo "Key already present: $KEY_FILE (left unchanged)" +} + +create_key() { + local dir key + dir="$(dirname "$KEY_FILE")" + mkdir -p "$dir" + chmod 700 "$dir" + key="$(od -An -N32 -tx1 /dev/urandom | tr -d ' \n')" + if [[ ! $key =~ ^[0-9a-f]{$KEY_HEX_LENGTH}$ ]]; then + echo "Error: could not read 32 random bytes from /dev/urandom" >&2 + exit 1 + fi + # umask + noclobber: the file is born 600 and a racing creator loses. + ( + umask 077 + set -o noclobber + printf '%s\n' "$key" >"$KEY_FILE" + ) + echo "Created key: $KEY_FILE (mode 600)" +} + +main() { + if [[ -e $KEY_FILE ]]; then + check_existing_key + else + create_key + fi +} + +while [[ $# -gt 0 ]]; do + case $1 in + --key-file) + KEY_FILE="$2" + shift 2 + ;; + -h | --help) + usage + ;; + *) + echo "Unknown option: $1" >&2 + exit 1 + ;; + esac +done + +main diff --git a/workout-poke.service b/workout-poke.service new file mode 100644 index 0000000..6c2f652 --- /dev/null +++ b/workout-poke.service @@ -0,0 +1,17 @@ +[Unit] +Description=Credit a phone workout the second it is poked over the LAN (tcp/8773) + +[Service] +# Resident stdlib HTTP server (screen_locker/_poke_server.py): the phone POSTs +# a finished session, it is HMAC-verified and credited through the same ingest +# as workout-sync. Contract: docs/DOCS-workout-poke-contract.md. +# No NoNewPrivileges/ProtectSystem: the credit runs `sudo +# adjust_shutdown_schedule.sh`, which NoNewPrivileges would break. +# A missing or bad ~/.config/workout_poke/key makes it exit 1 with a critical +# log line; RestartSec keeps that from spinning. +ExecStart=/usr/bin/python3 -m screen_locker.poke_server +Restart=always +RestartSec=30 + +[Install] +WantedBy=default.target