diff --git a/gatelock/gatelock/__init__.py b/gatelock/gatelock/__init__.py index b05bbf6..7813ce6 100644 --- a/gatelock/gatelock/__init__.py +++ b/gatelock/gatelock/__init__.py @@ -18,6 +18,12 @@ from gatelock._arbiter import ( grab_strength, ) from gatelock._detect import OutputChangeDetector +from gatelock._escape import ( + EscapeDraft, + EscapeHistory, + EscapePolicy, + EscapeTracker, +) from gatelock._guards import assert_not_under_pytest, wait_for_x_server from gatelock._outputs import ( Output, @@ -48,6 +54,10 @@ __all__ = [ "Arbiter", "ArbiterVerdict", "Claim", + "EscapeDraft", + "EscapeHistory", + "EscapePolicy", + "EscapeTracker", "GateRoot", "LockConfig", "LockWindow", diff --git a/gatelock/gatelock/_escape.py b/gatelock/gatelock/_escape.py new file mode 100644 index 0000000..057ac98 --- /dev/null +++ b/gatelock/gatelock/_escape.py @@ -0,0 +1,300 @@ +"""A sanctioned, rate-limited way out of a lock. + +A hard lock without an escape hatch is a trap. screen-locker learned this and +grew "sick mode": you cannot always do the workout, so there is a bounded, +justified, tamper-evident way to say so. diet-guard's exit is cheap by nature +(log a meal). wake-alarm had neither -- its only exit was solving four +challenges in a row, which is a poor thing to require of someone who has just +woken up, and no exit at all if the alarm misfires at the wrong hour. + +This module is the generic core of that hatch, so every app gets the *same* +mechanism instead of a copy that drifts: + +* **Rolling budgets.** Once any window is exhausted the hatch disappears + entirely -- it is not a slider, it is a hard stop. +* **Escalating lockout.** Each recent use doubles the wait, so the hatch gets + progressively less convenient than just doing the thing. +* **A written justification.** Long enough to be inconvenient to invent, shown + back to you next time so recycled excuses are visible. +* **Tamper evidence.** Every entry is HMAC-signed with a root-owned key, so + hand-editing the history is detectable. + +App-specific policy stays with the app: screen-locker's commitment penalty and +workout debt are not general, and are deliberately not modelled here. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime, timedelta, timezone +import json +import logging +from typing import TYPE_CHECKING, Any + +from gatelock.log_integrity import compute_entry_hmac, verify_entry_hmac + +if TYPE_CHECKING: + from pathlib import Path + +_logger = logging.getLogger(__name__) + +_DATE_FORMAT = "%Y-%m-%d" + +# The severity scale the dialog presents. +_MIN_SEVERITY = 1 +_MAX_SEVERITY = 10 + + +@dataclass(frozen=True) +class EscapePolicy: + """How hard one app's escape hatch is to use. + + Defaults mirror screen-locker's sick mode, which is the proven shape. + + Attributes: + name: Short identifier used in logs and the summary line. + label: Human wording for one use, e.g. "sick day". + budget_per_7_days: Uses allowed in a rolling week. + budget_per_30_days: Uses allowed in a rolling month. + budget_per_90_days: Uses allowed in a rolling quarter. + lockout_seconds: Base wait before the hatch actually opens. + lockout_multiplier_per_recent: Wait multiplier per use in 30 days. + justification_min_chars: Minimum length of the written reason. + history_review_count: How many past reasons to show back to the user. + """ + + name: str + label: str + budget_per_7_days: int = 1 + budget_per_30_days: int = 3 + budget_per_90_days: int = 10 + lockout_seconds: int = 120 + lockout_multiplier_per_recent: int = 2 + justification_min_chars: int = 120 + history_review_count: int = 10 + + +@dataclass +class EscapeHistory: + """Persistent record of every use of one app's hatch.""" + + used_days: list[str] = field(default_factory=list) + justifications: list[dict[str, Any]] = field(default_factory=list) + + +@dataclass(frozen=True) +class EscapeDraft: + """A proposed use of the hatch, before validation.""" + + reason: str + onset: str + severity: int + description: str + + +def _today_iso() -> str: + """Return today's date as ``YYYY-MM-DD`` (UTC).""" + return datetime.now(tz=timezone.utc).strftime(_DATE_FORMAT) + + +def _parse_iso(date_str: str) -> datetime | None: + """Parse ``YYYY-MM-DD`` into a UTC datetime, or None if unparsable. + + An unparsable entry is *ignored* rather than fatal, but logged loudly: + silently counting it as zero would quietly widen the budget. + """ + try: + return datetime.strptime(date_str, _DATE_FORMAT).replace(tzinfo=timezone.utc) + except ValueError as exc: + _logger.warning( + "escape history holds an unparsable date %r (%s) -- that entry is " + "IGNORED when counting against the budget", + date_str, + exc, + ) + return None + + +class EscapeTracker: + """Budget, lockout and tamper-evident history for one app's hatch.""" + + def __init__( + self, + policy: EscapePolicy, + path: Path, + *, + key_file: Path | None = None, + ) -> None: + """Bind a policy to the file its history lives in.""" + self._policy = policy + self._path = path + self._key_file = key_file + self.history = EscapeHistory() + self.tampered = 0 + """How many loaded entries failed their integrity check.""" + + @property + def policy(self) -> EscapePolicy: + """The policy this tracker enforces.""" + return self._policy + + def load(self) -> EscapeHistory: + """Read history from disk, tolerating a missing or corrupt file. + + **Integrity problems never reduce recorded usage.** An entry that + fails its HMAC check is kept for budget counting and reported loudly, + rather than dropped. Dropping would be the intuitive move and is + exactly wrong: ``verify_entry_hmac`` also returns False when the key + file is merely *unreadable*, so "drop what you cannot verify" would + turn a chmod on the key into an unlimited escape hatch. Tampering is + something to shout about, not something that should widen the budget. + """ + try: + raw = json.loads(self._path.read_text(encoding="utf-8")) + except FileNotFoundError: + self.history = EscapeHistory() + return self.history + except (OSError, json.JSONDecodeError) as exc: + _logger.warning("could not read escape history %s: %s", self._path, exc) + self.history = EscapeHistory() + return self.history + if not isinstance(raw, dict): + self.history = EscapeHistory() + return self.history + justifications = [ + entry for entry in raw.get("justifications", []) if isinstance(entry, dict) + ] + self.tampered = sum(1 for entry in justifications if not self._verified(entry)) + self.history = EscapeHistory( + used_days=[d for d in raw.get("used_days", []) if isinstance(d, str)], + justifications=justifications, + ) + return self.history + + def _verified(self, entry: dict[str, Any]) -> bool: + """Whether an entry's HMAC still matches its contents. + + A failure is reported, never acted on by discarding the entry -- see + :meth:`load`. + """ + if verify_entry_hmac(entry, key_file=self._key_file): + return True + _logger.error( + "escape history entry dated %r failed its HMAC check -- the file " + "has been edited outside the app, or the signing key is " + "unreadable. The entry still counts against the budget.", + entry.get("date", "?"), + ) + return False + + def save(self) -> bool: + """Write history back to disk. False if it could not be written.""" + payload = { + "used_days": self.history.used_days, + "justifications": self.history.justifications, + } + try: + self._path.parent.mkdir(parents=True, exist_ok=True) + self._path.write_text(json.dumps(payload, indent=2), encoding="utf-8") + except OSError: + _logger.exception("could not save escape history %s", self._path) + return False + return True + + def count_in_window(self, days: int, *, today: str | None = None) -> int: + """How many uses fall in the trailing ``days`` window.""" + today_str = today or _today_iso() + today_dt = _parse_iso(today_str) + if today_dt is None: + return 0 + cutoff = today_dt - timedelta(days=days) + count = 0 + for entry in self.history.used_days: + parsed = _parse_iso(entry) + if parsed is not None and cutoff < parsed <= today_dt: + count += 1 + return count + + def is_budget_exhausted(self, *, today: str | None = None) -> bool: + """Whether any rolling window has reached its limit. + + When True the app must not offer the hatch at all -- not offer it + greyed out, not offer it with a longer wait. Gone. + """ + policy = self._policy + return ( + self.count_in_window(7, today=today) >= policy.budget_per_7_days + or self.count_in_window(30, today=today) >= policy.budget_per_30_days + or self.count_in_window(90, today=today) >= policy.budget_per_90_days + ) + + def compute_lockout_seconds(self, *, today: str | None = None) -> int: + """Escalating wait: ``base * multiplier ** uses_in_30_days``.""" + recent = self.count_in_window(30, today=today) + multiplier: int = self._policy.lockout_multiplier_per_recent**recent + return int(self._policy.lockout_seconds * multiplier) + + def budget_summary(self, *, today: str | None = None) -> str: + """One-line summary of remaining budget, for the dialog.""" + policy = self._policy + week = self.count_in_window(7, today=today) + month = self.count_in_window(30, today=today) + quarter = self.count_in_window(90, today=today) + return ( + f"{policy.label}: {week}/{policy.budget_per_7_days}w · " + f"{month}/{policy.budget_per_30_days}m · " + f"{quarter}/{policy.budget_per_90_days}q" + ) + + def validate(self, draft: EscapeDraft) -> str | None: + """Return a user-facing complaint about ``draft``, or None if valid.""" + if not draft.reason.strip(): + return "Say what the problem is." + if not draft.onset.strip(): + return "Say when it started." + if not _MIN_SEVERITY <= draft.severity <= _MAX_SEVERITY: + return f"Severity must be between {_MIN_SEVERITY} and {_MAX_SEVERITY}." + minimum = self._policy.justification_min_chars + if len(draft.description.strip()) < minimum: + written = len(draft.description.strip()) + return f"Explain properly: {written}/{minimum} characters." + return None + + def record(self, draft: EscapeDraft, *, today: str | None = None) -> bool: + """Sign and append a use of the hatch. False if it could not be saved.""" + today_str = today or _today_iso() + entry: dict[str, Any] = { + "date": today_str, + "policy": self._policy.name, + "reason": draft.reason.strip(), + "onset": draft.onset.strip(), + "severity": draft.severity, + "description": draft.description.strip(), + } + signature = compute_entry_hmac(entry, key_file=self._key_file) + if signature is not None: + entry["hmac"] = signature + self.history.justifications.append(entry) + if today_str not in self.history.used_days: + self.history.used_days.append(today_str) + return self.save() + + def recent_justifications(self, count: int | None = None) -> list[dict[str, Any]]: + """The most recent justifications, newest last.""" + limit = count if count is not None else self._policy.history_review_count + return self.history.justifications[-limit:] + + def format_recent(self, count: int | None = None) -> str: + """Render past justifications for read-back in the dialog. + + Showing these is half the deterrent: a recycled excuse is obvious when + the last ten are on screen above the form. + """ + entries = self.recent_justifications(count) + if not entries: + return "(no previous uses)" + return "\n".join( + f"{entry.get('date', '?')} [{entry.get('severity', '?')}/10] " + f"{entry.get('reason', '')}: {entry.get('description', '')[:120]}" + for entry in entries + ) diff --git a/gatelock/gatelock/tests/test_escape.py b/gatelock/gatelock/tests/test_escape.py new file mode 100644 index 0000000..5503ab5 --- /dev/null +++ b/gatelock/gatelock/tests/test_escape.py @@ -0,0 +1,343 @@ +"""Tests for the shared escape hatch. + +The load-bearing property is that nothing here can ever *widen* the budget: +not a corrupt file, not an unparsable date, not a missing signing key. +""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING +from unittest.mock import patch + +import pytest + +from gatelock._escape import ( + EscapeDraft, + EscapePolicy, + EscapeTracker, + _parse_iso, + _today_iso, +) + +if TYPE_CHECKING: + from pathlib import Path + +TODAY = "2026-07-25" +POLICY = EscapePolicy(name="sick", label="Sick") + + +@pytest.fixture +def key_file(tmp_path: Path) -> Path: + """A throwaway signing key, so tests never touch the real root-owned one.""" + path = tmp_path / "hmac.key" + path.write_bytes(b"0" * 32) + return path + + +@pytest.fixture +def tracker(tmp_path: Path, key_file: Path) -> EscapeTracker: + """A tracker over an empty history.""" + return EscapeTracker(POLICY, tmp_path / "escape.json", key_file=key_file) + + +def good_draft() -> EscapeDraft: + """A justification that passes validation.""" + return EscapeDraft( + reason="migraine", onset="03:00", severity=8, description="x" * 130 + ) + + +class TestPolicyDefaults: + """The defaults mirror screen-locker's proven sick mode.""" + + def test_defaults(self) -> None: + """Budgets, lockout and justification length match sick mode.""" + assert POLICY.budget_per_7_days == 1 + assert POLICY.budget_per_30_days == 3 + assert POLICY.budget_per_90_days == 10 + assert POLICY.lockout_seconds == 120 + assert POLICY.lockout_multiplier_per_recent == 2 + assert POLICY.justification_min_chars == 120 + + +class TestParseIso: + """Date parsing never guesses.""" + + def test_valid(self) -> None: + """A well-formed date parses.""" + parsed = _parse_iso(TODAY) + assert parsed is not None + assert parsed.year == 2026 + + def test_invalid_returns_none(self) -> None: + """Garbage is ignored rather than crashing the lock.""" + assert _parse_iso("not-a-date") is None + + def test_today_iso_round_trips(self) -> None: + """The generated 'today' is parseable by the same parser.""" + assert _parse_iso(_today_iso()) is not None + + +class TestLoad: + """Reading history, and what must never happen while doing so.""" + + def test_missing_file(self, tracker: EscapeTracker) -> None: + """No file yet means no uses.""" + assert tracker.load().used_days == [] + + def test_corrupt_json(self, tracker: EscapeTracker, tmp_path: Path) -> None: + """A truncated file degrades instead of raising.""" + (tmp_path / "escape.json").write_text("{not json", encoding="utf-8") + assert tracker.load().used_days == [] + + def test_unreadable_file(self, tracker: EscapeTracker, tmp_path: Path) -> None: + """An unreadable file degrades instead of raising.""" + (tmp_path / "escape.json").write_text("{}", encoding="utf-8") + with patch("pathlib.Path.read_text", side_effect=OSError("denied")): + assert tracker.load().used_days == [] + + def test_non_dict_payload(self, tracker: EscapeTracker, tmp_path: Path) -> None: + """A JSON list is not a history.""" + (tmp_path / "escape.json").write_text("[]", encoding="utf-8") + assert tracker.load().used_days == [] + + def test_round_trip(self, tracker: EscapeTracker) -> None: + """A recorded use survives a reload.""" + tracker.record(good_draft(), today=TODAY) + reloaded = EscapeTracker(POLICY, tracker._path, key_file=tracker._key_file) + reloaded.load() + assert reloaded.history.used_days == [TODAY] + assert reloaded.tampered == 0 + + def test_non_dict_entries_are_skipped( + self, tracker: EscapeTracker, tmp_path: Path + ) -> None: + """A stray scalar in the justification list is ignored.""" + (tmp_path / "escape.json").write_text( + json.dumps({"used_days": [TODAY], "justifications": ["oops", 3]}), + encoding="utf-8", + ) + assert tracker.load().justifications == [] + + def test_non_string_days_are_skipped( + self, tracker: EscapeTracker, tmp_path: Path + ) -> None: + """A non-string date cannot count against the budget.""" + (tmp_path / "escape.json").write_text( + json.dumps({"used_days": [TODAY, 7], "justifications": []}), + encoding="utf-8", + ) + assert tracker.load().used_days == [TODAY] + + +class TestIntegrityNeverWidensTheBudget: + """The anti-bypass property of this module.""" + + def test_tampered_entry_is_kept_and_reported(self, tracker: EscapeTracker) -> None: + """Editing the file is shouted about -- and still counts against you.""" + tracker.record(good_draft(), today=TODAY) + raw = json.loads(tracker._path.read_text(encoding="utf-8")) + raw["justifications"][0]["description"] = "actually I just felt like it" + tracker._path.write_text(json.dumps(raw), encoding="utf-8") + + reloaded = EscapeTracker(POLICY, tracker._path, key_file=tracker._key_file) + reloaded.load() + + assert reloaded.tampered == 1 + # The crucial bit: the edit did NOT buy back any budget. + assert reloaded.history.used_days == [TODAY] + assert reloaded.is_budget_exhausted(today=TODAY) is True + + def test_unreadable_key_does_not_reset_the_budget( + self, tracker: EscapeTracker, tmp_path: Path + ) -> None: + """chmod-ing the signing key must not become an unlimited hatch.""" + tracker.record(good_draft(), today=TODAY) + missing = tmp_path / "gone.key" + + reloaded = EscapeTracker(POLICY, tracker._path, key_file=missing) + reloaded.load() + + assert reloaded.tampered == 1 + assert reloaded.is_budget_exhausted(today=TODAY) is True + + def test_unsigned_entry_still_counts( + self, tracker: EscapeTracker, tmp_path: Path + ) -> None: + """An entry with no signature at all is still a use.""" + (tmp_path / "escape.json").write_text( + json.dumps({"used_days": [TODAY], "justifications": [{"date": TODAY}]}), + encoding="utf-8", + ) + tracker.load() + assert tracker.tampered == 1 + assert tracker.is_budget_exhausted(today=TODAY) is True + + +class TestBudget: + """Rolling windows.""" + + def test_counts_only_inside_the_window(self, tracker: EscapeTracker) -> None: + """An old use falls out of the 7-day window.""" + tracker.history.used_days = ["2026-07-24", "2026-06-01"] + assert tracker.count_in_window(7, today=TODAY) == 1 + assert tracker.count_in_window(90, today=TODAY) == 2 + + def test_unparsable_today_counts_nothing(self, tracker: EscapeTracker) -> None: + """A broken 'today' cannot silently exhaust the budget.""" + tracker.history.used_days = [TODAY] + assert tracker.count_in_window(7, today="garbage") == 0 + + def test_unparsable_entry_is_skipped(self, tracker: EscapeTracker) -> None: + """A corrupt stored date is ignored, loudly.""" + tracker.history.used_days = ["nonsense", TODAY] + assert tracker.count_in_window(7, today=TODAY) == 1 + + @pytest.mark.parametrize( + ("days", "expected"), + [(["2026-07-25"], True), ([], False)], + ) + def test_week_window( + self, tracker: EscapeTracker, days: list[str], *, expected: bool + ) -> None: + """One use exhausts the weekly budget of 1.""" + tracker.history.used_days = days + assert tracker.is_budget_exhausted(today=TODAY) is expected + + def test_month_window(self, tracker: EscapeTracker) -> None: + """Three uses in 30 days exhausts, even if none is in the last week.""" + tracker.history.used_days = ["2026-07-05", "2026-07-06", "2026-07-07"] + assert tracker.is_budget_exhausted(today=TODAY) is True + + def test_quarter_window(self, tracker: EscapeTracker) -> None: + """Ten uses in 90 days exhausts, spread out enough to clear the others.""" + tracker.history.used_days = [f"2026-05-{day:02d}" for day in range(1, 11)] + assert tracker.is_budget_exhausted(today=TODAY) is True + + def test_nothing_used_is_not_exhausted(self, tracker: EscapeTracker) -> None: + """A clean history leaves the hatch available.""" + assert tracker.is_budget_exhausted(today=TODAY) is False + + +class TestLockout: + """The wait grows the more you lean on the hatch.""" + + @pytest.mark.parametrize( + ("recent", "seconds"), [(0, 120), (1, 240), (2, 480), (3, 960)] + ) + def test_doubles_per_recent_use( + self, tracker: EscapeTracker, recent: int, seconds: int + ) -> None: + """base * 2 ** uses_in_30_days.""" + tracker.history.used_days = [f"2026-07-{10 + i:02d}" for i in range(recent)] + assert tracker.compute_lockout_seconds(today=TODAY) == seconds + + +class TestValidation: + """What counts as a real justification.""" + + def test_accepts_a_proper_one(self, tracker: EscapeTracker) -> None: + """A complete draft passes.""" + assert tracker.validate(good_draft()) is None + + @pytest.mark.parametrize( + ("draft", "fragment"), + [ + (EscapeDraft("", "03:00", 5, "x" * 130), "what the problem is"), + (EscapeDraft("flu", "", 5, "x" * 130), "when it started"), + (EscapeDraft("flu", "03:00", 0, "x" * 130), "between 1 and 10"), + (EscapeDraft("flu", "03:00", 11, "x" * 130), "between 1 and 10"), + (EscapeDraft("flu", "03:00", 5, "short"), "Explain properly"), + ], + ) + def test_rejections( + self, tracker: EscapeTracker, draft: EscapeDraft, fragment: str + ) -> None: + """Each missing or lazy field is refused with a specific complaint.""" + message = tracker.validate(draft) + assert message is not None + assert fragment in message + + +class TestRecord: + """Writing a use down.""" + + def test_records_and_signs(self, tracker: EscapeTracker) -> None: + """A use is appended, dated and signed.""" + assert tracker.record(good_draft(), today=TODAY) is True + entry = tracker.history.justifications[-1] + assert entry["date"] == TODAY + assert entry["policy"] == "sick" + assert "hmac" in entry + + def test_unsigned_when_key_unavailable( + self, tracker: EscapeTracker, tmp_path: Path + ) -> None: + """Without a key the entry is still recorded, just unsigned.""" + tracker._key_file = tmp_path / "absent.key" + assert tracker.record(good_draft(), today=TODAY) is True + assert "hmac" not in tracker.history.justifications[-1] + + def test_same_day_twice_counts_once(self, tracker: EscapeTracker) -> None: + """Two uses on one day are one day of budget.""" + tracker.record(good_draft(), today=TODAY) + tracker.record(good_draft(), today=TODAY) + assert tracker.history.used_days == [TODAY] + assert len(tracker.history.justifications) == 2 + + def test_uses_today_when_unspecified(self, tracker: EscapeTracker) -> None: + """Omitting the date records the real today.""" + tracker.record(good_draft()) + assert tracker.history.used_days == [_today_iso()] + + def test_save_failure_is_reported(self, tracker: EscapeTracker) -> None: + """A read-only disk is surfaced, not swallowed.""" + with patch("pathlib.Path.write_text", side_effect=OSError("ro")): + assert tracker.record(good_draft(), today=TODAY) is False + + +class TestReadBack: + """Showing past excuses back to the user is half the deterrent.""" + + def test_empty(self, tracker: EscapeTracker) -> None: + """A clean history says so.""" + assert tracker.format_recent() == "(no previous uses)" + assert tracker.recent_justifications() == [] + + def test_lists_recent(self, tracker: EscapeTracker) -> None: + """Past entries are rendered with date, severity and reason.""" + tracker.record(good_draft(), today=TODAY) + rendered = tracker.format_recent() + assert TODAY in rendered + assert "8/10" in rendered + assert "migraine" in rendered + + def test_respects_explicit_limit(self, tracker: EscapeTracker) -> None: + """An explicit count overrides the policy default.""" + for day in range(1, 6): + tracker.record(good_draft(), today=f"2026-07-{day:02d}") + assert len(tracker.recent_justifications(2)) == 2 + + def test_handles_entries_missing_fields( + self, tracker: EscapeTracker, tmp_path: Path + ) -> None: + """A partial entry renders without raising.""" + (tmp_path / "escape.json").write_text( + json.dumps({"used_days": [], "justifications": [{}]}), encoding="utf-8" + ) + tracker.load() + assert "?" in tracker.format_recent() + + +class TestSummary: + """The one-line budget readout.""" + + def test_summary_shows_all_three_windows(self, tracker: EscapeTracker) -> None: + """Week, month and quarter counts all appear.""" + tracker.history.used_days = [TODAY] + assert tracker.budget_summary(today=TODAY) == "Sick: 1/1w · 1/3m · 1/10q" + + def test_policy_property(self, tracker: EscapeTracker) -> None: + """The tracker exposes the policy it enforces.""" + assert tracker.policy is POLICY