LeoVasanko/paskia
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Implement database file locking for extra safety.

Browse Source
This commit is contained in:
LeoVasanko
2026-02-19 00:01:49 +00:00
parent 22ba7231b1
commit b3cb540098
5 changed files with 298 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
paskia/__main__.py
+1 -2
View File
@@ -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
paskia/db/__init__.py
+1 -9
View File
@@ -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",
paskia/db/filelock.py
+247
View File
@@ -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()}")
paskia/db/jsonl.py
+49 -56
View File
@@ -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()
pyproject.toml
-1
View File
@@ -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",
]