kawaandClaude Haiku 4.5
ed41f7f520
Printers and groups (Client) are now scoped to an Owner identified by an opaque
bearer key (secrets.token_urlsafe(32)) stored in an httponly cookie, defaulting
to temporary. First-visit modal offers backup-key download (marks permanent) or
temporary-only choice. /session/restore re-attaches a fresh browser to a saved
key. Every printer-facing route enforces ownership (404 on mismatch, not just
filtering) since printer IDs are sequential ints. Drivers stay global/shared.
On upgrade, pre-existing printer/client rows backfill to a synthetic legacy Owner;
its key is written to {DATA_DIR}/legacy_owner_key.txt for manual restore.
SECURITY: Added Origin/Referer same-origin check on POST /session/restore to
block login-CSRF/session-fixation attacks (cross-site form POST can't re-point
victim's cookie at attacker's Owner without hitting that check first).
Tests: 140 pass (2 deselected: pre-existing locale-flaky, unrelated to this change).
Verified live: modal on first visit, isolation between browsers, backup-key
download and restore flow work end-to-end.
Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
72 lines
2.6 KiB
Python
72 lines
2.6 KiB
Python
"""Peewee SQLite database instance and initialization."""
|
|
from peewee import SqliteDatabase
|
|
|
|
from imptune.config import DB_PATH
|
|
|
|
# Deferred init — path is set at runtime via init_db() so tests can override DB_PATH
|
|
db = SqliteDatabase(None)
|
|
|
|
|
|
def init_db() -> None:
|
|
"""Initialize the SQLite database with WAL mode and foreign keys.
|
|
|
|
Idempotent — safe to call on every application startup.
|
|
Creates all ORM tables if they do not already exist.
|
|
|
|
Closes any existing connection before re-initializing so that test
|
|
fixtures can monkeypatch DB_PATH between test runs.
|
|
"""
|
|
from imptune.db.models import Client, Driver, Owner, Printer, Icon
|
|
|
|
# Re-read DB_PATH each time so tests can patch imptune.config.DB_PATH
|
|
import imptune.config as cfg
|
|
|
|
# Close any lingering connection from a previous run (important for tests)
|
|
if not db.is_closed():
|
|
db.close()
|
|
|
|
db.init(
|
|
cfg.DB_PATH,
|
|
pragmas={
|
|
"journal_mode": "wal",
|
|
"foreign_keys": 1,
|
|
},
|
|
)
|
|
db.connect(reuse_if_open=True)
|
|
db.create_tables([Owner, Client, Driver, Printer, Icon], safe=True)
|
|
_migrate_owner_column(cfg.DATA_DIR)
|
|
|
|
|
|
def _migrate_owner_column(data_dir: str) -> None:
|
|
"""Add owner_id to printer/client if missing, backfilling pre-existing rows.
|
|
|
|
Runs against a DB created before per-owner scoping existed. Idempotent:
|
|
a no-op once the column exists and no rows are left with a NULL owner_id.
|
|
"""
|
|
from pathlib import Path
|
|
|
|
from imptune.db.models import Client, Owner, Printer
|
|
from imptune.services.session import generate_key
|
|
|
|
for table in ("printer", "client"):
|
|
columns = {row[1] for row in db.execute_sql(f"PRAGMA table_info({table})")}
|
|
if "owner_id" not in columns:
|
|
db.execute_sql(
|
|
f"ALTER TABLE {table} ADD COLUMN owner_id INTEGER REFERENCES owner (id)"
|
|
)
|
|
|
|
orphaned = Printer.select().where(Printer.owner.is_null()).count() or Client.select().where(
|
|
Client.owner.is_null()
|
|
).count()
|
|
if not orphaned:
|
|
return
|
|
|
|
legacy_owner = Owner.create(key=generate_key(), is_permanent=True)
|
|
Printer.update(owner=legacy_owner.id).where(Printer.owner.is_null()).execute()
|
|
Client.update(owner=legacy_owner.id).where(Client.owner.is_null()).execute()
|
|
|
|
key_path = Path(data_dir) / "legacy_owner_key.txt"
|
|
key_path.write_text(legacy_owner.key, encoding="utf-8")
|
|
print(f"[imptune] Pre-existing printers/clients migrated to a legacy owner. "
|
|
f"Restore key written to {key_path} — paste it into /session/restore to reclaim them.")
|