diff --git a/CLAUDE.md b/CLAUDE.md index cc0149c..6c6df3d 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -14,6 +14,7 @@ pytest tests/ -k "test_name" # single test by name **Run dev server:** ```bash export DATA_DIR=/tmp/imptune_data +export COOKIE_SECURE=false # plain HTTP — omit if serving behind TLS uvicorn imptune.main:app --reload --port 8000 ``` @@ -33,18 +34,21 @@ pip install -r requirements.txt -r requirements-dev.txt ImpTune make printer deploy packages (`.intunewin` for Intune, `.zip` for NinjaRMM) from Windows driver ZIPs + web UI. No external services — single FastAPI + SQLite + Docker volume. **Request flow:** -1. Driver upload → `api/drivers.py` → `services/inf_parser.py` parse INF → `storage/driver_store.py` store by SHA256 → Peewee `Driver` record -2. Printer config → `api/printers.py` → `db/models.py` `Printer` record (links Driver FK) +1. Driver upload → `api/drivers.py` → `services/inf_parser.py` parse INF → `storage/driver_store.py` store by SHA256 → Peewee `Driver` record (shared/global — visible to every Owner) +2. Printer config → `api/printers.py` → `db/models.py` `Printer` record (links Driver FK, scoped to `request.state.owner`) 3. Icon upload → `api/icons.py` → Pillow validate PNG 256×256 → SHA256 storage → `Icon` record 4. Package export → `api/packages.py` → `generators/script_generator.py` render Jinja2 PS1 templates → `generators/intunewin_builder.py` encrypt ZIP (AES-256-CBC + HMAC-SHA256) **Key modules:** -- `imptune/config.py` — `DATA_DIR`, `DB_PATH`, `DRIVERS_DIR`, `ICONS_DIR` from env -- `imptune/db/database.py` — SQLite WAL mode + `foreign_keys=1`; all models inherit `BaseModel` +- `imptune/config.py` — `DATA_DIR`, `DB_PATH`, `DRIVERS_DIR`, `ICONS_DIR`, `COOKIE_SECURE` from env +- `imptune/db/database.py` — SQLite WAL mode + `foreign_keys=1`; all models inherit `BaseModel`; `init_db()` also backfills `owner_id` on pre-per-owner-scoping DBs into a synthetic legacy `Owner` (key written to `{DATA_DIR}/legacy_owner_key.txt`) +- `imptune/services/session.py` — `OwnerSessionMiddleware` resolves `request.state.owner` from the `imptune_owner_key` cookie, creating one on first visit (skips `/health`) - `imptune/services/inf_parser.py` — auto-detect encoding (UTF-16/UTF-8/cp1252), resolve `%TOKEN%` from `[Strings]`, handle multi-model INFs - `imptune/generators/intunewin_builder.py` — Python-native `.intunewin` (ZIP-in-ZIP); IV 16 bytes (not 32); match reference tool `1.8.6.0` output - `imptune/templates/scripts/` — Jinja2 templates for `install.ps1`, `uninstall.ps1`, `detect.ps1` +**Per-owner storage:** `Printer`/`Client` (groups) are scoped to an `Owner` identified by an opaque bearer key in a cookie — no accounts. `Driver` stays global/shared. Every route taking a `printer_id`/`client_id` must filter/check `.owner == request.state.owner` (404, not 403, on mismatch) — printer IDs are small sequential ints, so a list-only filter isn't enough. Onboarding modal (`templates/base.html`, gated on `request.state.is_new_owner`) offers "download backup key" (`GET /session/key/download`, marks `Owner.is_permanent`) vs. temporary; `/session/restore` re-attaches a browser to a previously downloaded key. In tests, use the `owner` fixture (`tests/conftest.py`) when creating `Printer`/`Client` rows directly via the ORM so the `client` fixture's cookie-scoped requests can see them. + **UI stack:** Pico CSS + HTMX 2 + Alpine.js 3 + Jinja2 server-side templates. **HTMX pattern:** Forms `hx-post`, swap `#driver-list` / `#printer-list` / `#client-list` targets. Errors return inline HTML fragments (HTTP 400/409) via `_error_response()`. Success return partials from `templates/partials/`. @@ -64,4 +68,5 @@ ImpTune make printer deploy packages (`.intunewin` for Intune, `.zip` for NinjaR | Var | Default | Purpose | |-----|---------|---------| | `DATA_DIR` | `/data` | Storage root (DB + drivers + icons) | -| `PORT` | `8000` | Server port | \ No newline at end of file +| `PORT` | `8000` | Server port | +| `COOKIE_SECURE` | `true` | Owner-session cookie `Secure` flag. Set `false` for local plain-HTTP dev (`uvicorn --reload`) or the browser drops the cookie and a new Owner is created on every request. | \ No newline at end of file diff --git a/imptune/api/clients.py b/imptune/api/clients.py index db40556..785985d 100644 --- a/imptune/api/clients.py +++ b/imptune/api/clients.py @@ -27,7 +27,7 @@ def _error_response(message: str, status_code: int = 400) -> HTMLResponse: def _render_client_list(request: Request) -> HTMLResponse: """Render the client list partial for HTMX swap.""" - clients = list(Client.select().order_by(Client.name)) + clients = list(Client.select().where(Client.owner == request.state.owner).order_by(Client.name)) return templates.TemplateResponse( request=request, name="partials/client_list.html", @@ -47,7 +47,7 @@ def create_client(request: Request, name: str = Form(...)) -> HTMLResponse: return _error_response("Client name is required.") try: - Client.create(name=name) + Client.create(name=name, owner=request.state.owner) except IntegrityError: return _error_response(f"Client '{name}' already exists.", status_code=409) diff --git a/imptune/api/icons.py b/imptune/api/icons.py index af123da..8c25c13 100644 --- a/imptune/api/icons.py +++ b/imptune/api/icons.py @@ -5,7 +5,7 @@ import hashlib import io from pathlib import Path -from fastapi import APIRouter, UploadFile +from fastapi import APIRouter, Request, UploadFile from fastapi.responses import HTMLResponse from PIL import Image @@ -18,7 +18,7 @@ MAX_ICON_BYTES = 750 * 1024 # 750 KB @router.post("/{printer_id}/icon", response_class=HTMLResponse) -def upload_icon(printer_id: int, file: UploadFile) -> HTMLResponse: +def upload_icon(request: Request, printer_id: int, file: UploadFile) -> HTMLResponse: """Accept a printer icon PNG, validate it, store it, and update the Icon record. Validation rules: @@ -29,8 +29,10 @@ def upload_icon(printer_id: int, file: UploadFile) -> HTMLResponse: Replaces any previously uploaded icon for this printer. Returns an HTMX-friendly HTML fragment. """ - # Check printer exists - printer = Printer.get_or_none(Printer.id == printer_id) + # Check printer exists and belongs to this owner + printer = Printer.get_or_none( + (Printer.id == printer_id) & (Printer.owner == request.state.owner) + ) if printer is None: return HTMLResponse( content="
Printer not found.
", diff --git a/imptune/api/packages.py b/imptune/api/packages.py index 0d321ac..ec7b6e1 100644 --- a/imptune/api/packages.py +++ b/imptune/api/packages.py @@ -6,11 +6,11 @@ import shutil import tempfile import zipfile -from fastapi import APIRouter +from fastapi import APIRouter, Request from fastapi.responses import PlainTextResponse, Response import imptune.config as cfg -from imptune.db.models import Icon, Printer +from imptune.db.models import Icon, Owner, Printer from imptune.generators.intunewin_builder import build_intunewin from imptune.generators.script_generator import render_detect, render_install, render_uninstall from imptune.storage.driver_store import DriverStore @@ -18,9 +18,9 @@ from imptune.storage.driver_store import DriverStore router = APIRouter(prefix="/printers") -def _get_printer_and_driver(printer_id: int): - """Fetch printer and validate driver — returns (printer, driver, driver_name) or PlainTextResponse error.""" - printer = Printer.get_or_none(Printer.id == printer_id) +def _get_printer_and_driver(printer_id: int, owner: Owner): + """Fetch printer (scoped to owner) and validate driver — returns (printer, driver, driver_name) or PlainTextResponse error.""" + printer = Printer.get_or_none((Printer.id == printer_id) & (Printer.owner == owner)) if printer is None: return None, PlainTextResponse("Printer not found", status_code=404) @@ -48,9 +48,9 @@ def _get_driver_zip_path(driver) -> str: @router.get("/{printer_id}/packages/ninja") -def get_ninja_package(printer_id: int): +def get_ninja_package(request: Request, printer_id: int): """Download a NinjaRMM-ready ZIP containing install.ps1 and the driver files.""" - result, error = _get_printer_and_driver(printer_id) + result, error = _get_printer_and_driver(printer_id, request.state.owner) if error is not None: return error @@ -96,9 +96,9 @@ def get_ninja_package(printer_id: int): @router.get("/{printer_id}/packages/intunewin") -def get_intunewin_package(printer_id: int): +def get_intunewin_package(request: Request, printer_id: int): """Download a Microsoft Intune .intunewin deployment package.""" - result, error = _get_printer_and_driver(printer_id) + result, error = _get_printer_and_driver(printer_id, request.state.owner) if error is not None: return error diff --git a/imptune/api/pages.py b/imptune/api/pages.py index e069db4..593ab04 100644 --- a/imptune/api/pages.py +++ b/imptune/api/pages.py @@ -17,12 +17,16 @@ templates = Jinja2Templates(directory=str(Path(__file__).parent.parent / "templa def dashboard(request: Request): from imptune.db.models import Printer + owner = request.state.owner recent_printers = list( - Printer.select().order_by(Printer.created_at.desc()).limit(5) + Printer.select() + .where(Printer.owner == owner) + .order_by(Printer.created_at.desc()) + .limit(5) ) recent_packages = list( Printer.select() - .where(Printer.driver.is_null(False)) + .where((Printer.owner == owner) & Printer.driver.is_null(False)) .order_by(Printer.created_at.desc()) .limit(5) ) @@ -56,9 +60,11 @@ def drivers_page(request: Request): def printers_page(request: Request): from imptune.db.models import Client, Driver, Printer + owner = request.state.owner query = ( Printer.select(Printer, Client) .join(Client, JOIN.LEFT_OUTER) + .where(Printer.owner == owner) .order_by(Client.name, Printer.name) ) grouped: dict[str, list] = defaultdict(list) @@ -66,7 +72,7 @@ def printers_page(request: Request): client_name = p.client.name if p.client_id else "Unassigned" grouped[client_name].append(p) - clients = list(Client.select().order_by(Client.name)) + clients = list(Client.select().where(Client.owner == owner).order_by(Client.name)) all_drivers = list(Driver.select().order_by(Driver.uploaded_at.desc())) driver_data = [] @@ -89,7 +95,7 @@ def printers_page(request: Request): def printers_new_page(request: Request): from imptune.db.models import Client, Driver - clients = list(Client.select().order_by(Client.name)) + clients = list(Client.select().where(Client.owner == request.state.owner).order_by(Client.name)) all_drivers = list(Driver.select().order_by(Driver.uploaded_at.desc())) driver_data = [ {"driver": d, "names": json.loads(d.driver_desc) if d.driver_desc else []} @@ -111,7 +117,7 @@ def printer_detail(request: Request, printer_id: int): .join(Client, JOIN.LEFT_OUTER) .switch(Printer) .join(Driver, JOIN.LEFT_OUTER) - .where(Printer.id == printer_id) + .where((Printer.id == printer_id) & (Printer.owner == request.state.owner)) .first() ) if printer is None: @@ -148,7 +154,7 @@ def printer_detail(request: Request, printer_id: int): def clients_page(request: Request): from imptune.db.models import Client - clients = list(Client.select().order_by(Client.name)) + clients = list(Client.select().where(Client.owner == request.state.owner).order_by(Client.name)) return templates.TemplateResponse( request=request, name="clients.html", @@ -160,7 +166,8 @@ def clients_page(request: Request): def client_detail(request: Request, client_id: int): from imptune.db.models import Client, Driver, Printer - client = Client.get_or_none(Client.id == client_id) + owner = request.state.owner + client = Client.get_or_none((Client.id == client_id) & (Client.owner == owner)) if client is None: return HTMLResponse( content="Client not found.
", @@ -170,12 +177,12 @@ def client_detail(request: Request, client_id: int): query = ( Printer.select(Printer, Client) .join(Client, JOIN.LEFT_OUTER) - .where(Printer.client == client_id) + .where((Printer.client == client_id) & (Printer.owner == owner)) .order_by(Printer.name) ) grouped = {client.name: list(query)} - clients = list(Client.select().order_by(Client.name)) + clients = list(Client.select().where(Client.owner == owner).order_by(Client.name)) all_drivers = list(Driver.select().order_by(Driver.uploaded_at.desc())) driver_data = [ {"driver": d, "names": json.loads(d.driver_desc) if d.driver_desc else []} @@ -203,7 +210,7 @@ def packages_page(request: Request): .join(Client, JOIN.LEFT_OUTER) .switch(Printer) .join(Driver, JOIN.LEFT_OUTER) - .where(Printer.driver.is_null(False)) + .where((Printer.owner == request.state.owner) & Printer.driver.is_null(False)) .order_by(Printer.name) ) return templates.TemplateResponse( diff --git a/imptune/api/printers.py b/imptune/api/printers.py index 2e88a56..0af007e 100644 --- a/imptune/api/printers.py +++ b/imptune/api/printers.py @@ -33,9 +33,11 @@ def _render_printer_list(request: Request) -> HTMLResponse: """Query printers with LEFT JOIN on client and render grouped partial.""" import json + owner = request.state.owner query = ( Printer.select(Printer, Client) .join(Client, JOIN.LEFT_OUTER) + .where(Printer.owner == owner) .order_by(Client.name, Printer.name) ) grouped: dict[str, list[Printer]] = defaultdict(list) @@ -43,7 +45,7 @@ def _render_printer_list(request: Request) -> HTMLResponse: client_name = p.client.name if p.client_id else "Unassigned" grouped[client_name].append(p) - clients = list(Client.select().order_by(Client.name)) + clients = list(Client.select().where(Client.owner == owner).order_by(Client.name)) all_drivers = list(Driver.select().order_by(Driver.uploaded_at.desc())) driver_data = [ {"driver": d, "names": json.loads(d.driver_desc) if d.driver_desc else []} @@ -94,8 +96,13 @@ def create_printer( color_mode_bool = color_mode == "on" collate_bool = collate == "on" - # Resolve optional FK IDs + owner = request.state.owner + + # Resolve optional FK IDs — client must belong to this owner client_fk = int(client_id) if client_id.strip() else None + if client_fk is not None: + if Client.get_or_none((Client.id == client_fk) & (Client.owner == owner)) is None: + return _error_response(f"Client {client_fk} not found.", status_code=404) driver_fk = int(driver_id) if driver_id.strip() else None Printer.create( @@ -106,6 +113,7 @@ def create_printer( color_mode=color_mode_bool, paper_size=paper_size, collate=collate_bool, + owner=owner, client=client_fk, driver=driver_fk, ) @@ -116,7 +124,11 @@ def create_printer( @router.delete("/{printer_id}", response_class=HTMLResponse) def delete_printer(request: Request, printer_id: int) -> HTMLResponse: """Delete a printer by ID. Returns updated printer list partial.""" - deleted = Printer.delete().where(Printer.id == printer_id).execute() + deleted = ( + Printer.delete() + .where((Printer.id == printer_id) & (Printer.owner == request.state.owner)) + .execute() + ) if not deleted: return _error_response(f"Printer {printer_id} not found.", status_code=404) @@ -140,7 +152,8 @@ def update_printer( """Update an existing printer configuration in-place.""" from datetime import UTC, datetime - printer = Printer.get_or_none(Printer.id == printer_id) + owner = request.state.owner + printer = Printer.get_or_none((Printer.id == printer_id) & (Printer.owner == owner)) if printer is None: return _error_response(f"Printer {printer_id} not found.", status_code=404) @@ -159,6 +172,11 @@ def update_printer( if paper_size not in _VALID_PAPER: return _error_response(f"Invalid paper size: {paper_size}.") + client_fk = int(client_id) if client_id.strip() else None + if client_fk is not None: + if Client.get_or_none((Client.id == client_fk) & (Client.owner == owner)) is None: + return _error_response(f"Client {client_fk} not found.", status_code=404) + printer.name = name printer.ip_address = ip_address printer.port_name = port_name @@ -166,7 +184,7 @@ def update_printer( printer.color_mode = color_mode == "on" printer.paper_size = paper_size printer.collate = collate == "on" - printer.client = int(client_id) if client_id.strip() else None + printer.client = client_fk printer.driver = int(driver_id) if driver_id.strip() else None printer.updated_at = datetime.now(UTC).replace(tzinfo=None) printer.save() diff --git a/imptune/api/scripts.py b/imptune/api/scripts.py index cbc53e1..0b16b12 100644 --- a/imptune/api/scripts.py +++ b/imptune/api/scripts.py @@ -1,18 +1,18 @@ """Script download endpoints — generates and serves PowerShell scripts for a printer.""" import json -from fastapi import APIRouter +from fastapi import APIRouter, Request from fastapi.responses import PlainTextResponse -from imptune.db.models import Printer +from imptune.db.models import Owner, Printer from imptune.generators.script_generator import render_detect, render_install, render_uninstall router = APIRouter(prefix="/printers") -def _get_printer_and_driver(printer_id: int): - """Fetch printer and validate driver — returns (printer, driver_name) or PlainTextResponse error.""" - printer = Printer.get_or_none(Printer.id == printer_id) +def _get_printer_and_driver(printer_id: int, owner: Owner): + """Fetch printer (scoped to owner) and validate driver — returns (printer, driver_name) or PlainTextResponse error.""" + printer = Printer.get_or_none((Printer.id == printer_id) & (Printer.owner == owner)) if printer is None: return None, PlainTextResponse("Printer not found", status_code=404) @@ -35,8 +35,8 @@ def _get_printer_and_driver(printer_id: int): return (printer, driver, driver_name), None -def _install_response(printer_id: int): - result, error = _get_printer_and_driver(printer_id) +def _install_response(printer_id: int, owner: Owner): + result, error = _get_printer_and_driver(printer_id, owner) if error is not None: return error printer, driver, driver_name = result @@ -57,8 +57,8 @@ def _install_response(printer_id: int): ) -def _uninstall_response(printer_id: int): - result, error = _get_printer_and_driver(printer_id) +def _uninstall_response(printer_id: int, owner: Owner): + result, error = _get_printer_and_driver(printer_id, owner) if error is not None: return error printer, driver, driver_name = result @@ -73,8 +73,8 @@ def _uninstall_response(printer_id: int): ) -def _detect_response(printer_id: int): - result, error = _get_printer_and_driver(printer_id) +def _detect_response(printer_id: int, owner: Owner): + result, error = _get_printer_and_driver(printer_id, owner) if error is not None: return error printer, driver, driver_name = result @@ -86,36 +86,36 @@ def _detect_response(printer_id: int): @router.get("/{printer_id}/scripts/install") -def get_install_script(printer_id: int): +def get_install_script(request: Request, printer_id: int): """Download the PowerShell install script for a printer.""" - return _install_response(printer_id) + return _install_response(printer_id, request.state.owner) @router.get("/{printer_id}/scripts/install.ps1") -def get_install_script_ps1(printer_id: int): +def get_install_script_ps1(request: Request, printer_id: int): """Download the PowerShell install script for a printer (.ps1 alias).""" - return _install_response(printer_id) + return _install_response(printer_id, request.state.owner) @router.get("/{printer_id}/scripts/uninstall") -def get_uninstall_script(printer_id: int): +def get_uninstall_script(request: Request, printer_id: int): """Download the PowerShell uninstall script for a printer.""" - return _uninstall_response(printer_id) + return _uninstall_response(printer_id, request.state.owner) @router.get("/{printer_id}/scripts/uninstall.ps1") -def get_uninstall_script_ps1(printer_id: int): +def get_uninstall_script_ps1(request: Request, printer_id: int): """Download the PowerShell uninstall script for a printer (.ps1 alias).""" - return _uninstall_response(printer_id) + return _uninstall_response(printer_id, request.state.owner) @router.get("/{printer_id}/scripts/detect") -def get_detect_script(printer_id: int): +def get_detect_script(request: Request, printer_id: int): """Download the PowerShell detection script for a printer.""" - return _detect_response(printer_id) + return _detect_response(printer_id, request.state.owner) @router.get("/{printer_id}/scripts/detect.ps1") -def get_detect_script_ps1(printer_id: int): +def get_detect_script_ps1(request: Request, printer_id: int): """Download the PowerShell detection script for a printer (.ps1 alias).""" - return _detect_response(printer_id) + return _detect_response(printer_id, request.state.owner) diff --git a/imptune/api/session.py b/imptune/api/session.py new file mode 100644 index 0000000..18e7682 --- /dev/null +++ b/imptune/api/session.py @@ -0,0 +1,73 @@ +"""Owner session routes — backup-key download and restore-on-new-browser.""" +from __future__ import annotations + +from pathlib import Path + +from fastapi import APIRouter, Form, Request +from fastapi.responses import HTMLResponse, PlainTextResponse, RedirectResponse +from fastapi.templating import Jinja2Templates + +import imptune.config as cfg +from imptune.db.models import Owner +from imptune.services.session import COOKIE_MAX_AGE, COOKIE_NAME, is_same_origin + +router = APIRouter(prefix="/session") + +templates = Jinja2Templates( + directory=str(Path(__file__).parent.parent / "templates") +) + + +@router.get("/key/download") +def download_key(request: Request) -> PlainTextResponse: + """Mark the current owner permanent and hand back its key as a backup file.""" + owner: Owner = request.state.owner + if not owner.is_permanent: + owner.is_permanent = True + owner.save() + + return PlainTextResponse( + content=owner.key, + headers={"Content-Disposition": 'attachment; filename="imptune-backup-key.txt"'}, + ) + + +@router.get("/restore", response_class=HTMLResponse) +def restore_page(request: Request, error: str = "") -> HTMLResponse: + return templates.TemplateResponse( + request=request, + name="session_restore.html", + context={"error": error}, + ) + + +@router.post("/restore") +def restore_session(request: Request, key: str = Form(...)): + """Re-associate this browser with a previously downloaded backup key.""" + if not is_same_origin(request): + return templates.TemplateResponse( + request=request, + name="session_restore.html", + context={"error": "Request rejected — please submit this form directly from this site."}, + status_code=403, + ) + + owner = Owner.get_or_none(Owner.key == key.strip()) + if owner is None: + return templates.TemplateResponse( + request=request, + name="session_restore.html", + context={"error": "Key not found."}, + status_code=404, + ) + + response = RedirectResponse(url="/", status_code=303) + response.set_cookie( + COOKIE_NAME, + owner.key, + max_age=COOKIE_MAX_AGE, + httponly=True, + samesite="lax", + secure=cfg.COOKIE_SECURE, + ) + return response diff --git a/imptune/config.py b/imptune/config.py index 125792a..dfac5d7 100644 --- a/imptune/config.py +++ b/imptune/config.py @@ -7,6 +7,8 @@ load_dotenv() DATA_DIR = os.environ.get("DATA_DIR", "/data") PORT = int(os.environ.get("PORT", "8000")) +# Set to "false" for local plain-HTTP dev — browsers drop Secure cookies over HTTP. +COOKIE_SECURE = os.environ.get("COOKIE_SECURE", "true").lower() != "false" DB_PATH = str(Path(DATA_DIR) / "imptune.db") DRIVERS_DIR = str(Path(DATA_DIR) / "drivers") diff --git a/imptune/db/database.py b/imptune/db/database.py index b3d19a1..e7e8f47 100644 --- a/imptune/db/database.py +++ b/imptune/db/database.py @@ -16,7 +16,7 @@ def init_db() -> None: 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, Printer, Icon + 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 @@ -33,4 +33,39 @@ def init_db() -> None: }, ) db.connect(reuse_if_open=True) - db.create_tables([Client, Driver, Printer, Icon], safe=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.") diff --git a/imptune/db/models.py b/imptune/db/models.py index c68eb2b..3fdd432 100644 --- a/imptune/db/models.py +++ b/imptune/db/models.py @@ -24,14 +24,27 @@ class BaseModel(Model): database = db -class Client(BaseModel): - """Represents a deployment target (AD client / OU).""" +class Owner(BaseModel): + """Cookie-scoped identity — no accounts, just an opaque bearer key.""" - name = CharField(unique=True) + key = CharField(unique=True, index=True) + is_permanent = BooleanField(default=False) + created_at = DateTimeField(default=_utcnow) + + class Meta: + table_name = "owner" + + +class Client(BaseModel): + """Represents a deployment target (AD client / OU), scoped to an Owner.""" + + name = CharField() + owner = ForeignKeyField(Owner, backref="clients") created_at = DateTimeField(default=_utcnow) class Meta: table_name = "client" + indexes = ((("owner", "name"), True),) class Driver(BaseModel): @@ -56,6 +69,7 @@ class Printer(BaseModel): name = CharField() ip_address = CharField() port_name = CharField() + owner = ForeignKeyField(Owner, backref="printers") client = ForeignKeyField(Client, null=True, backref="printers") driver = ForeignKeyField(Driver, null=True, backref="printers") duplex_mode = CharField(default="OneSided") diff --git a/imptune/main.py b/imptune/main.py index 71d15b5..a6dfb16 100644 --- a/imptune/main.py +++ b/imptune/main.py @@ -5,9 +5,10 @@ from pathlib import Path from fastapi import FastAPI from fastapi.staticfiles import StaticFiles -from imptune.api import clients, drivers, health, icons, pages, packages, printers, scripts +from imptune.api import clients, drivers, health, icons, pages, packages, printers, scripts, session from imptune.config import DATA_DIR, DRIVERS_DIR, ICONS_DIR from imptune.db.database import db, init_db +from imptune.services.session import OwnerSessionMiddleware @asynccontextmanager @@ -23,6 +24,7 @@ async def lifespan(app: FastAPI): app = FastAPI(title="ImpTune", lifespan=lifespan) +app.add_middleware(OwnerSessionMiddleware) # Serve baked-in static assets (pico.min.css, htmx.min.js, alpine.min.js) _static_dir = Path(__file__).parent / "static" @@ -37,3 +39,4 @@ app.include_router(clients.router) app.include_router(scripts.router) app.include_router(packages.router) app.include_router(icons.router) +app.include_router(session.router) diff --git a/imptune/services/session.py b/imptune/services/session.py new file mode 100644 index 0000000..b1ffe5c --- /dev/null +++ b/imptune/services/session.py @@ -0,0 +1,84 @@ +"""Cookie-scoped Owner session — opaque bearer key, no accounts.""" +from __future__ import annotations + +import secrets + +from starlette.middleware.base import BaseHTTPMiddleware +from starlette.requests import Request + +import imptune.config as cfg +from imptune.db.models import Owner + +COOKIE_NAME = "imptune_owner_key" +COOKIE_MAX_AGE = 10 * 365 * 24 * 60 * 60 # 10 years + + +def generate_key() -> str: + return secrets.token_urlsafe(32) + + +def is_same_origin(request: Request) -> bool: + """Origin/Referer check — the CSRF guard for /session/restore. + + Restoring a key re-points the cookie at a *different* Owner, so unlike + the rest of the app's unprotected POSTs (which only ever mutate the + caller's own data), a forged cross-site POST here is a login-CSRF / + session-fixation vector: an attacker who knows their own key can force + a victim's browser onto the attacker's Owner. Browsers always send + Origin (and usually Referer) on form POSTs, same-site or not, so + requiring a match — and rejecting when both are absent — blocks a plain + auto-submitting HTML form without needing a token. + """ + expected = f"{request.url.scheme}://{request.url.netloc}" + + origin = request.headers.get("origin") + if origin is not None: + return origin == expected + + referer = request.headers.get("referer") + if referer: + from urllib.parse import urlparse + + parsed = urlparse(referer) + return f"{parsed.scheme}://{parsed.netloc}" == expected + + return False + + +class OwnerSessionMiddleware(BaseHTTPMiddleware): + """Resolves request.state.owner from a cookie, creating one on first visit.""" + + async def dispatch(self, request: Request, call_next): + if request.url.path == "/health": + return await call_next(request) + + key = request.cookies.get(COOKIE_NAME) + owner = Owner.get_or_none(Owner.key == key) if key else None + is_new = owner is None + if owner is None: + owner = Owner.create(key=generate_key(), is_permanent=False) + + request.state.owner = owner + request.state.is_new_owner = is_new + + response = await call_next(request) + + # Routes like /session/restore intentionally set this cookie themselves + # (to a *different* owner than the one this middleware just minted) — + # don't clobber that with the auto-provisioned one. + route_already_set_cookie = any( + header.lower() == b"set-cookie" and value.startswith(f"{COOKIE_NAME}=".encode()) + for header, value in response.raw_headers + ) + + if is_new and not route_already_set_cookie: + response.set_cookie( + COOKIE_NAME, + owner.key, + max_age=COOKIE_MAX_AGE, + httponly=True, + samesite="lax", + secure=cfg.COOKIE_SECURE, + ) + + return response diff --git a/imptune/templates/base.html b/imptune/templates/base.html index 3bea785..35efdef 100644 --- a/imptune/templates/base.html +++ b/imptune/templates/base.html @@ -315,6 +315,10 @@Paste the key from your imptune-backup-key.txt backup file to
+recover your printers, configs, and groups on this browser.
{{ error }}