From db3ca83f292b241d97d5e0cec28444e79ef06fbc Mon Sep 17 00:00:00 2001 From: timmy Date: Wed, 19 Aug 2026 04:34:55 +0000 Subject: [PATCH] security: encrypt synchronized Today state (Closes #1112) --- src/main.py | 33 ++-- src/today_store.py | 332 +++++++++++++++++++++++++++++--------- tests/test_today_store.py | 106 ++++++++++++ 3 files changed, 380 insertions(+), 91 deletions(-) diff --git a/src/main.py b/src/main.py index fc8e3f0..2f95163 100644 --- a/src/main.py +++ b/src/main.py @@ -71,6 +71,7 @@ from src.security_event_store import SecurityEventStore, SecurityEventStoreError from src.suggestion_engine import compute from src.later_store import LaterStore from src.today_store import TodayPlanFull, TodaySessionConflict, TodayStore +from src.state_encryption import PrivateStateEncryptionError from src.views import FRONTEND_BUILD, router as frontend_router @@ -2367,7 +2368,7 @@ async def get_completed_filed_reviews(response: Response): login = await _confirmed_login() try: snapshot = await asyncio.to_thread(_completed_filed_review_store().get, login) - except (OSError, sqlite3.Error): + except (OSError, sqlite3.Error, PrivateStateEncryptionError): raise HTTPException( status_code=503, detail="Completed Filed review synchronization is unavailable", @@ -2388,7 +2389,7 @@ async def merge_completed_filed_reviews(payload: CompletedFiledReviewBatch): ) except ValueError as exc: raise HTTPException(status_code=422, detail=str(exc)) - except (OSError, sqlite3.Error): + except (OSError, sqlite3.Error, PrivateStateEncryptionError): raise HTTPException( status_code=503, detail="Completed Filed review synchronization is unavailable", @@ -2401,7 +2402,7 @@ async def get_saved_searches(response: Response): login = await _confirmed_login() try: snapshot = await asyncio.to_thread(_saved_search_store().get, login) - except (OSError, sqlite3.Error): + except (OSError, sqlite3.Error, PrivateStateEncryptionError): raise HTTPException( status_code=503, detail="Saved Search synchronization is unavailable", @@ -2431,7 +2432,7 @@ async def replace_saved_searches(payload: SavedSearchCollection): ) except ValueError as exc: raise HTTPException(status_code=422, detail=str(exc)) - except (OSError, sqlite3.Error): + except (OSError, sqlite3.Error, PrivateStateEncryptionError): raise HTTPException( status_code=503, detail="Saved Search synchronization is unavailable", @@ -2487,7 +2488,7 @@ async def get_today_plan(): login = await _confirmed_login() try: return await asyncio.to_thread(_today_store().get, login) - except (OSError, sqlite3.Error): + except (OSError, sqlite3.Error, PrivateStateEncryptionError): raise HTTPException( status_code=503, detail="Today synchronization is unavailable", @@ -2521,7 +2522,7 @@ async def update_today_plan(payload: TodayOperation | TodayOperationBatch): ) except TodayPlanFull: raise HTTPException(status_code=409, detail="Today is limited to 5 items") - except (OSError, sqlite3.Error): + except (OSError, sqlite3.Error, PrivateStateEncryptionError): raise HTTPException( status_code=503, detail="Today synchronization is unavailable", @@ -2534,7 +2535,7 @@ async def get_today_session(): login = await _confirmed_login() try: return await asyncio.to_thread(_today_store().get_session, login) - except (OSError, sqlite3.Error): + except (OSError, sqlite3.Error, PrivateStateEncryptionError): raise HTTPException( status_code=503, detail="Today session synchronization is unavailable", headers={"Retry-After": "1"}, @@ -2553,7 +2554,7 @@ async def update_today_session(payload: TodaySessionUpdate): status_code=409, detail={"code": "session_changed", "session": error.session}, ) - except (OSError, sqlite3.Error): + except (OSError, sqlite3.Error, PrivateStateEncryptionError): raise HTTPException( status_code=503, detail="Today session synchronization is unavailable", headers={"Retry-After": "1"}, @@ -2565,7 +2566,7 @@ async def get_today_recaps(response: Response): login = await _confirmed_login() try: recaps = await asyncio.to_thread(_today_store().list_recaps, login) - except (OSError, sqlite3.Error): + except (OSError, sqlite3.Error, PrivateStateEncryptionError): raise HTTPException( status_code=503, detail="Today recap history is unavailable", @@ -2587,7 +2588,7 @@ async def save_today_recap(payload: TodayRecap, response: Response): ) except ValueError as error: raise HTTPException(status_code=422, detail=str(error)) - except (OSError, sqlite3.Error): + except (OSError, sqlite3.Error, PrivateStateEncryptionError): raise HTTPException( status_code=503, detail="Today recap could not be saved", @@ -2639,7 +2640,7 @@ async def save_today_recap_and_log_time( raise ValueError("time log target must match the saved recap") except ValueError as error: raise HTTPException(status_code=422, detail=str(error)) - except (OSError, sqlite3.Error): + except (OSError, sqlite3.Error, PrivateStateEncryptionError): raise HTTPException(status_code=503, detail="Today recap could not be saved", headers={"Retry-After": "1"}) semaphore = asyncio.Semaphore(3) @@ -2708,7 +2709,7 @@ async def get_later_plan(): login = await _confirmed_login() try: return await asyncio.to_thread(_later_store().get, login) - except (OSError, sqlite3.Error): + except (OSError, sqlite3.Error, PrivateStateEncryptionError): raise HTTPException( status_code=503, detail="Later synchronization is unavailable", @@ -2738,7 +2739,7 @@ async def update_later_plan(payload: LaterOperation | LaterOperationBatch): ) except ValueError as error: raise HTTPException(status_code=422, detail=str(error)) - except (OSError, sqlite3.Error): + except (OSError, sqlite3.Error, PrivateStateEncryptionError): raise HTTPException( status_code=503, detail="Later synchronization is unavailable", @@ -3886,7 +3887,7 @@ async def _remove_notifications_from_live_snapshot(thread_ids: list[int]) -> Non _live_snapshot_value = updated try: await asyncio.to_thread(_live_snapshot_store.remove_notifications, read_ids) - except (OSError, sqlite3.Error): + except (OSError, sqlite3.Error, PrivateStateEncryptionError): # The upstream mutation already succeeded; keep process-local filtering. pass @@ -4066,7 +4067,7 @@ async def live_snapshot( events_revision=events_revision, notifications_revision=notifications_revision, ) - except (OSError, sqlite3.Error): + except (OSError, sqlite3.Error, PrivateStateEncryptionError): return JSONResponse( {"error": "Gitea live snapshot state is temporarily unavailable"}, status_code=503, @@ -4361,7 +4362,7 @@ async def defer_notification( status_code=503, headers={"Retry-After": "1"}, ) - except (OSError, sqlite3.Error): + except (OSError, sqlite3.Error, PrivateStateEncryptionError): return JSONResponse( {"error": "Later synchronization is unavailable. Please retry."}, status_code=503, diff --git a/src/today_store.py b/src/today_store.py index 0ac0784..039bd61 100644 --- a/src/today_store.py +++ b/src/today_store.py @@ -7,6 +7,7 @@ from datetime import date from pathlib import Path from src.private_state import connect_private_sqlite +from src.state_encryption import PrivateStateCipher, PrivateStateEncryptionError, private_state_encryption_key class TodayPlanFull(ValueError): @@ -31,6 +32,7 @@ class TodayStore: operation_limit: int = 4096, operation_retention_seconds: float = 30 * 24 * 60 * 60, recap_limit: int = 100, + encryption_key: bytes | None = None, clock=time.time, ): self.path = Path(path) @@ -40,6 +42,10 @@ class TodayStore: self.operation_retention_seconds = operation_retention_seconds self.recap_limit = recap_limit self.clock = clock + self._cipher = PrivateStateCipher( + encryption_key if encryption_key is not None else private_state_encryption_key(), + store="today", + ) self._initialize() def _initialize(self) -> None: @@ -162,31 +168,63 @@ class TodayStore: raise ValueError("login is required") return normalized - @staticmethod - def _snapshot(row) -> dict: + def _snapshot(self, row, login: str) -> tuple[dict, bool]: if row is None: - return {"revision": 0, "ids": [], "capacity_minutes": None, "estimates": {}} - ids = json.loads(row[1]) - estimates = json.loads(row[3] or "{}") + return {"revision": 0, "ids": [], "capacity_minutes": None, "estimates": {}}, False + if row[1].startswith("v1:"): + payload, legacy = self._cipher.open(row[1], binding=f"plan:{login}") + if not isinstance(payload, dict): + raise PrivateStateEncryptionError("private state could not be decrypted") + ids = payload.get("ids") + estimates = payload.get("estimates") + capacity_minutes = payload.get("capacity_minutes") + plan_date = payload.get("plan_date") + timezone = payload.get("timezone") + else: + legacy = True + ids = json.loads(row[1]) + estimates = json.loads(row[3] or "{}") + capacity_minutes = row[2] + plan_date = row[4] if len(row) > 4 else None + timezone = row[5] if len(row) > 5 else None + if not isinstance(ids, list) or not isinstance(estimates, dict): + raise PrivateStateEncryptionError("private state could not be decrypted") snapshot = { "revision": int(row[0]), "ids": ids, - "capacity_minutes": row[2], + "capacity_minutes": capacity_minutes, "estimates": {item_id: minutes for item_id, minutes in estimates.items() if item_id in ids}, } - if len(row) > 4 and row[4]: - snapshot["plan_date"] = row[4] - snapshot["timezone"] = row[5] - return snapshot + if plan_date: + snapshot["plan_date"] = plan_date + snapshot["timezone"] = timezone + return snapshot, legacy + + def _sealed_plan(self, login: str, snapshot: dict) -> str: + return self._cipher.seal({ + "ids": snapshot["ids"], + "capacity_minutes": snapshot["capacity_minutes"], + "estimates": snapshot["estimates"], + "plan_date": snapshot.get("plan_date"), + "timezone": snapshot.get("timezone"), + }, binding=f"plan:{login}") def get(self, login: str) -> dict: + login = self._normalize_login(login) with self._connect() as connection: row = connection.execute( "SELECT revision, ids, capacity_minutes, estimates, plan_date, timezone " "FROM today_plans WHERE login = ?", - (self._normalize_login(login),), + (login,), ).fetchone() - return self._snapshot(row) + snapshot, legacy = self._snapshot(row, login) + if row is not None and legacy: + connection.execute( + "UPDATE today_plans SET ids = ?, capacity_minutes = NULL, estimates = '{}', " + "plan_date = NULL, timezone = NULL WHERE login = ? AND ids = ?", + (self._sealed_plan(login, snapshot), login, row[1]), + ) + return snapshot @staticmethod def _empty_session() -> dict: @@ -195,20 +233,42 @@ class TodayStore: "elapsed_ms": 0, "running": False, "break_deadline_at": None, "updated_at": None, } - def get_session(self, login: str) -> dict: - with self._connect() as connection: - row = connection.execute( - "SELECT revision, device_id, identity, elapsed_ms, running, break_deadline_at, updated_at " - "FROM today_sessions WHERE login = ?", - (self._normalize_login(login),), - ).fetchone() + def _session_snapshot(self, row, login: str) -> tuple[dict, bool]: if row is None: - return self._empty_session() + return self._empty_session(), False + if row[1].startswith("v1:"): + payload, legacy = self._cipher.open(row[1], binding=f"session:{login}") + if not isinstance(payload, dict): + raise PrivateStateEncryptionError("private state could not be decrypted") + return {"revision": int(row[0]), **payload, "updated_at": row[6]}, legacy return { "revision": int(row[0]), "device_id": row[1], "identity": row[2], "elapsed_ms": int(row[3]), "running": bool(row[4]), "break_deadline_at": row[5], "updated_at": row[6], - } + }, True + + def _sealed_session(self, login: str, session: dict) -> str: + return self._cipher.seal({ + "device_id": session["device_id"], "identity": session["identity"], + "elapsed_ms": session["elapsed_ms"], "running": session["running"], + "break_deadline_at": session["break_deadline_at"], + }, binding=f"session:{login}") + + def get_session(self, login: str) -> dict: + login = self._normalize_login(login) + with self._connect() as connection: + row = connection.execute( + "SELECT revision, device_id, identity, elapsed_ms, running, break_deadline_at, updated_at " + "FROM today_sessions WHERE login = ?", (login,), + ).fetchone() + session, legacy = self._session_snapshot(row, login) + if row is not None and legacy: + connection.execute( + "UPDATE today_sessions SET device_id = ?, identity = '', elapsed_ms = 0, " + "running = 0, break_deadline_at = NULL WHERE login = ? AND device_id = ?", + (self._sealed_session(login, session), login, row[1]), + ) + return session def update_session( self, login: str, *, base_revision: int, device_id: str, @@ -222,24 +282,25 @@ class TodayStore: "SELECT revision, device_id, identity, elapsed_ms, running, break_deadline_at, updated_at " "FROM today_sessions WHERE login = ?", (login,) ).fetchone() - current_revision = int(current[0]) if current else 0 + current_session, _legacy = self._session_snapshot(current, login) + current_revision = current_session["revision"] if base_revision != current_revision: - session = self._empty_session() if current is None else { - "revision": current_revision, "device_id": current[1], "identity": current[2], - "elapsed_ms": int(current[3]), "running": bool(current[4]), - "break_deadline_at": current[5], "updated_at": current[6], - } - raise TodaySessionConflict(session) + raise TodaySessionConflict(current_session) revision = current_revision + 1 + session = { + "revision": revision, "device_id": device_id, "identity": identity, + "elapsed_ms": elapsed_ms, "running": bool(running), + "break_deadline_at": break_deadline_at, "updated_at": updated_at, + } connection.execute( "INSERT INTO today_sessions(login, revision, device_id, identity, elapsed_ms, running, break_deadline_at, updated_at) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?) " "ON CONFLICT(login) DO UPDATE SET revision=excluded.revision, device_id=excluded.device_id, " "identity=excluded.identity, elapsed_ms=excluded.elapsed_ms, running=excluded.running, " "break_deadline_at=excluded.break_deadline_at, updated_at=excluded.updated_at", - (login, revision, device_id, identity, elapsed_ms, int(running), break_deadline_at, updated_at), + (login, revision, self._sealed_session(login, session), "", 0, 0, None, updated_at), ) - return self.get_session(login) + return session @staticmethod def _normalize_recap_items(items: list[dict]) -> list[dict]: @@ -273,9 +334,24 @@ class TodayStore: }) return normalized - @staticmethod - def _recap_snapshot(session_id: str, created_at: float, serialized: str) -> dict: - items = json.loads(serialized) + def _recap_snapshot( + self, login: str, encrypted_session_id: str, created_at: float, serialized: str + ) -> tuple[dict, bool]: + if encrypted_session_id.startswith("v1:"): + session_id, legacy_session = self._cipher.open( + encrypted_session_id, binding=f"recap-id:{login}" + ) + items, legacy_items = self._cipher.open( + serialized, binding=f"recap-items:{login}:{session_id}" + ) + else: + session_id, legacy_session = encrypted_session_id, True + try: + items, legacy_items = json.loads(serialized), True + except json.JSONDecodeError as error: + raise PrivateStateEncryptionError("private state could not be decrypted") from error + if not isinstance(session_id, str) or not isinstance(items, list): + raise PrivateStateEncryptionError("private state could not be decrypted") estimated = sum(item["estimate_minutes"] for item in items if item["estimate_minutes"] is not None) actual = sum(item["actual_minutes"] for item in items) return { @@ -285,7 +361,20 @@ class TodayStore: "estimated_minutes": estimated, "actual_minutes": actual, "variance_minutes": actual - estimated, - } + }, legacy_session or legacy_items + + def _migrate_recap(self, connection, login: str, stored_session: str, snapshot: dict) -> None: + encrypted_session = self._cipher.seal( + snapshot["session_id"], binding=f"recap-id:{login}" + ) + encrypted_items = self._cipher.seal( + snapshot["items"], binding=f"recap-items:{login}:{snapshot['session_id']}" + ) + connection.execute( + "UPDATE today_recaps SET session_id = ?, items = ? " + "WHERE login = ? AND session_id = ?", + (encrypted_session, encrypted_items, login, stored_session), + ) def save_recap(self, login: str, session_id: str, items: list[dict]) -> dict: login = self._normalize_login(login) @@ -293,19 +382,25 @@ class TodayStore: raise ValueError("session_id is required and bounded") session_id = session_id.strip() normalized = self._normalize_recap_items(items) - serialized = json.dumps(normalized, separators=(",", ":")) with self._connect() as connection: connection.execute("BEGIN IMMEDIATE") - existing = connection.execute( - "SELECT created_at, items FROM today_recaps WHERE login = ? AND session_id = ?", - (login, session_id), - ).fetchone() - if existing is not None: - return self._recap_snapshot(session_id, existing[0], existing[1]) + existing_rows = connection.execute( + "SELECT session_id, created_at, items FROM today_recaps WHERE login = ?", (login,) + ).fetchall() + for existing in existing_rows: + snapshot, legacy = self._recap_snapshot(login, *existing) + if snapshot["session_id"] == session_id: + if legacy: + self._migrate_recap(connection, login, existing[0], snapshot) + return snapshot created_at = self.clock() + encrypted_session_id = self._cipher.seal(session_id, binding=f"recap-id:{login}") + serialized = self._cipher.seal( + normalized, binding=f"recap-items:{login}:{session_id}" + ) connection.execute( "INSERT INTO today_recaps(login, session_id, created_at, items) VALUES (?, ?, ?, ?)", - (login, session_id, created_at, serialized), + (login, encrypted_session_id, created_at, serialized), ) connection.execute( "DELETE FROM today_recaps WHERE login = ? AND rowid NOT IN " @@ -313,53 +408,135 @@ class TodayStore: "ORDER BY created_at DESC, rowid DESC LIMIT ?)", (login, login, self.recap_limit), ) - return self._recap_snapshot(session_id, created_at, serialized) + return self._recap_snapshot(login, encrypted_session_id, created_at, serialized)[0] def list_recaps(self, login: str, *, limit: int = 30) -> list[dict]: bounded_limit = max(1, min(int(limit), self.recap_limit, 100)) + login = self._normalize_login(login) with self._connect() as connection: rows = connection.execute( "SELECT session_id, created_at, items FROM today_recaps WHERE login = ? " "ORDER BY created_at DESC, rowid DESC LIMIT ?", - (self._normalize_login(login), bounded_limit), + (login, bounded_limit), ).fetchall() - return [self._recap_snapshot(*row) for row in rows] + snapshots = [] + for row in rows: + snapshot, legacy = self._recap_snapshot(login, *row) + if legacy: + self._migrate_recap(connection, login, row[0], snapshot) + snapshots.append(snapshot) + return snapshots + + def _time_log_snapshot(self, login: str, row) -> tuple[dict, bool]: + encrypted_session, encrypted_identity, payload_value, status_value = row + if encrypted_session.startswith("v1:"): + session_id = self._cipher.open( + encrypted_session, binding=f"time-log-session:{login}" + )[0] + identity = self._cipher.open( + encrypted_identity, binding=f"time-log-identity:{login}:{session_id}" + )[0] + payload, legacy = self._cipher.open( + str(payload_value), binding=f"time-log-payload:{login}:{session_id}:{identity}" + ) + if not isinstance(session_id, str) or not isinstance(identity, str) or not isinstance(payload, dict): + raise PrivateStateEncryptionError("private state could not be decrypted") + return { + "session_id": session_id, "identity": identity, + "actual_minutes": payload.get("actual_minutes"), "status": payload.get("status"), + "stored_session": encrypted_session, "stored_identity": encrypted_identity, + }, legacy + return { + "session_id": encrypted_session, "identity": encrypted_identity, + "actual_minutes": payload_value, "status": status_value, + "stored_session": encrypted_session, "stored_identity": encrypted_identity, + }, True + + def _sealed_time_log(self, login: str, session_id: str, identity: str, actual: int, status: str) -> tuple[str, str, str]: + return ( + self._cipher.seal(session_id, binding=f"time-log-session:{login}"), + self._cipher.seal(identity, binding=f"time-log-identity:{login}:{session_id}"), + self._cipher.seal( + {"actual_minutes": actual, "status": status}, + binding=f"time-log-payload:{login}:{session_id}:{identity}", + ), + ) + + def _migrate_time_log(self, connection, login: str, snapshot: dict) -> dict: + sealed_session, sealed_identity, sealed_payload = self._sealed_time_log( + login, snapshot["session_id"], snapshot["identity"], + snapshot["actual_minutes"], snapshot["status"], + ) + connection.execute( + "UPDATE today_time_logs SET session_id = ?, identity = ?, actual_minutes = ?, status = 'sealed' " + "WHERE login = ? AND session_id = ? AND identity = ?", + (sealed_session, sealed_identity, sealed_payload, login, + snapshot["stored_session"], snapshot["stored_identity"]), + ) + return {**snapshot, "stored_session": sealed_session, "stored_identity": sealed_identity} def begin_time_log(self, login: str, session_id: str, identity: str, actual_minutes: int) -> str: """Claim one recap item for upstream logging, returning its current state.""" login = self._normalize_login(login) with self._connect() as connection: connection.execute("BEGIN IMMEDIATE") - row = connection.execute( - "SELECT actual_minutes, status FROM today_time_logs " - "WHERE login = ? AND session_id = ? AND identity = ?", - (login, session_id, identity), - ).fetchone() - if row is None: + rows = connection.execute( + "SELECT session_id, identity, actual_minutes, status FROM today_time_logs WHERE login = ?", + (login,), + ).fetchall() + match = None + for row in rows: + candidate, legacy = self._time_log_snapshot(login, row) + if candidate["session_id"] == session_id and candidate["identity"] == identity: + match = self._migrate_time_log(connection, login, candidate) if legacy else candidate + break + if match is None: + sealed_session, sealed_identity, sealed_payload = self._sealed_time_log( + login, session_id, identity, actual_minutes, "pending" + ) connection.execute( "INSERT INTO today_time_logs(login, session_id, identity, actual_minutes, status) " - "VALUES (?, ?, ?, ?, 'pending')", - (login, session_id, identity, actual_minutes), + "VALUES (?, ?, ?, ?, 'sealed')", + (login, sealed_session, sealed_identity, sealed_payload), ) return "claimed" - if row[0] != actual_minutes: + if match["actual_minutes"] != actual_minutes: raise ValueError("logged recap time cannot be changed") - if row[1] == "failed": + if match["status"] == "failed": + sealed_payload = self._sealed_time_log( + login, session_id, identity, actual_minutes, "pending" + )[2] connection.execute( - "UPDATE today_time_logs SET status = 'pending' " + "UPDATE today_time_logs SET actual_minutes = ?, status = 'sealed' " "WHERE login = ? AND session_id = ? AND identity = ?", - (login, session_id, identity), + (sealed_payload, login, match["stored_session"], match["stored_identity"]), ) return "claimed" - return row[1] + return match["status"] def finish_time_log(self, login: str, session_id: str, identity: str, *, succeeded: bool) -> None: + login = self._normalize_login(login) with self._connect() as connection: - connection.execute( - "UPDATE today_time_logs SET status = ? " - "WHERE login = ? AND session_id = ? AND identity = ? AND status = 'pending'", - ("logged" if succeeded else "failed", self._normalize_login(login), session_id, identity), - ) + rows = connection.execute( + "SELECT session_id, identity, actual_minutes, status FROM today_time_logs WHERE login = ?", + (login,), + ).fetchall() + for row in rows: + match, legacy = self._time_log_snapshot(login, row) + if match["session_id"] != session_id or match["identity"] != identity or match["status"] != "pending": + continue + if legacy: + match = self._migrate_time_log(connection, login, match) + sealed_payload = self._sealed_time_log( + login, session_id, identity, match["actual_minutes"], + "logged" if succeeded else "failed", + )[2] + connection.execute( + "UPDATE today_time_logs SET actual_minutes = ?, status = 'sealed' " + "WHERE login = ? AND session_id = ? AND identity = ?", + (sealed_payload, login, match["stored_session"], match["stored_identity"]), + ) + return def apply( self, @@ -384,7 +561,7 @@ class TodayStore: "SELECT revision, ids, capacity_minutes, estimates, plan_date, timezone " "FROM today_plans WHERE login = ?", (login,) ).fetchone() - snapshot = self._snapshot(row) + snapshot, _legacy = self._snapshot(row, login) duplicate = connection.execute( "SELECT 1 FROM today_operations WHERE login = ? AND operation_id = ?", (login, operation_id), @@ -417,19 +594,22 @@ class TodayStore: changed = True revision = snapshot["revision"] + (1 if changed else 0) - serialized_ids = json.dumps(ids, separators=(",", ":")) - serialized_estimates = json.dumps(estimates, separators=(",", ":")) + sealed = self._sealed_plan(login, { + "ids": ids, "capacity_minutes": snapshot["capacity_minutes"], + "estimates": estimates, "plan_date": snapshot.get("plan_date"), + "timezone": snapshot.get("timezone"), + }) if row is None: connection.execute( "INSERT INTO today_plans(login, revision, ids, capacity_minutes, estimates, plan_date, timezone) " "VALUES (?, ?, ?, ?, ?, ?, ?)", - (login, revision, serialized_ids, snapshot["capacity_minutes"], serialized_estimates, - snapshot.get("plan_date"), snapshot.get("timezone")), + (login, revision, sealed, None, "{}", None, None), ) elif changed: connection.execute( - "UPDATE today_plans SET revision = ?, ids = ?, estimates = ? WHERE login = ?", - (revision, serialized_ids, serialized_estimates, login), + "UPDATE today_plans SET revision = ?, ids = ?, capacity_minutes = NULL, " + "estimates = '{}', plan_date = NULL, timezone = NULL WHERE login = ?", + (revision, sealed, login), ) self._record_operation(connection, login, operation_id) result = { @@ -452,7 +632,7 @@ class TodayStore: "SELECT revision, ids, capacity_minutes, estimates, plan_date, timezone " "FROM today_plans WHERE login = ?", (login,) ).fetchone() - snapshot = self._snapshot(row) + snapshot, _legacy = self._snapshot(row, login) ids = list(snapshot["ids"]) capacity_minutes = snapshot["capacity_minutes"] estimates = dict(snapshot["estimates"]) @@ -581,19 +761,21 @@ class TodayStore: self._record_operation(connection, login, operation_id) accepted.append(operation_id) - serialized = json.dumps(ids, separators=(",", ":")) - serialized_estimates = json.dumps(estimates, separators=(",", ":")) + sealed = self._sealed_plan(login, { + "ids": ids, "capacity_minutes": capacity_minutes, "estimates": estimates, + "plan_date": plan_date, "timezone": timezone, + }) if row is None: connection.execute( "INSERT INTO today_plans(login, revision, ids, capacity_minutes, estimates, plan_date, timezone) " "VALUES (?, ?, ?, ?, ?, ?, ?)", - (login, revision, serialized, capacity_minutes, serialized_estimates, plan_date, timezone), + (login, revision, sealed, None, "{}", None, None), ) elif accepted: connection.execute( "UPDATE today_plans SET revision = ?, ids = ?, capacity_minutes = ?, estimates = ?, " "plan_date = ?, timezone = ? WHERE login = ?", - (revision, serialized, capacity_minutes, serialized_estimates, plan_date, timezone, login), + (revision, sealed, None, "{}", None, None, login), ) result = { "revision": revision, diff --git a/tests/test_today_store.py b/tests/test_today_store.py index 90034f3..9acbfc4 100644 --- a/tests/test_today_store.py +++ b/tests/test_today_store.py @@ -4,9 +4,115 @@ import pytest import httpx from src import main +from src.state_encryption import PrivateStateEncryptionError from src.today_store import TodayPlanFull, TodaySessionConflict, TodayStore +def test_task_bearing_today_journey_is_encrypted_at_rest_and_survives_restart(tmp_path): + path = tmp_path / "today.sqlite3" + key = b"t" * 32 + store = TodayStore(path, encryption_key=key, clock=lambda: 1234.5) + issue = "issue:private-roadmap/launch:4242:" + device = "private-phone-canary" + recap_session = "private-recap-canary" + + plan = store.apply_batch("Timmy", [ + {"operation_id": "add", "action": "add", "item_id": issue}, + {"operation_id": "configure", "action": "configure", "item_id": "plan", + "capacity_minutes": 180, "estimates": {issue: 75}}, + ]) + session = store.update_session( + "timmy", base_revision=0, device_id=device, identity=issue, + elapsed_ms=90_000, running=True, + ) + recap = store.save_recap("timmy", recap_session, [ + {"identity": issue, "estimate_minutes": 75, "actual_minutes": 91} + ]) + assert store.begin_time_log("timmy", recap_session, issue, 91) == "claimed" + store.finish_time_log("timmy", recap_session, issue, succeeded=True) + + retained = b"".join( + candidate.read_bytes() + for candidate in (path, path.with_name(path.name + "-wal")) + if candidate.exists() + ) + for canary in (issue, device, recap_session): + assert canary.encode() not in retained + assert retained.count(b"v1:") >= 4 + + reopened = TodayStore(path, encryption_key=key) + assert reopened.get("timmy") == { + key: value for key, value in plan.items() + if key not in {"accepted_operation_ids", "duplicate_operation_ids", "rejected_operations"} + } + assert reopened.get_session("timmy") == session + assert reopened.list_recaps("timmy") == [recap] + assert reopened.begin_time_log("timmy", recap_session, issue, 91) == "logged" + + +def test_legacy_today_rows_migrate_on_use_without_changing_behavior(tmp_path): + path = tmp_path / "today.sqlite3" + key = b"m" * 32 + store = TodayStore(path, encryption_key=key) + issue = "issue:legacy-private/project:7:" + with sqlite3.connect(path) as connection: + connection.execute( + "INSERT INTO today_plans VALUES (?, ?, ?, ?, ?, ?, ?)", + ("timmy", 4, f'["{issue}"]', 120, f'{{"{issue}":45}}', "2026-08-19", "UTC"), + ) + connection.execute( + "INSERT INTO today_sessions VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + ("timmy", 3, "legacy-phone", issue, 5000, 1, None, 99.0), + ) + connection.execute( + "INSERT INTO today_recaps VALUES (?, ?, ?, ?)", + ("timmy", "legacy-session", 88.0, + f'[{ {"identity": issue, "estimate_minutes": 45, "actual_minutes": 50} }]'.replace("'", '"')), + ) + connection.execute( + "INSERT INTO today_time_logs VALUES (?, ?, ?, ?, ?)", + ("timmy", "legacy-session", issue, 50, "pending"), + ) + + assert store.get("timmy")["revision"] == 4 + assert store.get_session("timmy")["revision"] == 3 + assert store.list_recaps("timmy")[0]["items"][0]["identity"] == issue + assert store.begin_time_log("timmy", "legacy-session", issue, 50) == "pending" + + retained = path.read_bytes() + for canary in (issue, "legacy-phone", "legacy-session"): + assert canary.encode() not in retained + + +@pytest.mark.anyio +async def test_tampered_or_account_substituted_today_state_fails_closed_with_retry(monkeypatch, tmp_path): + path = tmp_path / "today.sqlite3" + store = TodayStore(path, encryption_key=b"a" * 32) + store.apply("timmy", "seed", "add", "issue:private/repo:9:") + with sqlite3.connect(path) as connection: + payload = connection.execute( + "SELECT ids FROM today_plans WHERE login = 'timmy'" + ).fetchone()[0] + connection.execute( + "INSERT INTO today_plans(login, revision, ids, estimates) VALUES ('alexander', 1, ?, '{}')", + (payload,), + ) + + with pytest.raises(PrivateStateEncryptionError): + store.get("alexander") + + async def user(): + return {"login": "alexander"} + + monkeypatch.setattr(main, "current_user", user) + monkeypatch.setattr(main, "_today_store", lambda: store) + with pytest.raises(main.HTTPException) as raised: + await main.get_today_plan() + assert raised.value.status_code == 503 + assert raised.value.detail == "Today synchronization is unavailable" + assert raised.value.headers == {"Retry-After": "1"} + + def test_active_session_is_durable_and_account_scoped(tmp_path): path = tmp_path / "today.sqlite3" store = TodayStore(path, clock=lambda: 1234.5)