diff --git a/gatelock/README.md b/gatelock/README.md index b9cc954..a79592d 100644 --- a/gatelock/README.md +++ b/gatelock/README.md @@ -65,6 +65,13 @@ overridden as a unit. projects; `DEFAULT_HMAC_KEY_FILE` (`/etc/workout-locker/hmac.key`) is unchanged so existing signed history keeps verifying. +`gatelock.morning_session.morning_skip` is the one reader of wake-alarm's +signed `morning_session.json` (the wake-early carrot: no lock until the signed +`exempt_until`, 11:00 for a morning completed in time). screen-locker, +leetcode-guard and book-guard all call it; the policy itself stays in +wake-alarm. Path and retry budget are arguments so each consumer's tests keep +redirecting their own constant. + ## Development ```bash diff --git a/gatelock/gatelock/morning_session.py b/gatelock/gatelock/morning_session.py new file mode 100644 index 0000000..53e6857 --- /dev/null +++ b/gatelock/gatelock/morning_session.py @@ -0,0 +1,149 @@ +# Copyright (c) 2026 Krzysztof Rudnicki +"""Read the morning-session carrot wake-alarm signed for today. + +The phone's morning (weigh-in, shower, dress, desk) is published to the PC +and distilled by wake-alarm into ``morning_session.json``, HMAC-signed, always +dated today, with an ``exempt_until`` while the morning is live or once it was +completed in time (11:00). Every gate that honours the carrot reads it here. +This module never re-derives that policy: the whole decision is *signature +ok, dated today, now < exempt_until*. Missing, stale, failed or tampered all +mean "no skip" -- the gate runs as it always has. + +A run deferral, never ledger state: nothing is written, so a deleted file +cannot mint an unlocked day. + +The one wrinkle is a PC booted mid-morning: the file is missing or yesterday's +until wake-alarm's ``Persistent=true`` timer catches up, which needs the +network. The arming run waits for that, bounded; status paths never do. + +The path and the retry budget are parameters, not module state, so each +consumer's test suite keeps redirecting its own constant. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import UTC, datetime +import json +import logging +from pathlib import Path +import time +from typing import TYPE_CHECKING, Final + +from gatelock.log_integrity import verify_entry_hmac + +if TYPE_CHECKING: + from collections.abc import Callable + +_logger = logging.getLogger(__name__) + +MORNING_SESSION_FILE: Final = ( + Path.home() / ".local" / "state" / "wake_alarm" / "morning_session.json" +) +"""Where wake-alarm writes it. Keep in step with ``wake_alarm/_constants.py``.""" + +# Only inside this local window is a non-today file worth waiting for: the +# refresher runs 05:00-11:00, so outside it "not today's" is simply the truth. +MORNING_WINDOW: Final = ((5, 0), (11, 0)) +MORNING_RETRY_SECONDS: Final = 30.0 +MORNING_POLL_SECONDS: Final = 5.0 + + +@dataclass(frozen=True) +class MorningSkip: + """A skip the morning session earned: what the phone said, and until when.""" + + outcome: str + exempt_until: datetime + + def __str__(self) -> str: + """Say which morning earned it and when the gate comes back.""" + return f"{self.outcome} session, no lock until {self.exempt_until:%H:%M}" + + +def _read_verified(path: Path) -> dict[str, object] | None: + """The file's entry when it exists, parses and verifies; else None. + + Missing is debug (it is the normal state outside the morning); anything + else is a warning, because it means the refresher or the key is broken. + """ + if not path.exists(): + _logger.debug("No morning session file at %s", path) + return None + try: + entry = json.loads(path.read_text()) + except (OSError, json.JSONDecodeError) as exc: + _logger.warning("Cannot read %s: %s", path, exc) + return None + if not isinstance(entry, dict) or not verify_entry_hmac(entry): + _logger.warning("Morning session file is not a signed object") + return None + return entry + + +def _load_today(path: Path, now: datetime) -> dict[str, object] | None: + """Today's verified entry, or None. ``now`` is already local.""" + entry = _read_verified(path) + if entry is None or entry.get("date") != now.date().isoformat(): + return None + return entry + + +def _skip_from(entry: dict[str, object], now: datetime) -> MorningSkip | None: + """The skip an entry grants at ``now``, if its window is still open.""" + until = entry.get("exempt_until") + if not isinstance(until, str): + return None + try: + exempt_until = datetime.fromisoformat(until) + except ValueError: + _logger.warning("Morning session exempt_until unreadable: %r", until) + return None + if exempt_until.tzinfo is None or now >= exempt_until: + # A naive instant cannot be compared with a local one, and guessing + # its zone would be inventing a carrot the signer never granted. + return None + return MorningSkip(outcome=str(entry.get("outcome")), exempt_until=exempt_until) + + +def _in_window(now: datetime) -> bool: + (start_h, start_m), (end_h, end_m) = MORNING_WINDOW + minutes = now.hour * 60 + now.minute + return start_h * 60 + start_m <= minutes < end_h * 60 + end_m + + +def _local_now() -> datetime: + return datetime.now(tz=UTC).astimezone() + + +def morning_skip( + path: Path = MORNING_SESSION_FILE, + now: datetime | None = None, + *, + wait: bool, + retry_seconds: float = MORNING_RETRY_SECONDS, + sleep: Callable[[float], None] = time.sleep, +) -> MorningSkip | None: + """The skip the morning session earned at ``now``, or None. + + With ``wait`` (the arming path only), a file that is not today's during + the morning window is retried for up to ``retry_seconds`` so a freshly + booted PC gives wake-alarm's catch-up run a chance to land. + """ + now = (now or _local_now()).astimezone() + entry = _load_today(path, now) + waited = 0.0 + while entry is None and wait and _in_window(now) and waited < retry_seconds: + sleep(MORNING_POLL_SECONDS) + waited += MORNING_POLL_SECONDS + now = _local_now() + entry = _load_today(path, now) + if entry is None: + if waited: + _logger.warning( + "No morning session for today after %.0fs; " + "is wake-alarm-session.timer running?", + waited, + ) + return None + return _skip_from(entry, now) diff --git a/gatelock/gatelock/tests/test_morning_session.py b/gatelock/gatelock/tests/test_morning_session.py new file mode 100644 index 0000000..5620f32 --- /dev/null +++ b/gatelock/gatelock/tests/test_morning_session.py @@ -0,0 +1,155 @@ +# Copyright (c) 2026 Krzysztof Rudnicki +"""gatelock.morning_session -- the carrot file, read and never re-derived.""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +import json +from typing import TYPE_CHECKING +from unittest.mock import patch + +import pytest + +from gatelock import morning_session +from gatelock.log_integrity import compute_entry_hmac +from gatelock.morning_session import MorningSkip, morning_skip + +if TYPE_CHECKING: + from collections.abc import Iterator + from pathlib import Path + +NOW = datetime(2026, 9, 19, 9, 0).astimezone() + + +@pytest.fixture +def path(tmp_path: Path) -> Path: + """Where the tests write the signed file -- never the real one.""" + return tmp_path / "morning_session.json" + + +@pytest.fixture(autouse=True) +def _signing_key(tmp_path: Path) -> Iterator[None]: + key = tmp_path / "hmac.key" + key.write_bytes(b"1" * 32) + with patch("gatelock.log_integrity.DEFAULT_HMAC_KEY_FILE", key): + yield + + +def _write(path: Path, entry: dict[str, object], *, sign: bool = True) -> None: + """Write an entry; signed unless told otherwise.""" + if sign: + entry["hmac"] = compute_entry_hmac(entry) + path.write_text(json.dumps(entry)) + + +def _entry( + outcome: str = "completed", + exempt_until: str | None = "11:00", + date: str = "2026-09-19", +) -> dict[str, object]: + entry: dict[str, object] = {"date": date, "outcome": outcome} + if exempt_until is not None: + entry["exempt_until"] = ( + datetime.fromisoformat(f"{date}T{exempt_until}").astimezone().isoformat() + ) + return entry + + +def _today_entry() -> dict[str, object]: + today = datetime.now(tz=UTC).astimezone().strftime("%Y-%m-%d") + return _entry(date=today, exempt_until="23:59") + + +class TestVerdict: + """One rule: signed, today, not yet expired.""" + + def test_completed_grants_until_the_signed_instant(self, path: Path) -> None: + _write(path, _entry()) + skip = morning_skip(path, NOW, wait=False) + assert skip == MorningSkip("completed", NOW.replace(hour=11)) + assert str(skip) == "completed session, no lock until 11:00" + + def test_expired_and_absent_exemptions_grant_nothing(self, path: Path) -> None: + _write(path, _entry()) + assert morning_skip(path, NOW + timedelta(hours=3), wait=False) is None + _write(path, _entry("failed", exempt_until=None)) + assert morning_skip(path, NOW, wait=False) is None + _write(path, {**_entry(), "exempt_until": "later"}) + assert morning_skip(path, NOW, wait=False) is None + + def test_a_naive_exempt_until_grants_nothing(self, path: Path) -> None: + _write(path, {**_entry(), "exempt_until": "2026-09-19T11:00:00"}) + assert morning_skip(path, NOW, wait=False) is None + + def test_yesterdays_file_is_not_today(self, path: Path) -> None: + _write(path, _entry(date="2026-09-18")) + assert morning_skip(path, NOW, wait=False) is None + + def test_missing_unreadable_tampered_are_all_nothing(self, path: Path) -> None: + assert morning_skip(path, NOW, wait=False) is None + _write(path, _entry(), sign=False) + assert morning_skip(path, NOW, wait=False) is None + path.write_text("[1]") + assert morning_skip(path, NOW, wait=False) is None + path.write_text("{") + assert morning_skip(path, NOW, wait=False) is None + path.unlink() + path.mkdir() + assert morning_skip(path, NOW, wait=False) is None + + def test_default_now_is_the_wall_clock(self, path: Path) -> None: + _write(path, _today_entry()) + assert morning_skip(path, wait=False) is not None + + def test_default_path_is_wake_alarms_state_file(self) -> None: + assert morning_session.MORNING_SESSION_FILE.parts[-3:] == ( + "state", + "wake_alarm", + "morning_session.json", + ) + + +class TestBootRetry: + """The arming path waits for the refresher, bounded; status never does.""" + + def test_waits_inside_the_window_then_warns( + self, path: Path, caplog: pytest.LogCaptureFixture + ) -> None: + naps: list[float] = [] + # The window is re-checked against the real clock after each nap. + with patch.object(morning_session, "MORNING_WINDOW", ((0, 0), (23, 59))): + skip = morning_skip( + path, NOW, wait=True, retry_seconds=10.0, sleep=naps.append + ) + assert skip is None + assert naps == [5.0, 5.0] + assert "is wake-alarm-session.timer running" in caplog.text + + def test_a_file_landing_mid_wait_is_honoured(self, path: Path) -> None: + def land(_seconds: float) -> None: + _write(path, _today_entry()) + + with patch.object(morning_session, "MORNING_WINDOW", ((0, 0), (23, 59))): + skip = morning_skip(path, NOW, wait=True, retry_seconds=10.0, sleep=land) + assert skip is not None + + def test_never_waits_outside_the_window_or_without_wait(self, path: Path) -> None: + naps: list[float] = [] + late = NOW.replace(hour=14) + assert ( + morning_skip(path, late, wait=True, retry_seconds=10.0, sleep=naps.append) + is None + ) + assert ( + morning_skip(path, NOW, wait=False, retry_seconds=10.0, sleep=naps.append) + is None + ) + assert naps == [] + + def test_a_zero_budget_never_waits(self, path: Path) -> None: + naps: list[float] = [] + assert ( + morning_skip(path, NOW, wait=True, retry_seconds=0.0, sleep=naps.append) + is None + ) + assert naps == [] diff --git a/gatelock/pyproject.toml b/gatelock/pyproject.toml index be2be15..6946c0e 100644 --- a/gatelock/pyproject.toml +++ b/gatelock/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "gatelock" -version = "0.8.2" +version = "0.9.0" description = "Shared per-output lock-window + HMAC log-integrity backend for blocking-overlay apps" requires-python = ">=3.14" dependencies = [] # pure stdlib — tkinter is bundled with Python