From 03f0dd14f0dcf2a1599c203f72e3a55740e9f581 Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Thu, 19 Feb 2026 00:01:49 +0000 Subject: [PATCH] Implement database file locking for extra safety. --- paskia/__main__.py | 3 +- paskia/db/__init__.py | 10 +- paskia/db/filelock.py | 247 ++++++++++++++++++++++++++++++++++++++++++ paskia/db/jsonl.py | 105 +++++++++--------- pyproject.toml | 1 - 5 files changed, 298 insertions(+), 68 deletions(-) create mode 100644 paskia/db/filelock.py diff --git a/paskia/__main__.py b/paskia/__main__.py index dd7e282..7574121 100644 --- a/paskia/__main__.py +++ b/paskia/__main__.py @@ -1,5 +1,4 @@ import argparse -import asyncio import json import logging import os @@ -102,7 +101,7 @@ def main(): # Read-only load to get stored config (no writes, no global state) db_path = os.environ.get("PASKIA_DB", f"{args.rp_id}.paskiadb") - stored_db = asyncio.run(load_readonly(db_path, rp_id=args.rp_id)) + stored_db = load_readonly(db_path, rp_id=args.rp_id) stored_config = stored_db.config # Apply defaults from stored config diff --git a/paskia/db/__init__.py b/paskia/db/__init__.py index 9b37c94..0f0cc21 100644 --- a/paskia/db/__init__.py +++ b/paskia/db/__init__.py @@ -1,7 +1,7 @@ """ Database module for WebAuthn passkey authentication. -Read: Access data() directly, use build_* to convert to public structs. +Read: Access data() directly for structs. CTX: data().session_ctx(key) returns SessionContext with effective permissions. Write: Functions validate and commit, or raise ValueError. @@ -10,7 +10,6 @@ Usage: # Read (after init) user_data = db.data().users[user_uuid] - user = db.build_user(user_uuid) # Context ctx = db.data().session_ctx(session_key) @@ -109,13 +108,6 @@ __all__ = [ "stop_background", "start_cleanup", "stop_cleanup", - # Builders - "build_credential", - "build_permission", - "build_reset_token", - "build_role", - "build_session", - "build_user", # Read ops # Write ops "add_permission_to_org", diff --git a/paskia/db/filelock.py b/paskia/db/filelock.py new file mode 100644 index 0000000..874f1f9 --- /dev/null +++ b/paskia/db/filelock.py @@ -0,0 +1,247 @@ +"""Cross-platform locked file for the database (no separate .lock files). + +Unix: open() + fcntl.flock (advisory, cooperative among processes that flock). +Windows: CreateFileW with FILE_SHARE_READ (OS-enforced, allows readers, blocks writers). + +A single file descriptor is opened once for both reading and writing. +The lock is acquired atomically (on Windows) or immediately after open (on Unix), +and the same descriptor is used for the lifetime of the process: first to read +the existing content, then to append new writes. +""" + +import logging +import os +import sys +from pathlib import Path + +_logger = logging.getLogger(__name__) + + +def _fatal(msg: str) -> None: + """Log a fatal error and exit immediately, bypassing exception handlers.""" + _logger.critical(msg) + os._exit(1) + + +if sys.platform == "win32": + import ctypes + from ctypes import wintypes + + _kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) + + _GENERIC_READ = 0x80000000 + _GENERIC_WRITE = 0x40000000 + _FILE_SHARE_READ = 0x00000001 + _OPEN_EXISTING = 3 + _OPEN_ALWAYS = 4 + _FILE_ATTRIBUTE_NORMAL = 0x80 + _FILE_BEGIN = 0 + _FILE_END = 2 + _ERROR_SHARING_VIOLATION = 32 + _INVALID_FILE_SIZE = 0xFFFFFFFF + + _kernel32.CreateFileW.restype = wintypes.HANDLE + _kernel32.CreateFileW.argtypes = [ + wintypes.LPCWSTR, + wintypes.DWORD, + wintypes.DWORD, + ctypes.c_void_p, + wintypes.DWORD, + wintypes.DWORD, + wintypes.HANDLE, + ] + _kernel32.ReadFile.restype = wintypes.BOOL + _kernel32.ReadFile.argtypes = [ + wintypes.HANDLE, + ctypes.c_void_p, + wintypes.DWORD, + ctypes.POINTER(wintypes.DWORD), + ctypes.c_void_p, + ] + _kernel32.WriteFile.restype = wintypes.BOOL + _kernel32.WriteFile.argtypes = [ + wintypes.HANDLE, + ctypes.c_void_p, + wintypes.DWORD, + ctypes.POINTER(wintypes.DWORD), + ctypes.c_void_p, + ] + _kernel32.GetFileSize.restype = wintypes.DWORD + _kernel32.GetFileSize.argtypes = [ + wintypes.HANDLE, + ctypes.POINTER(wintypes.DWORD), + ] + _kernel32.SetFilePointer.restype = wintypes.DWORD + _kernel32.SetFilePointer.argtypes = [ + wintypes.HANDLE, + wintypes.LONG, + ctypes.POINTER(wintypes.LONG), + wintypes.DWORD, + ] + _kernel32.CloseHandle.restype = wintypes.BOOL + _kernel32.CloseHandle.argtypes = [wintypes.HANDLE] + + def _is_invalid_handle(handle) -> bool: + return ctypes.c_void_p(handle).value == ctypes.c_void_p(-1).value + +else: + import fcntl + + +class LockedFile: + """A file opened with an exclusive write lock. + + Usage:: + + f = LockedFile() + f.open(path) # open + lock (read+write) + content = f.read() # read entire content + f.write(data) # append data (seeks to end first) + f.close() # release lock + close fd + + Unix: fcntl.flock (advisory) — read-only callers that don't flock are unaffected. + Windows: CreateFileW with FILE_SHARE_READ — OS blocks other writers. + """ + + def __init__(self) -> None: + self._fd: int | None = None # Unix fd or Windows HANDLE + + def open(self, path: Path, *, create: bool = False) -> None: + """Open *path* for read+write with an exclusive lock. + + Args: + path: File to open and lock. + create: If True, create the file if it doesn't exist (bootstrap). + + Raises: + SystemExit: If the file is locked by another process or not found. + """ + if self._fd is not None: + return # Already open (idempotent) + + if sys.platform == "win32": + self._open_win32(path, create) + else: + self._open_unix(path, create) + + def open_and_read(self, path: Path) -> bytes: + """Open *path* with exclusive lock and read all content. + + Combined operation for efficient use with asyncio.to_thread(). + """ + self.open(path) + return self.read() + + def read(self) -> bytes: + """Read the entire file content from the beginning.""" + if self._fd is None: + raise RuntimeError("LockedFile.read() called on a closed file") + + if sys.platform == "win32": + return self._read_win32() + else: + return self._read_unix() + + def write(self, data: bytes) -> None: + """Append *data* to the end of the file.""" + if self._fd is None: + raise RuntimeError("LockedFile.write() called on a closed file") + + if sys.platform == "win32": + self._write_win32(data) + else: + self._write_unix(data) + + def close(self) -> None: + """Release the lock and close the file.""" + if self._fd is None: + return + if sys.platform == "win32": + _kernel32.CloseHandle(self._fd) + else: + os.close(self._fd) + self._fd = None + + @property + def is_open(self) -> bool: + return self._fd is not None + + # -- Unix ---------------------------------------------------------------- + + def _open_unix(self, path: Path, create: bool) -> None: + flags = os.O_RDWR | (os.O_CREAT if create else 0) + try: + fd = os.open(path, flags, 0o666) + except FileNotFoundError: + _fatal(f"Database file not found: {path.resolve()}") + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError: + os.close(fd) + _fatal(f"🛑 {path.resolve()}: database already locked by another instance") + self._fd = fd + + def _read_unix(self) -> bytes: + os.lseek(self._fd, 0, os.SEEK_SET) + chunks = [] + while True: + chunk = os.read(self._fd, 1 << 20) # 1 MiB + if not chunk: + break + chunks.append(chunk) + return b"".join(chunks) + + def _write_unix(self, data: bytes) -> None: + os.lseek(self._fd, 0, os.SEEK_END) + os.write(self._fd, data) + + # -- Windows ------------------------------------------------------------- + + def _open_win32(self, path: Path, create: bool) -> None: + disposition = _OPEN_ALWAYS if create else _OPEN_EXISTING + handle = _kernel32.CreateFileW( + str(path), + _GENERIC_READ | _GENERIC_WRITE, + _FILE_SHARE_READ, + None, + disposition, + _FILE_ATTRIBUTE_NORMAL, + None, + ) + if _is_invalid_handle(handle): + err = ctypes.get_last_error() + if err == _ERROR_SHARING_VIOLATION: + _fatal( + f"🛑 {path.resolve()}: database already locked by another instance" + ) + _fatal(f"Failed to open database {path.resolve()}: Windows error {err}") + self._fd = handle + + def _read_win32(self) -> bytes: + _kernel32.SetFilePointer(self._fd, 0, None, _FILE_BEGIN) + size = _kernel32.GetFileSize(self._fd, None) + if size == _INVALID_FILE_SIZE: + raise OSError( + f"GetFileSize failed: Windows error {ctypes.get_last_error()}" + ) + if size == 0: + return b"" + buf = ctypes.create_string_buffer(size) + bytes_read = wintypes.DWORD() + ok = _kernel32.ReadFile(self._fd, buf, size, ctypes.byref(bytes_read), None) + if not ok: + raise OSError(f"ReadFile failed: Windows error {ctypes.get_last_error()}") + return buf.raw[: bytes_read.value] + + def _write_win32(self, data: bytes) -> None: + _kernel32.SetFilePointer(self._fd, 0, None, _FILE_END) + written = wintypes.DWORD() + ok = _kernel32.WriteFile( + self._fd, + data, + len(data), + ctypes.byref(written), + None, + ) + if not ok: + raise OSError(f"WriteFile failed: Windows error {ctypes.get_last_error()}") diff --git a/paskia/db/jsonl.py b/paskia/db/jsonl.py index fbb3e42..05327bc 100644 --- a/paskia/db/jsonl.py +++ b/paskia/db/jsonl.py @@ -2,6 +2,7 @@ JSONL persistence layer for the database. """ +import asyncio import copy import logging import os @@ -13,10 +14,10 @@ from pathlib import Path from typing import Any from uuid import UUID -import aiofiles import jsondiff import msgspec +from paskia.db.filelock import LockedFile from paskia.db.logging import log_change from paskia.db.migrations import DBVER, apply_all_migrations, apply_migrations_readonly from paskia.db.structs import DB, Config, SessionContext @@ -27,7 +28,7 @@ _logger = logging.getLogger(__name__) DB_PATH_DEFAULT = "paskia.jsonl" -async def load_readonly(db_path: str, *, rp_id: str = "localhost") -> DB: +def load_readonly(db_path: str, *, rp_id: str = "localhost") -> DB: """Replay JSONL and apply migrations to produce a DB, without writing anything. This is suitable for reading settings before the server starts. @@ -40,8 +41,8 @@ async def load_readonly(db_path: str, *, rp_id: str = "localhost") -> DB: data_dict: dict = {} version = 0 try: - async with aiofiles.open(path, "rb") as f: - content = await f.read() + with open(path, "rb") as f: + content = f.read() for line_num, line in enumerate(content.split(b"\n"), 1): line = line.strip() if not line: @@ -112,54 +113,6 @@ def create_change_record( # Actions that are allowed to create a new database file _BOOTSTRAP_ACTIONS = frozenset({"bootstrap"}) -# Flag to prevent duplicate error messages on fatal flush failure -_flush_failed = False - - -async def flush_changes( - db_path: Path, - pending_changes: deque[_ChangeRecord], -) -> None: - """Write all pending changes to disk. - - Args: - db_path: Path to the JSONL database file - pending_changes: Queue of pending change records (will be cleared on success) - - On failure, logs an error and sends SIGTERM to trigger graceful shutdown. - """ - global _flush_failed - if _flush_failed or not pending_changes: - return - - if not db_path.exists(): - first_action = pending_changes[0].a - if first_action not in _BOOTSTRAP_ACTIONS: - _logger.error( - "Refusing to create database file with action '%s' - " - "only bootstrap can create a new database", - first_action, - ) - _flush_failed = True - os.kill(os.getpid(), signal.SIGTERM) - return - - changes_to_write = list(pending_changes) - - try: - lines = [_change_encoder.encode(change) for change in changes_to_write] - if not lines: - pending_changes.clear() - return - - async with aiofiles.open(db_path, "ab") as f: - await f.write(b"\n".join(lines) + b"\n") - pending_changes.clear() - except OSError as e: - _logger.error("Failed to flush database: %s", e) - _flush_failed = True - os.kill(os.getpid(), signal.SIGTERM) - class JsonlStore: """JSONL persistence layer for a DB instance.""" @@ -167,6 +120,8 @@ class JsonlStore: def __init__(self, db: DB, db_path: str = DB_PATH_DEFAULT): self.db: DB = db self.db_path = Path(db_path) + self._file = LockedFile() + self._flush_failed = False self._previous_builtins: dict[str, Any] = {} self._pending_changes: deque[_ChangeRecord] = deque() self._current_action: str = "system" @@ -185,11 +140,12 @@ class JsonlStore: if not self.db_path.exists(): return + # Open with exclusive write lock and read contents — single threadpool call + content = await asyncio.to_thread(self._file.open_and_read, self.db_path) + # Replay change log to reconstruct state data_dict: dict = {} try: - async with aiofiles.open(self.db_path, "rb") as f: - content = await f.read() for line_num, line in enumerate(content.split(b"\n"), 1): line = line.strip() if not line: @@ -330,5 +286,42 @@ class JsonlStore: self._transaction_snapshot = None async def flush(self) -> None: - """Write all pending changes to disk.""" - await flush_changes(self.db_path, self._pending_changes) + """Write all pending changes to disk. + + On failure, logs an error and sends SIGTERM to trigger graceful shutdown. + """ + if self._flush_failed or not self._pending_changes: + return + + if not self._file.is_open: + first_action = self._pending_changes[0].a + if first_action not in _BOOTSTRAP_ACTIONS: + _logger.error( + "Refusing to create database file with action '%s' - " + "only bootstrap can create a new database", + first_action, + ) + self._flush_failed = True + os.kill(os.getpid(), signal.SIGTERM) + return + # Bootstrap: create and open the file with lock + await asyncio.to_thread(self._file.open, self.db_path, create=True) + + changes_to_write = list(self._pending_changes) + + try: + lines = [_change_encoder.encode(change) for change in changes_to_write] + if not lines: + self._pending_changes.clear() + return + + await asyncio.to_thread(self._file.write, b"\n".join(lines) + b"\n") + self._pending_changes.clear() + except OSError as e: + _logger.error("Failed to flush database: %s", e) + self._flush_failed = True + os.kill(os.getpid(), signal.SIGTERM) + + def close(self) -> None: + """Release the file lock and close the file.""" + self._file.close() diff --git a/pyproject.toml b/pyproject.toml index 36a8f70..522630a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -21,7 +21,6 @@ dependencies = [ "pyjwt[crypto]>=2.11.0", "jsondiff>=2.2.1", "msgspec>=0.20.0", - "aiofiles>=25.1.0", "fastapi-vue>=1.1.0", "ua-parser[regex]>=1.0.1", ]