gatelock: extract the shared escape-hatch core
Some checks are pending
gatelock pre-commit / pre-commit (push) Waiting to run
gatelock tests / test (3.10) (push) Waiting to run
gatelock tests / test (3.11) (push) Waiting to run
gatelock tests / test (3.12) (push) Waiting to run

Every locker here already had a bounded way out -- screen-locker's sick
mode, diet-guard's log-a-meal -- except the wake alarm, which had only the
challenge itself. Rather than copy sick mode a third time, the mechanism
moves into gatelock so all three consumers share one implementation and one
set of budget rules.

`_escape.py` owns the parts that are policy-free: rolling 7/30/90-day
budgets, the justification draft and its validation, the escalating lockout
that doubles with each recent use, HMAC-signed history entries, and the
"last ten, played back to you" summary. App-specific policy (screen-locker's
commitment penalty and workout debt) stays in the consumer.

Entries that fail HMAC verification are KEPT rather than dropped. Dropping
them looks safer but is not: `verify_entry_hmac` also returns False when the
key is merely unreadable, so "discard what you cannot verify" would turn a
chmod on the key file into an unlimited escape hatch. They stay in the
budget count and are reported through a `tampered` counter instead.

278 tests, 100% branch coverage, no lint suppressions.
This commit is contained in:
Krzysztof Rudnicki 2026-07-25 15:35:20 +02:00
parent a9b65812d7
commit 7327c287f4
3 changed files with 653 additions and 0 deletions

View File

@ -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",

View File

@ -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
)

View File

@ -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