75 lines
2.7 KiB
Python
75 lines
2.7 KiB
Python
"""Filesystem boundary for SQLite state containing private operator data."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sqlite3
|
|
import stat
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
PRIVATE_DIRECTORY_MODE = stat.S_IRWXU
|
|
PRIVATE_FILE_MODE = stat.S_IRUSR | stat.S_IWUSR
|
|
|
|
|
|
def _reject_symlink(path: Path, *, label: str) -> None:
|
|
try:
|
|
metadata = path.lstat()
|
|
except FileNotFoundError:
|
|
return
|
|
if stat.S_ISLNK(metadata.st_mode):
|
|
raise ValueError(f"private SQLite {label} must not be a symlink: {path}")
|
|
|
|
|
|
def prepare_private_sqlite_path(path: str | Path, *, create: bool = True) -> Path:
|
|
"""Create or repair a private SQLite path without following a database symlink."""
|
|
database = Path(path)
|
|
parent = database.parent
|
|
_reject_symlink(parent, label="directory")
|
|
if create:
|
|
parent.mkdir(parents=True, mode=PRIVATE_DIRECTORY_MODE, exist_ok=True)
|
|
elif not parent.exists():
|
|
raise FileNotFoundError(database)
|
|
_reject_symlink(parent, label="directory")
|
|
if not parent.is_dir():
|
|
raise ValueError(f"private SQLite parent is not a directory: {parent}")
|
|
os.chmod(parent, PRIVATE_DIRECTORY_MODE, follow_symlinks=False)
|
|
|
|
_reject_symlink(database, label="database")
|
|
flags = os.O_RDWR | (os.O_CREAT if create else 0)
|
|
flags |= getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
|
|
descriptor = os.open(database, flags, PRIVATE_FILE_MODE)
|
|
try:
|
|
metadata = os.fstat(descriptor)
|
|
if not stat.S_ISREG(metadata.st_mode):
|
|
raise ValueError(f"private SQLite database is not a regular file: {database}")
|
|
os.fchmod(descriptor, PRIVATE_FILE_MODE)
|
|
finally:
|
|
os.close(descriptor)
|
|
|
|
for suffix in ("-wal", "-shm", "-journal"):
|
|
sidecar = Path(str(database) + suffix)
|
|
_reject_symlink(sidecar, label="sidecar")
|
|
try:
|
|
os.chmod(sidecar, PRIVATE_FILE_MODE, follow_symlinks=False)
|
|
except FileNotFoundError:
|
|
pass
|
|
return database
|
|
|
|
|
|
def connect_private_sqlite(
|
|
path: str | Path, *, existing_only: bool = False, **kwargs: Any
|
|
) -> sqlite3.Connection:
|
|
"""Open SQLite only after enforcing its private directory and file modes."""
|
|
database = prepare_private_sqlite_path(path, create=not existing_only)
|
|
target: str | Path = database
|
|
if existing_only:
|
|
target = f"{database.resolve().as_uri()}?mode=rw"
|
|
kwargs["uri"] = True
|
|
connection = sqlite3.connect(target, **kwargs)
|
|
# SQLite derives new journal/WAL/SHM permissions from the database. Repair
|
|
# sidecars left by older releases as soon as a connection has opened them.
|
|
prepare_private_sqlite_path(database, create=not existing_only)
|
|
return connection
|