From dbe1eec53b55bd06783cac910c38841a6ee5c6a7 Mon Sep 17 00:00:00 2001 From: Victor Martinez Date: Sat, 24 Oct 2026 16:32:42 +0200 Subject: [PATCH] feat: persist profile-scoped multi-source station provenance (#276) --- fluxtuner/core/custom_stations.py | 12 ++ fluxtuner/core/db.py | 74 ++++++- fluxtuner/core/m3u_import.py | 59 ++++++ fluxtuner/core/station_provenance.py | 192 ++++++++++++++++++ fluxtuner/gui/gtk_m3u_import.py | 33 ++- fluxtuner/gui/window.py | 1 + fluxtuner/tui_custom_stations.py | 1 + fluxtuner/tui_m3u_import.py | 52 ++++- fluxtuner/web/library.py | 4 +- fluxtuner/web/m3u_import.py | 38 +++- fluxtuner/web/remote_import_receipts.py | 100 +++++++++ fluxtuner/web/routes/imports.py | 27 ++- fluxtuner/web/static/js/station-imports.js | 33 ++- tests/test_db.py | 7 +- tests/test_gtk_m3u_import.py | 64 +++++- tests/test_gtk_orchestration.py | 1 + tests/test_m3u_import_new_source.py | 167 +++++++++++++++ tests/test_m3u_import_provenance.py | 190 +++++++++++++++++ tests/test_station_provenance.py | 110 ++++++++++ ...t_station_provenance_manual_integration.py | 148 ++++++++++++++ tests/test_station_provenance_migration.py | 157 ++++++++++++++ ...t_station_provenance_remote_integration.py | 98 +++++++++ tests/test_station_provenance_security.py | 172 ++++++++++++++++ tests/test_tui_m3u_import.py | 87 ++++++++ tests/test_web_reauth_storage.py | 4 +- tests/test_web_remote_provenance.py | 164 +++++++++++++++ 26 files changed, 1955 insertions(+), 40 deletions(-) create mode 100644 fluxtuner/core/station_provenance.py create mode 100644 fluxtuner/web/remote_import_receipts.py create mode 100644 tests/test_m3u_import_new_source.py create mode 100644 tests/test_m3u_import_provenance.py create mode 100644 tests/test_station_provenance.py create mode 100644 tests/test_station_provenance_manual_integration.py create mode 100644 tests/test_station_provenance_migration.py create mode 100644 tests/test_station_provenance_remote_integration.py create mode 100644 tests/test_station_provenance_security.py create mode 100644 tests/test_web_remote_provenance.py diff --git a/fluxtuner/core/custom_stations.py b/fluxtuner/core/custom_stations.py index 6783925..f86fa9c 100644 --- a/fluxtuner/core/custom_stations.py +++ b/fluxtuner/core/custom_stations.py @@ -7,6 +7,7 @@ from urllib.parse import urlsplit from fluxtuner.core import db +from fluxtuner.core import station_provenance as provenance from fluxtuner.core.stations import station_metadata MAX_NAME_LENGTH = 200 @@ -63,6 +64,7 @@ def add_custom_station( profile_id: int, name: str, url: str, + record_manual_provenance: bool = False, ) -> bool: """Add a manual station to one existing profile's favorites. @@ -71,6 +73,8 @@ def add_custom_station( belongs to the new profile favorite, so other profiles remain unchanged. The caller owns the transaction and must commit successful changes. """ + if type(record_manual_provenance) is not bool: + raise ValueError("Invalid manual provenance option.") clean_name = _validate_name(name) clean_url = _validate_url(url) if type(profile_id) is not int or profile_id <= 0: @@ -124,6 +128,14 @@ def add_custom_station( (profile_id, int(row["id"]), clean_name, now, now), ) added = cursor.rowcount > 0 + if added and record_manual_provenance: + source_id = provenance.create_source(conn, profile_id=profile_id, kind="manual_entry") + provenance.record_observation( + conn, + profile_id=profile_id, + source_id=source_id, + station_id=int(row["id"]), + ) except Exception: conn.execute("ROLLBACK TO SAVEPOINT add_custom_station") conn.execute("RELEASE SAVEPOINT add_custom_station") diff --git a/fluxtuner/core/db.py b/fluxtuner/core/db.py index 64633ee..b8ee035 100644 --- a/fluxtuner/core/db.py +++ b/fluxtuner/core/db.py @@ -20,7 +20,7 @@ DEFAULT_USER_NAME = "default" DEFAULT_PROFILE_NAME = "default" SCHEMA_MIGRATION_NAME = "schema_v6" -CURRENT_SCHEMA_VERSION = 3 +CURRENT_SCHEMA_VERSION = 4 APPROVAL_APPROVED = "approved" APPROVAL_PENDING = "pending" APPROVAL_REJECTED = "rejected" @@ -157,6 +157,12 @@ def migrate_schema(conn: sqlite3.Connection) -> None: conn.execute("PRAGMA user_version = 3") continue + if stored_version == 3: + _create_station_provenance_schema(conn) + stored_version = 4 + conn.execute("PRAGMA user_version = 4") + continue + raise RuntimeError(f"Unsupported database schema version: {stored_version}.") conn.commit() @@ -428,6 +434,72 @@ def create_schema(conn: sqlite3.Connection) -> None: ON playlist_stations(playlist_id, position); """, ) + _create_station_provenance_schema(conn) + + +def _create_station_provenance_schema(conn: sqlite3.Connection) -> None: + """Create profile-scoped station provenance storage, without committing.""" + _execute_schema_script( + conn, + """ + CREATE TABLE IF NOT EXISTS station_sources ( + id INTEGER PRIMARY KEY, + profile_id INTEGER NOT NULL, + kind TEXT NOT NULL, + source_ref TEXT NOT NULL, + display_name TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + FOREIGN KEY (profile_id) + REFERENCES profiles(id) ON DELETE CASCADE, + UNIQUE (profile_id, kind, source_ref), + CHECK ( + kind IN ('local_collection', 'remote_collection', 'manual_entry') + ), + CHECK ( + length(source_ref) BETWEEN 1 AND 128 + AND length(trim(source_ref)) = length(source_ref) + ), + CHECK ( + display_name IS NULL + OR length(display_name) BETWEEN 1 AND 120 + ) + ); + + CREATE INDEX IF NOT EXISTS idx_station_sources_profile_kind + ON station_sources(profile_id, kind); + + CREATE TABLE IF NOT EXISTS station_source_memberships ( + source_id INTEGER NOT NULL, + station_id INTEGER NOT NULL, + first_observed_at TEXT NOT NULL, + last_observed_at TEXT NOT NULL, + PRIMARY KEY (source_id, station_id), + FOREIGN KEY (source_id) + REFERENCES station_sources(id) ON DELETE CASCADE, + FOREIGN KEY (station_id) + REFERENCES stations(id) ON DELETE CASCADE + ); + + CREATE INDEX IF NOT EXISTS idx_station_source_memberships_station + ON station_source_memberships(station_id); + + CREATE TABLE IF NOT EXISTS remote_import_receipts ( + token_hash TEXT PRIMARY KEY, + user_id INTEGER NOT NULL, + profile_id INTEGER NOT NULL, + session_hash TEXT NOT NULL, + payload_hash TEXT NOT NULL, + expires_at TEXT NOT NULL, + consumed_at TEXT, + FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE, + FOREIGN KEY (profile_id) REFERENCES profiles(id) ON DELETE CASCADE + ); + + CREATE INDEX IF NOT EXISTS idx_remote_import_receipts_expires + ON remote_import_receipts(expires_at); + """, + ) def _column_names(conn: sqlite3.Connection, table_name: str) -> set[str]: diff --git a/fluxtuner/core/m3u_import.py b/fluxtuner/core/m3u_import.py index 02a2e6d..3b783e2 100644 --- a/fluxtuner/core/m3u_import.py +++ b/fluxtuner/core/m3u_import.py @@ -11,6 +11,7 @@ from urllib.parse import urlsplit from fluxtuner.core import db +from fluxtuner.core import station_provenance as provenance from fluxtuner.core.custom_stations import ( MAX_NAME_LENGTH, _has_controls, @@ -53,6 +54,14 @@ class M3UImportResult: already_in_playlist: int +@dataclass(frozen=True, slots=True) +class M3UImportWithSourceResult: + """A newly created provenance source and the unchanged import result.""" + + result: M3UImportResult + source_id: int + + def _extinf_name(line: str) -> str: """Find the title separator without splitting quoted attribute values.""" quoted = False @@ -306,6 +315,7 @@ def import_m3u( profile_id: int, playlist_name: str, entries: Sequence[M3UEntry], + source_id: int | None = None, ) -> M3UImportResult: """Import stations into favorites and one playlist without committing. @@ -325,6 +335,8 @@ def import_m3u( if conn.execute("SELECT 1 FROM profiles WHERE id = ?", (profile_id,)).fetchone() is None: raise ValueError("The selected profile no longer exists.") accepted = _validated_entries(entries) + if source_id is not None: + provenance.get_source(conn, profile_id=profile_id, source_id=source_id) if not conn.in_transaction: conn.execute("BEGIN") @@ -367,6 +379,14 @@ def import_m3u( if station is None: raise RuntimeError("Could not find the imported station.") + if source_id is not None: + provenance.record_observation( + conn, + profile_id=profile_id, + source_id=source_id, + station_id=int(station["id"]), + ) + # Link the existing row directly: the normal playlist helper # upserts station metadata, which imports must preserve. cursor = conn.execute( @@ -393,3 +413,42 @@ def import_m3u( raise conn.execute("RELEASE SAVEPOINT m3u_import") return result + + +def import_m3u_with_new_source( + conn: sqlite3.Connection, + *, + profile_id: int, + playlist_name: str, + entries: Sequence[M3UEntry], + display_name: str | None = None, + kind: str = "local_collection", +) -> M3UImportWithSourceResult: + """Atomically create a collection source and import its stations. + + The caller controls commit/rollback. Remote kind is reserved for + callers that have obtained a validated remote collection snapshot. + """ + if not conn.in_transaction: + conn.execute("BEGIN") + conn.execute("SAVEPOINT m3u_import_new_source") + try: + source_id = provenance.create_source( + conn, + profile_id=profile_id, + kind=kind, + display_name=display_name, + ) + result = import_m3u( + conn, + profile_id=profile_id, + playlist_name=playlist_name, + entries=entries, + source_id=source_id, + ) + except Exception: + conn.execute("ROLLBACK TO SAVEPOINT m3u_import_new_source") + conn.execute("RELEASE SAVEPOINT m3u_import_new_source") + raise + conn.execute("RELEASE SAVEPOINT m3u_import_new_source") + return M3UImportWithSourceResult(result=result, source_id=source_id) diff --git a/fluxtuner/core/station_provenance.py b/fluxtuner/core/station_provenance.py new file mode 100644 index 0000000..5c2d28c --- /dev/null +++ b/fluxtuner/core/station_provenance.py @@ -0,0 +1,192 @@ +# SPDX-License-Identifier: MIT +"""Profile-owned provenance for shared station records.""" + +from __future__ import annotations + +import secrets +import sqlite3 +import unicodedata +from typing import Any + +from fluxtuner.core import db + +SOURCE_KINDS = frozenset({"local_collection", "remote_collection", "manual_entry"}) +MAX_DISPLAY_NAME_LENGTH = 120 + + +def _profile(conn: sqlite3.Connection, profile_id: int) -> None: + if type(profile_id) is not int or profile_id <= 0: + raise ValueError("An existing profile is required.") + row = conn.execute("SELECT 1 FROM profiles WHERE id = ?", (profile_id,)).fetchone() + if row is None: + raise ValueError("The selected profile no longer exists.") + + +def _source(conn: sqlite3.Connection, profile_id: int, source_id: int) -> sqlite3.Row: + _profile(conn, profile_id) + if type(source_id) is not int or source_id <= 0: + raise ValueError("Source not found for the selected profile.") + row = conn.execute( + "SELECT * FROM station_sources WHERE id = ? AND profile_id = ?", + (source_id, profile_id), + ).fetchone() + if row is None: + raise ValueError("Source not found for the selected profile.") + return row + + +def _display_name(value: str | None) -> str | None: + if value is None: + return None + if not isinstance(value, str): + raise ValueError("Invalid source display name.") + name = value.strip() + if not name or len(name) > MAX_DISPLAY_NAME_LENGTH: + raise ValueError("Invalid source display name.") + if any(unicodedata.category(char).startswith("C") for char in name): + raise ValueError("Invalid source display name.") + if any(char in name for char in ("/", "\\", "?", "#", ":", "@")): + raise ValueError("Source display name cannot contain a path or URL.") + return name + + +def _public_source(row: sqlite3.Row) -> dict[str, Any]: + return { + "id": int(row["id"]), + "kind": str(row["kind"]), + "display_name": row["display_name"], + "created_at": str(row["created_at"]), + "updated_at": str(row["updated_at"]), + } + + +def create_source( + conn: sqlite3.Connection, + *, + profile_id: int, + kind: str, + display_name: str | None = None, +) -> int: + """Create a profile-owned source; caller owns the transaction.""" + _profile(conn, profile_id) + if type(kind) is not str or kind not in SOURCE_KINDS: + raise ValueError("Unsupported collection source kind.") + name = _display_name(display_name) + now = db.utc_now() + cursor = conn.execute( + """ + INSERT INTO station_sources ( + profile_id, kind, source_ref, display_name, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?) + """, + (profile_id, kind, secrets.token_urlsafe(24), name, now, now), + ) + source_id = cursor.lastrowid + if source_id is None or source_id <= 0: + raise RuntimeError("Could not create station source.") + return source_id + + +def get_source(conn: sqlite3.Connection, *, profile_id: int, source_id: int) -> dict[str, Any]: + """Return owned source metadata without revealing its opaque reference.""" + return _public_source(_source(conn, profile_id, source_id)) + + +def list_sources(conn: sqlite3.Connection, *, profile_id: int) -> list[dict[str, Any]]: + """List profile-owned sources without revealing opaque references.""" + _profile(conn, profile_id) + rows = conn.execute( + "SELECT id, kind, display_name, created_at, updated_at " + "FROM station_sources WHERE profile_id = ? ORDER BY id", + (profile_id,), + ).fetchall() + return [_public_source(row) for row in rows] + + +def record_observation( + conn: sqlite3.Connection, + *, + profile_id: int, + source_id: int, + station_id: int, +) -> None: + """Observe one existing station; never modify its shared station record.""" + _source(conn, profile_id, source_id) + if type(station_id) is not int or station_id <= 0: + raise ValueError("An existing station is required.") + if conn.execute("SELECT 1 FROM stations WHERE id = ?", (station_id,)).fetchone() is None: + raise ValueError("The selected station no longer exists.") + now = db.utc_now() + conn.execute( + """ + INSERT INTO station_source_memberships ( + source_id, station_id, first_observed_at, last_observed_at + ) VALUES (?, ?, ?, ?) + ON CONFLICT(source_id, station_id) + DO UPDATE SET last_observed_at = excluded.last_observed_at + """, + (source_id, station_id, now, now), + ) + conn.execute( + "UPDATE station_sources SET updated_at = ? WHERE id = ? AND profile_id = ?", + (now, source_id, profile_id), + ) + + +def list_station_sources( + conn: sqlite3.Connection, *, profile_id: int, station_id: int +) -> list[dict[str, Any]]: + """Return only this profile's observations of a shared station.""" + _profile(conn, profile_id) + if type(station_id) is not int or station_id <= 0: + raise ValueError("An existing station is required.") + rows = conn.execute( + """ + SELECT src.id, src.kind, src.display_name, + src.created_at, src.updated_at, + rel.first_observed_at, rel.last_observed_at + FROM station_sources AS src + JOIN station_source_memberships AS rel ON rel.source_id = src.id + WHERE src.profile_id = ? AND rel.station_id = ? + ORDER BY src.id + """, + (profile_id, station_id), + ).fetchall() + return [ + { + **_public_source(row), + "first_observed_at": str(row["first_observed_at"]), + "last_observed_at": str(row["last_observed_at"]), + } + for row in rows + ] + + +def list_source_stations( + conn: sqlite3.Connection, *, profile_id: int, source_id: int +) -> list[dict[str, Any]]: + """List station identities observed in one owned source.""" + _source(conn, profile_id, source_id) + rows = conn.execute( + """ + SELECT s.id, s.name, + rel.first_observed_at, rel.last_observed_at + FROM station_source_memberships AS rel + JOIN stations AS s ON s.id = rel.station_id + JOIN station_sources AS src ON src.id = rel.source_id + WHERE src.id = ? AND src.profile_id = ? + ORDER BY s.id + """, + (source_id, profile_id), + ).fetchall() + return [dict(row) for row in rows] + + +def delete_source(conn: sqlite3.Connection, *, profile_id: int, source_id: int) -> bool: + """Delete an owned source and observations, never shared stations.""" + _source(conn, profile_id, source_id) + cursor = conn.execute( + "DELETE FROM station_sources WHERE id = ? AND profile_id = ?", + (source_id, profile_id), + ) + return cursor.rowcount > 0 diff --git a/fluxtuner/gui/gtk_m3u_import.py b/fluxtuner/gui/gtk_m3u_import.py index f97a776..7f621cc 100644 --- a/fluxtuner/gui/gtk_m3u_import.py +++ b/fluxtuner/gui/gtk_m3u_import.py @@ -18,6 +18,7 @@ M3UImportResult, M3UPreview, import_m3u, + import_m3u_with_new_source, load_station_collection, ) from fluxtuner.network.network_policy import SystemStreamTargetResolver @@ -32,6 +33,8 @@ class M3UImportSession: existing_urls: frozenset[str] source_name: str profile_is_current: Callable[[], bool] + local_source: bool = False + remote_source: bool = False _closed: bool = False def cancel(self) -> None: @@ -58,12 +61,24 @@ def commit(self, name: str, *, existing: bool) -> M3UImportResult: raise ValueError("The selected playlist no longer exists.") if not existing and destination is not None: raise ValueError("This playlist already exists. Select it from the list.") - result = import_m3u( - conn, - profile_id=self.profile_id, - playlist_name=name, - entries=self.preview.entries, - ) + if self.local_source and self.remote_source: + raise ValueError("Conflicting collection source context.") + if self.local_source or self.remote_source: + outcome = import_m3u_with_new_source( + conn, + profile_id=self.profile_id, + playlist_name=name, + entries=self.preview.entries, + kind="remote_collection" if self.remote_source else "local_collection", + ) + result = outcome.result + else: + result = import_m3u( + conn, + profile_id=self.profile_id, + playlist_name=name, + entries=self.preview.entries, + ) conn.commit() self._closed = True return result @@ -81,6 +96,7 @@ def prepare_m3u_import( return prepare_m3u_preview( preview, source_name=Path(path).name, + local_source=True, profile_id=profile_id, profile_is_current=profile_is_current, ) @@ -103,6 +119,7 @@ def prepare_remote_m3u_import( source_name=source_name, profile_id=profile_id, profile_is_current=profile_is_current, + remote_source=True, ) @@ -112,6 +129,8 @@ def prepare_m3u_preview( source_name: str, profile_id: int, profile_is_current: Callable[[], bool], + local_source: bool = False, + remote_source: bool = False, ) -> M3UImportSession: """Create a session from an already loaded local or remote preview.""" if not profile_is_current(): @@ -143,6 +162,8 @@ def prepare_m3u_preview( existing_urls=urls, source_name=source_name, profile_is_current=profile_is_current, + local_source=local_source, + remote_source=remote_source, ) diff --git a/fluxtuner/gui/window.py b/fluxtuner/gui/window.py index 3602528..dd85c36 100644 --- a/fluxtuner/gui/window.py +++ b/fluxtuner/gui/window.py @@ -620,6 +620,7 @@ def respond(current: Gtk.Dialog, response: int) -> None: profile_id=profile_id, name=name_entry.get_text(), url=url_entry.get_text(), + record_manual_provenance=True, ) conn.commit() except Exception as exc: # noqa: BLE001 - contained GTK storage failure. diff --git a/fluxtuner/tui_custom_stations.py b/fluxtuner/tui_custom_stations.py index aa468b9..b294608 100644 --- a/fluxtuner/tui_custom_stations.py +++ b/fluxtuner/tui_custom_stations.py @@ -108,6 +108,7 @@ def save_station(self) -> None: profile_id=self._profile_id, name=self.query_one("#custom-station-name", Input).value, url=self.query_one("#custom-station-url", Input).value, + record_manual_provenance=True, ) conn.commit() except Exception as exc: # noqa: BLE001 - contained form/storage failure. diff --git a/fluxtuner/tui_m3u_import.py b/fluxtuner/tui_m3u_import.py index 883fe71..6688fe0 100644 --- a/fluxtuner/tui_m3u_import.py +++ b/fluxtuner/tui_m3u_import.py @@ -23,6 +23,7 @@ M3UImportResult, M3UPreview, import_m3u, + import_m3u_with_new_source, load_m3u_file, load_station_collection, ) @@ -65,6 +66,9 @@ def _save_import( name: str, existing: bool, profile_is_current: Callable[[], bool], + *, + local_source: bool = False, + remote_source: bool = False, ) -> M3UImportOutcome: if not profile_is_current(): raise ValueError("The profile changed. Reopen the import dialog.") @@ -78,12 +82,24 @@ def _save_import( raise ValueError( "A playlist with this name already exists. Choose it as an existing destination." ) - result = import_m3u( - conn, - profile_id=profile_id, - playlist_name=name, - entries=preview.entries, - ) + if local_source and remote_source: + raise ValueError("Conflicting collection source context.") + if local_source or remote_source: + outcome = import_m3u_with_new_source( + conn, + profile_id=profile_id, + playlist_name=name, + entries=preview.entries, + kind="remote_collection" if remote_source else "local_collection", + ) + result = outcome.result + else: + result = import_m3u( + conn, + profile_id=profile_id, + playlist_name=name, + entries=preview.entries, + ) conn.commit() return M3UImportOutcome(name, result) @@ -142,6 +158,8 @@ def __init__( self._profile_id = profile_id self._profile_is_current = profile_is_current self._preview: M3UPreview | None = None + self._local_source = False + self._remote_source = False self._names: tuple[str, ...] = () self._generation = 0 self._pending = "" @@ -222,6 +240,8 @@ def path_changed(self) -> None: return self._generation += 1 self._preview = None + self._local_source = False + self._remote_source = False self._names = () self._pending = "" self._message("") @@ -249,6 +269,8 @@ def preview_file(self) -> None: return self._generation += 1 self._preview = None + self._local_source = False + self._remote_source = False self._names = () self._pending = "preview" self._controls() @@ -272,6 +294,8 @@ async def _read_file(self, path: Path | str, generation: int) -> None: if not self._profile_is_current(): raise ValueError("The profile changed. Reopen the import dialog.") self._preview, self._names = preview, names + self._local_source = isinstance(path, Path) + self._remote_source = isinstance(path, str) select = self.query_one("#m3u-destination", Select) select.set_options( [ @@ -312,6 +336,8 @@ async def _read_file(self, path: Path | str, generation: int) -> None: except Exception as exc: # noqa: BLE001 - contained file/profile failure. if self._current(generation): self._preview = None + self._local_source = False + self._remote_source = False self._message(str(exc)) finally: if self._current(generation): @@ -340,7 +366,14 @@ def save_import(self) -> None: self._pending = "save" self._controls() self._message("Importing stations…") - self._save_file(self._preview, name, existing, self._generation) + self._save_file( + self._preview, + name, + existing, + self._generation, + local_source=self._local_source, + remote_source=self._remote_source, + ) @work(exclusive=True, group="m3u-save") async def _save_file( @@ -349,6 +382,9 @@ async def _save_file( name: str, existing: bool, generation: int, + *, + local_source: bool = False, + remote_source: bool = False, ) -> None: try: outcome = await asyncio.to_thread( @@ -358,6 +394,8 @@ async def _save_file( name, existing, self._profile_is_current, + local_source=local_source, + remote_source=remote_source, ) if self._current(generation): self._import_closed = True diff --git a/fluxtuner/web/library.py b/fluxtuner/web/library.py index c22fca0..2650c8c 100644 --- a/fluxtuner/web/library.py +++ b/fluxtuner/web/library.py @@ -278,7 +278,9 @@ def create_custom_station_payload( raise LookupError("The selected profile is unavailable.") profile_id = int(profile["id"]) - added = add_custom_station(conn, profile_id=profile_id, name=name, url=url) + added = add_custom_station( + conn, profile_id=profile_id, name=name, url=url, record_manual_provenance=True + ) saved = next( station for station in list_favorites(conn, profile_id=profile_id) diff --git a/fluxtuner/web/m3u_import.py b/fluxtuner/web/m3u_import.py index 00451b9..4ccae60 100644 --- a/fluxtuner/web/m3u_import.py +++ b/fluxtuner/web/m3u_import.py @@ -9,8 +9,13 @@ from fluxtuner.core import db from fluxtuner.core.favorites import list_favorites -from fluxtuner.core.m3u_import import import_m3u, parse_station_collection +from fluxtuner.core.m3u_import import ( + import_m3u, + import_m3u_with_new_source, + parse_station_collection, +) from fluxtuner.core.profiles import get_profile_by_name +from fluxtuner.web import remote_import_receipts as receipts class DestinationConflict(ValueError): @@ -54,6 +59,8 @@ def import_payload( profile_name: str | None, playlist_name: str, existing: bool, + remote_receipt: str | None = None, + session_token: str | None = None, ) -> dict[str, Any]: with closing(db.connect()) as conn, conn: conn.execute("BEGIN IMMEDIATE") @@ -68,11 +75,28 @@ def import_payload( if not existing and destination is not None: raise DestinationConflict("A playlist with this name already exists.") - result = import_m3u( - conn, - profile_id=profile_id, - playlist_name=playlist_name, - entries=preview.entries, - ) + if remote_receipt is not None: + receipts.consume( + conn, + token=remote_receipt, + user_id=user_id, + profile_id=profile_id, + session_token=session_token, + data=data, + ) + result = import_m3u_with_new_source( + conn, + profile_id=profile_id, + playlist_name=playlist_name, + entries=preview.entries, + kind="remote_collection", + ).result + else: + result = import_m3u( + conn, + profile_id=profile_id, + playlist_name=playlist_name, + entries=preview.entries, + ) conn.commit() return {"status": "ok", "name": playlist_name.strip(), **asdict(result)} diff --git a/fluxtuner/web/remote_import_receipts.py b/fluxtuner/web/remote_import_receipts.py new file mode 100644 index 0000000..01999e1 --- /dev/null +++ b/fluxtuner/web/remote_import_receipts.py @@ -0,0 +1,100 @@ +# SPDX-License-Identifier: LicenseRef-FluxTuner-Web-NC +"""One-use, profile-scoped attestations for Web remote station downloads.""" + +from __future__ import annotations + +import hashlib +import secrets +import sqlite3 +from contextlib import closing +from datetime import UTC, datetime, timedelta + +from fluxtuner.core import db +from fluxtuner.core.profiles import get_profile_by_name + +RECEIPT_LIFETIME = timedelta(minutes=15) + + +def utc_now(): + return datetime.now(UTC) + + +def encode_datetime(value): + return value.isoformat() + + +MAX_TOKEN_LENGTH = 128 + + +def _digest(data: bytes | str) -> str: + raw = data.encode("utf-8") if isinstance(data, str) else data + return hashlib.sha256(raw).hexdigest() + + +def issue_for_default_profile(*, user_id: int, session_token: str, data: bytes) -> str: + """Compatibility wrapper for callers using the default profile.""" + return issue_for_profile( + user_id=user_id, profile_name=None, session_token=session_token, data=data + ) + + +def issue_for_profile( + *, user_id: int, profile_name: str | None, session_token: str, data: bytes +) -> str: + """Persist attestation; never retain URL, raw session token or playlist bytes.""" + if not session_token or not data: + raise ValueError("Authenticated snapshot required.") + token = secrets.token_urlsafe(32) + now = utc_now() + with closing(db.connect()) as conn, conn: + conn.execute("BEGIN IMMEDIATE") + profile = get_profile_by_name( + conn, db.DEFAULT_PROFILE_NAME if profile_name is None else profile_name, user_id=user_id + ) + if profile is None: + raise ValueError("The selected profile is unavailable.") + conn.execute( + "DELETE FROM remote_import_receipts WHERE expires_at < ?", + (encode_datetime(now),), + ) + conn.execute( + "INSERT INTO remote_import_receipts " + "(token_hash, user_id, profile_id, session_hash, payload_hash, expires_at) " + "VALUES (?, ?, ?, ?, ?, ?)", + ( + _digest(token), + user_id, + int(profile["id"]), + _digest(session_token), + _digest(data), + encode_datetime(now + RECEIPT_LIFETIME), + ), + ) + conn.commit() + return token + + +def consume( + conn: sqlite3.Connection, + *, + token: str, + user_id: int, + profile_id: int, + session_token: str | None, + data: bytes, +) -> None: + """Consume only a matching unused receipt inside the caller transaction.""" + if not isinstance(token, str) or not 1 <= len(token) <= MAX_TOKEN_LENGTH: + raise ValueError("Invalid remote collection receipt.") + if not session_token: + raise ValueError("Invalid remote collection receipt.") + now = encode_datetime(utc_now()) + cursor = conn.execute( + "UPDATE remote_import_receipts SET consumed_at = ? " + "WHERE token_hash = ? AND user_id = ? AND profile_id = ? " + "AND session_hash = ? AND payload_hash = ? " + "AND consumed_at IS NULL AND expires_at > ?", + (now, _digest(token), user_id, profile_id, _digest(session_token), _digest(data), now), + ) + if cursor.rowcount != 1: + raise ValueError("Invalid or expired remote collection receipt.") diff --git a/fluxtuner/web/routes/imports.py b/fluxtuner/web/routes/imports.py index 128cbbb..98e93dc 100644 --- a/fluxtuner/web/routes/imports.py +++ b/fluxtuner/web/routes/imports.py @@ -12,6 +12,8 @@ from fluxtuner.network.remote_station_collection import fetch_remote_station_collection from fluxtuner.web import guards from fluxtuner.web import m3u_import as service +from fluxtuner.web import remote_import_receipts as receipts +from fluxtuner.web.security import SESSION_COOKIE_NAME router = APIRouter() @@ -68,9 +70,11 @@ async def _read_remote_url(request: Request) -> str: @router.post("/api/stations/import/remote") -async def fetch_remote_import(request: Request) -> Response: +async def fetch_remote_import( + request: Request, profile: str | None = Query(default=None, max_length=80) +) -> Response: """Return one validated byte snapshot; the browser owns preview and confirmation.""" - _authorize(request) + user_id = _authorize(request) url = await _read_remote_url(request) try: data = await asyncio.to_thread( @@ -82,9 +86,18 @@ async def fetch_remote_import(request: Request) -> Response: raise HTTPException( status_code=502, detail="Could not retrieve the remote collection." ) from exc - return Response( - content=data, media_type="application/octet-stream", headers={"Cache-Control": "no-store"} - ) + headers = {"Cache-Control": "no-store"} + session_token = getattr(request, "cookies", {}).get(SESSION_COOKIE_NAME) + if session_token: + receipt = await asyncio.to_thread( + receipts.issue_for_profile, + user_id=user_id, + profile_name=profile, + session_token=session_token, + data=data, + ) + headers["X-FluxTuner-Remote-Receipt"] = receipt + return Response(content=data, media_type="application/octet-stream", headers=headers) @router.post("/api/stations/import/preview") @@ -113,6 +126,8 @@ async def confirm_import( ) -> dict[str, Any]: user_id = _authorize(request) data = await _read_upload(request) + remote_receipt = request.headers.get("X-FluxTuner-Remote-Receipt") + session_token = getattr(request, "cookies", {}).get(SESSION_COOKIE_NAME) try: return await asyncio.to_thread( service.import_payload, @@ -121,6 +136,8 @@ async def confirm_import( profile_name=profile, playlist_name=playlist, existing=existing, + remote_receipt=remote_receipt, + session_token=session_token, ) except LookupError as exc: raise HTTPException(status_code=404, detail=str(exc)) from None diff --git a/fluxtuner/web/static/js/station-imports.js b/fluxtuner/web/static/js/station-imports.js index f32e2a2..7f2d79c 100644 --- a/fluxtuner/web/static/js/station-imports.js +++ b/fluxtuner/web/static/js/station-imports.js @@ -3,7 +3,7 @@ */ export function createStationImportController({ - apiFetch, elements, getCurrentUser, getSessionKey, beforeOpen, onSaved, + apiFetch, elements, getCurrentUser, getSessionKey, getProfileName = () => null, beforeOpen, onSaved, }) { const { openButton, dialog, form, fileInput, urlInput, previewButton, previewNode, @@ -14,6 +14,7 @@ export function createStationImportController({ let generation = 0; let context = null; let bytes = null; + let remoteReceipt = null; let playlists = []; let pending = ""; let bound = false; @@ -22,7 +23,8 @@ export function createStationImportController({ return Boolean( captured && getCurrentUser()?.id === captured.userId && - getSessionKey() === captured.sessionKey + getSessionKey() === captured.sessionKey && + (getProfileName() || null) === captured.profileName ); } @@ -46,6 +48,7 @@ export function createStationImportController({ function clearPreview() { bytes = null; + remoteReceipt = null; playlists = []; selectNode.replaceChildren(); selectNode.value = ""; @@ -81,7 +84,7 @@ export function createStationImportController({ if (!user || dialog.open) return; beforeOpen?.(); generation += 1; - context = { userId: user.id, sessionKey: getSessionKey() }; + context = { userId: user.id, sessionKey: getSessionKey(), profileName: getProfileName() || null }; pending = ""; form.reset(); clearPreview(); @@ -97,6 +100,12 @@ export function createStationImportController({ messageNode.textContent = ""; } + function profileEndpoint(endpoint, captured) { + if (!captured?.profileName) return endpoint; + const join = endpoint.includes("?") ? "&" : "?"; + return `${endpoint}${join}profile=${encodeURIComponent(captured.profileName)}`; + } + async function post(endpoint, body) { return apiFetch(endpoint, { method: "POST", @@ -183,7 +192,7 @@ export function createStationImportController({ try { let data; if (url) { - const download = await apiFetch("/api/stations/import/remote", { + const download = await apiFetch(profileEndpoint("/api/stations/import/remote", requestContext), { method: "POST", headers: { Accept: "application/octet-stream", "Content-Type": "application/json" }, body: JSON.stringify({ url }), @@ -198,13 +207,14 @@ export function createStationImportController({ throw new Error(detail); } data = await download.arrayBuffer(); + remoteReceipt = download.headers?.get?.("X-FluxTuner-Remote-Receipt") || null; if (!current(requestGeneration, requestContext)) return; if (data.byteLength > MAX_BYTES) throw new Error("The playlist exceeds 1 MiB."); } else { data = await file.arrayBuffer(); } if (!current(requestGeneration, requestContext)) return; - const response = await post("/api/stations/import/preview", data); + const response = await post(profileEndpoint("/api/stations/import/preview", requestContext), data); if (!current(requestGeneration, requestContext)) return; const payload = await readPayload(response); if (!current(requestGeneration, requestContext)) return; @@ -253,7 +263,18 @@ export function createStationImportController({ messageNode.textContent = "Importing stations…"; try { const query = new URLSearchParams({ playlist: name, existing: String(existing) }); - const response = await post(`/api/stations/import?${query}`, bytes); + const endpoint = profileEndpoint(`/api/stations/import?${query}`, requestContext); + const response = remoteReceipt + ? await apiFetch(endpoint, { + method: "POST", + headers: { + Accept: "application/json", + "Content-Type": "application/octet-stream", + "X-FluxTuner-Remote-Receipt": remoteReceipt, + }, + body: bytes, + }) + : await post(endpoint, bytes); if (!current(requestGeneration, requestContext)) return; const payload = await readPayload(response); if (!current(requestGeneration, requestContext)) return; diff --git a/tests/test_db.py b/tests/test_db.py index ff46566..4d2368f 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -34,6 +34,9 @@ def test_init_db_creates_expected_tables( with db.connect() as conn: assert db.table_names(conn) == { + "remote_import_receipts", + "station_sources", + "station_source_memberships", "web_reauth_attempts", "favorites", "history_entries", @@ -535,7 +538,7 @@ def test_fresh_database_contains_auth_retention_indexes(tmp_path) -> None: assert "idx_web_recovery_attempts_attempted_at" in indexes -def test_schema_v0_upgrades_through_v3(tmp_path) -> None: +def test_schema_v0_upgrades_through_v4(tmp_path) -> None: db_path = tmp_path / "fluxtuner.db" with db.connect(db_path) as conn: @@ -561,6 +564,6 @@ def test_schema_v0_upgrades_through_v3(tmp_path) -> None: } assert version == db.CURRENT_SCHEMA_VERSION - assert version == 3 + assert version == 4 assert "idx_web_login_attempts_attempted_at" in indexes assert "idx_web_recovery_attempts_attempted_at" in indexes diff --git a/tests/test_gtk_m3u_import.py b/tests/test_gtk_m3u_import.py index 2cb79a0..a9f1cd2 100644 --- a/tests/test_gtk_m3u_import.py +++ b/tests/test_gtk_m3u_import.py @@ -144,18 +144,18 @@ def test_profile_change_rejects_save_without_writing(library): def test_storage_failure_rolls_back_and_leaves_session_retryable(library, monkeypatch): session = prepare(library) before = counts() - original = gtk_m3u_import.import_m3u + original = gtk_m3u_import.import_m3u_with_new_source def fail(conn, **kwargs): db.create_playlist_record(conn, "Should roll back", profile_id=library.profile_id) raise RuntimeError("simulated storage failure") - monkeypatch.setattr(gtk_m3u_import, "import_m3u", fail) + monkeypatch.setattr(gtk_m3u_import, "import_m3u_with_new_source", fail) with pytest.raises(RuntimeError, match="simulated"): session.commit("Jazz", existing=False) assert counts() == before - monkeypatch.setattr(gtk_m3u_import, "import_m3u", original) + monkeypatch.setattr(gtk_m3u_import, "import_m3u_with_new_source", original) assert session.commit("Jazz", existing=False).added_to_playlist == 2 @@ -293,3 +293,61 @@ def test_window_remote_menu_captures_profile(monkeypatch): assert opener.call_args.kwargs["profile_id"] == 7 harness._playlist_profile_id.return_value = 8 assert opener.call_args.kwargs["profile_is_current"]() is False + + +def test_local_commit_persists_profile_scoped_provenance(library): + session = prepare(library) + assert session.local_source is True + + with db.connect() as conn: + assert conn.execute("SELECT COUNT(*) FROM station_sources").fetchone()[0] == 0 + + result = session.commit("Jazz", existing=False) + assert result.added_to_playlist == 2 + + with db.connect() as conn: + sources = conn.execute( + "SELECT id, profile_id, kind, display_name FROM station_sources" + ).fetchall() + assert len(sources) == 1 + assert sources[0]["profile_id"] == library.profile_id + assert sources[0]["kind"] == "local_collection" + assert sources[0]["display_name"] is None + assert conn.execute("SELECT COUNT(*) FROM station_source_memberships").fetchone()[0] == 2 + + +def test_remote_commit_records_remote_provenance(library, monkeypatch): + preview = prepare(library).preview + monkeypatch.setattr( + gtk_m3u_import, + "preview_remote_station_collection", + lambda *_args, **_kwargs: (b"snapshot", preview), + ) + session = gtk_m3u_import.prepare_remote_m3u_import( + "https://catalog.example/jazz.m3u", + profile_id=library.profile_id, + profile_is_current=lambda: True, + ) + assert session.local_source is False + assert session.remote_source is True + + session.commit("Remote Jazz", existing=False) + + with db.connect() as conn: + sources = conn.execute( + "SELECT profile_id, kind, display_name FROM station_sources" + ).fetchall() + assert len(sources) == 1 + assert sources[0]["profile_id"] == library.profile_id + assert sources[0]["kind"] == "remote_collection" + assert sources[0]["display_name"] is None + assert conn.execute("SELECT COUNT(*) FROM station_source_memberships").fetchone()[0] == 2 + + +def test_local_cancel_creates_no_provenance(library): + session = prepare(library) + session.cancel() + + with db.connect() as conn: + assert conn.execute("SELECT COUNT(*) FROM station_sources").fetchone()[0] == 0 + assert conn.execute("SELECT COUNT(*) FROM station_source_memberships").fetchone()[0] == 0 diff --git a/tests/test_gtk_orchestration.py b/tests/test_gtk_orchestration.py index c26e0fc..d4e1f9e 100644 --- a/tests/test_gtk_orchestration.py +++ b/tests/test_gtk_orchestration.py @@ -2095,6 +2095,7 @@ def test_custom_station_dialog_saves_captured_profile_and_refreshes(monkeypatch, profile_id=7, name="Personal radio", url="https://radio.example/live", + record_manual_provenance=True, ) conn.commit.assert_called_once_with() dialog.destroy.assert_called_once_with() diff --git a/tests/test_m3u_import_new_source.py b/tests/test_m3u_import_new_source.py new file mode 100644 index 0000000..c08fae3 --- /dev/null +++ b/tests/test_m3u_import_new_source.py @@ -0,0 +1,167 @@ +# SPDX-License-Identifier: MIT +"""Atomic creation of a local provenance source during import.""" + +from dataclasses import asdict + +import pytest + +from fluxtuner.core import db, m3u_import +from fluxtuner.core import station_provenance as provenance + + +@pytest.fixture +def library(tmp_path): + path = tmp_path / "new-source.db" + db.init_db(path) + with db.connect(path) as conn: + profile_id = db.ensure_default_profile(conn) + conn.commit() + yield conn, profile_id + + +def _entries(): + return m3u_import.parse_m3u( + b"#EXTM3U\n" + b"#EXTINF:-1,One\nhttps://one.example/live\n" + b"#EXTINF:-1,Two\nhttps://two.example/live\n" + ).entries + + +def _count(conn, table): + assert table in { + "station_sources", + "station_source_memberships", + "stations", + "favorites", + "playlists", + "playlist_stations", + } + return conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0] + + +def test_new_source_and_station_import_are_atomic(library): + conn, profile_id = library + outcome = m3u_import.import_m3u_with_new_source( + conn, + profile_id=profile_id, + playlist_name="Jazz", + entries=_entries(), + display_name="Jazz collection", + ) + assert outcome.result.added_favorites == 2 + assert outcome.result.added_to_playlist == 2 + assert _count(conn, "station_sources") == 1 + assert _count(conn, "station_source_memberships") == 2 + assert ( + provenance.get_source(conn, profile_id=profile_id, source_id=outcome.source_id)["kind"] + == "local_collection" + ) + assert ( + len( + provenance.list_source_stations( + conn, profile_id=profile_id, source_id=outcome.source_id + ) + ) + == 2 + ) + assert conn.in_transaction + + +def test_legacy_result_contract_remains_four_fields(library): + conn, profile_id = library + outcome = m3u_import.import_m3u_with_new_source( + conn, + profile_id=profile_id, + playlist_name="Jazz", + entries=_entries(), + ) + assert set(asdict(outcome.result)) == { + "added_favorites", + "existing_favorites", + "added_to_playlist", + "already_in_playlist", + } + assert asdict(outcome.result)["added_favorites"] == 2 + + +def test_invalid_import_does_not_leave_an_empty_source(library): + conn, profile_id = library + with pytest.raises(ValueError, match="no valid stations"): + m3u_import.import_m3u_with_new_source( + conn, + profile_id=profile_id, + playlist_name="Jazz", + entries=(), + ) + assert _count(conn, "station_sources") == 0 + assert _count(conn, "playlists") == 0 + assert _count(conn, "stations") == 0 + + +def test_failure_after_first_observation_reverts_source_and_library(library, monkeypatch): + conn, profile_id = library + db.create_playlist_record(conn, "Earlier work", profile_id=profile_id) + conn.commit() + + original = provenance.record_observation + calls = 0 + + def fail_on_second(*args, **kwargs): + nonlocal calls + calls += 1 + if calls == 2: + raise RuntimeError("injected source import failure") + return original(*args, **kwargs) + + monkeypatch.setattr(m3u_import.provenance, "record_observation", fail_on_second) + + with pytest.raises(RuntimeError, match="injected source import failure"): + m3u_import.import_m3u_with_new_source( + conn, + profile_id=profile_id, + playlist_name="Jazz", + entries=_entries(), + ) + + assert db.get_playlist_record(conn, "Earlier work", profile_id=profile_id) is not None + assert db.get_playlist_record(conn, "Jazz", profile_id=profile_id) is None + for table in ( + "station_sources", + "station_source_memberships", + "stations", + "favorites", + "playlist_stations", + ): + assert _count(conn, table) == 0 + + +def test_caller_rollback_removes_new_source_and_stations(library): + conn, profile_id = library + m3u_import.import_m3u_with_new_source( + conn, + profile_id=profile_id, + playlist_name="Jazz", + entries=_entries(), + ) + conn.rollback() + for table in ( + "station_sources", + "station_source_memberships", + "stations", + "favorites", + "playlists", + ): + assert _count(conn, table) == 0 + + +def test_invalid_profile_cannot_leave_source(library): + conn, _ = library + with pytest.raises(ValueError): + m3u_import.import_m3u_with_new_source( + conn, + profile_id=999999, + playlist_name="Jazz", + entries=_entries(), + ) + assert _count(conn, "station_sources") == 0 + assert _count(conn, "stations") == 0 diff --git a/tests/test_m3u_import_provenance.py b/tests/test_m3u_import_provenance.py new file mode 100644 index 0000000..498c229 --- /dev/null +++ b/tests/test_m3u_import_provenance.py @@ -0,0 +1,190 @@ +# SPDX-License-Identifier: MIT +"""Atomic station provenance during M3U imports.""" + +from __future__ import annotations + +import pytest + +from fluxtuner.core import db, m3u_import +from fluxtuner.core import station_provenance as provenance +from fluxtuner.core.profiles import get_or_create_profile + + +@pytest.fixture +def library(tmp_path): + path = tmp_path / "import-provenance.db" + db.init_db(path) + with db.connect(path) as conn: + first = db.ensure_default_profile(conn) + second = get_or_create_profile(conn, "Other") + conn.commit() + yield conn, first, second + + +def _entries(): + return m3u_import.parse_m3u( + b"#EXTM3U\n" + b"#EXTINF:-1,One\nhttps://one.example/live\n" + b"#EXTINF:-1,Two\nhttps://two.example/live\n" + ).entries + + +def _count(conn, table): + assert table in { + "stations", + "favorites", + "playlists", + "playlist_stations", + "station_sources", + "station_source_memberships", + } + return conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0] + + +def test_legacy_import_records_no_provenance(library): + conn, first, _ = library + result = m3u_import.import_m3u( + conn, profile_id=first, playlist_name="Legacy", entries=_entries() + ) + assert result.added_favorites == 2 + assert _count(conn, "stations") == 2 + assert _count(conn, "station_sources") == 0 + assert _count(conn, "station_source_memberships") == 0 + + +def test_import_observes_all_stations_for_owned_source(library): + conn, first, _ = library + source_id = provenance.create_source(conn, profile_id=first, kind="local_collection") + result = m3u_import.import_m3u( + conn, + profile_id=first, + playlist_name="Jazz", + entries=_entries(), + source_id=source_id, + ) + assert result.added_favorites == 2 + assert _count(conn, "station_source_memberships") == 2 + assert len(provenance.list_source_stations(conn, profile_id=first, source_id=source_id)) == 2 + + +def test_reimport_into_another_playlist_reuses_observations(library): + conn, first, _ = library + source_id = provenance.create_source(conn, profile_id=first, kind="remote_collection") + m3u_import.import_m3u( + conn, + profile_id=first, + playlist_name="First", + entries=_entries(), + source_id=source_id, + ) + before = { + row["station_id"]: row["first_observed_at"] + for row in conn.execute( + "SELECT station_id, first_observed_at FROM station_source_memberships" + ) + } + result = m3u_import.import_m3u( + conn, + profile_id=first, + playlist_name="Second", + entries=_entries(), + source_id=source_id, + ) + assert result.added_favorites == 0 + assert result.added_to_playlist == 2 + assert _count(conn, "station_source_memberships") == 2 + assert { + row["station_id"]: row["first_observed_at"] + for row in conn.execute( + "SELECT station_id, first_observed_at FROM station_source_memberships" + ) + } == before + + +def test_foreign_source_is_rejected_before_library_writes(library): + conn, first, second = library + foreign_source = provenance.create_source(conn, profile_id=second, kind="local_collection") + with pytest.raises(ValueError, match="Source not found"): + m3u_import.import_m3u( + conn, + profile_id=first, + playlist_name="Denied", + entries=_entries(), + source_id=foreign_source, + ) + assert _count(conn, "stations") == 0 + assert _count(conn, "favorites") == 0 + assert _count(conn, "playlists") == 0 + assert _count(conn, "station_source_memberships") == 0 + + +@pytest.mark.parametrize("invalid_source", [0, -1, True, 999999]) +def test_invalid_source_cannot_write_library(library, invalid_source): + conn, first, _ = library + with pytest.raises(ValueError): + m3u_import.import_m3u( + conn, + profile_id=first, + playlist_name="Denied", + entries=_entries(), + source_id=invalid_source, + ) + assert _count(conn, "stations") == 0 + assert _count(conn, "playlists") == 0 + + +def test_failed_provenance_write_rolls_back_entire_import(library, monkeypatch): + conn, first, _ = library + source_id = provenance.create_source(conn, profile_id=first, kind="local_collection") + db.create_playlist_record(conn, "Earlier work", profile_id=first) + conn.commit() + + original = provenance.record_observation + calls = 0 + + def fail_on_second(*args, **kwargs): + nonlocal calls + calls += 1 + if calls == 2: + raise RuntimeError("injected provenance failure") + return original(*args, **kwargs) + + monkeypatch.setattr(m3u_import.provenance, "record_observation", fail_on_second) + + with pytest.raises(RuntimeError, match="injected provenance failure"): + m3u_import.import_m3u( + conn, + profile_id=first, + playlist_name="New", + entries=_entries(), + source_id=source_id, + ) + + assert db.get_playlist_record(conn, "Earlier work", profile_id=first) is not None + assert db.get_playlist_record(conn, "New", profile_id=first) is None + assert _count(conn, "stations") == 0 + assert _count(conn, "favorites") == 0 + assert _count(conn, "playlist_stations") == 0 + assert _count(conn, "station_source_memberships") == 0 + assert _count(conn, "station_sources") == 1 + + +def test_caller_rollback_includes_library_and_provenance(library): + conn, first, _ = library + source_id = provenance.create_source(conn, profile_id=first, kind="remote_collection") + conn.commit() + + m3u_import.import_m3u( + conn, + profile_id=first, + playlist_name="New", + entries=_entries(), + source_id=source_id, + ) + assert _count(conn, "station_source_memberships") == 2 + conn.rollback() + + assert _count(conn, "station_source_memberships") == 0 + assert _count(conn, "stations") == 0 + assert _count(conn, "playlists") == 0 + assert _count(conn, "station_sources") == 1 diff --git a/tests/test_station_provenance.py b/tests/test_station_provenance.py new file mode 100644 index 0000000..e5b88ef --- /dev/null +++ b/tests/test_station_provenance.py @@ -0,0 +1,110 @@ +# SPDX-License-Identifier: MIT +"""Profile isolation and transaction tests for station provenance.""" + +import pytest + +from fluxtuner.core import db +from fluxtuner.core import station_provenance as provenance +from fluxtuner.core.custom_stations import add_custom_station +from fluxtuner.core.profiles import get_or_create_profile + + +@pytest.fixture +def library(tmp_path): + path = tmp_path / "provenance.db" + db.init_db(path) + with db.connect(path) as conn: + first = db.ensure_default_profile(conn) + second = get_or_create_profile(conn, "Other") + add_custom_station(conn, profile_id=first, name="Radio", url="https://example.test/live") + station = conn.execute("SELECT id FROM stations").fetchone()["id"] + conn.commit() + yield conn, first, second, station + + +def test_sources_are_profile_scoped(library): + conn, first, second, station = library + source = provenance.create_source( + conn, profile_id=first, kind="local_collection", display_name="Jazz.m3u" + ) + provenance.record_observation(conn, profile_id=first, source_id=source, station_id=station) + assert len(provenance.list_station_sources(conn, profile_id=first, station_id=station)) == 1 + assert provenance.list_station_sources(conn, profile_id=second, station_id=station) == [] + for operation in ( + lambda: provenance.get_source(conn, profile_id=second, source_id=source), + lambda: provenance.list_source_stations(conn, profile_id=second, source_id=source), + lambda: provenance.record_observation( + conn, profile_id=second, source_id=source, station_id=station + ), + lambda: provenance.delete_source(conn, profile_id=second, source_id=source), + ): + with pytest.raises(ValueError, match="Source not found"): + operation() + assert len(provenance.list_source_stations(conn, profile_id=first, source_id=source)) == 1 + + +def test_repeated_observation_does_not_duplicate_membership(library): + conn, first, _, station = library + source = provenance.create_source(conn, profile_id=first, kind="remote_collection") + before = dict(conn.execute("SELECT * FROM stations WHERE id = ?", (station,)).fetchone()) + provenance.record_observation(conn, profile_id=first, source_id=source, station_id=station) + first_date = conn.execute( + "SELECT first_observed_at FROM station_source_memberships" + ).fetchone()[0] + provenance.record_observation(conn, profile_id=first, source_id=source, station_id=station) + assert conn.execute("SELECT COUNT(*) FROM station_source_memberships").fetchone()[0] == 1 + assert ( + conn.execute("SELECT first_observed_at FROM station_source_memberships").fetchone()[0] + == first_date + ) + assert ( + dict(conn.execute("SELECT * FROM stations WHERE id = ?", (station,)).fetchone()) == before + ) + + +def test_source_deletion_preserves_station_and_favorite(library): + conn, first, _, station = library + source = provenance.create_source(conn, profile_id=first, kind="local_collection") + provenance.record_observation(conn, profile_id=first, source_id=source, station_id=station) + assert provenance.delete_source(conn, profile_id=first, source_id=source) + assert conn.execute("SELECT COUNT(*) FROM stations").fetchone()[0] == 1 + assert conn.execute("SELECT COUNT(*) FROM favorites").fetchone()[0] == 1 + assert conn.execute("SELECT COUNT(*) FROM station_source_memberships").fetchone()[0] == 0 + + +def test_caller_rollback_removes_source_and_observation(library): + conn, first, _, station = library + conn.execute("SAVEPOINT provenance_test") + source = provenance.create_source(conn, profile_id=first, kind="local_collection") + provenance.record_observation(conn, profile_id=first, source_id=source, station_id=station) + conn.execute("ROLLBACK TO SAVEPOINT provenance_test") + conn.execute("RELEASE SAVEPOINT provenance_test") + assert conn.execute("SELECT COUNT(*) FROM station_sources").fetchone()[0] == 0 + assert conn.execute("SELECT COUNT(*) FROM station_source_memberships").fetchone()[0] == 0 + + +@pytest.mark.parametrize("bad_profile", [None, 0, -1, True, 999999]) +def test_invalid_profile_never_creates_a_source(library, bad_profile): + conn, _, _, _ = library + with pytest.raises(ValueError): + provenance.create_source(conn, profile_id=bad_profile, kind="local_collection") + assert conn.execute("SELECT COUNT(*) FROM station_sources").fetchone()[0] == 0 + + +@pytest.mark.parametrize( + "bad_name", + ["https://private.example/feed", "/home/private/list.m3u", "Secret?token=value", "x" * 121], +) +def test_display_name_rejects_paths_and_urls(library, bad_name): + conn, first, _, _ = library + with pytest.raises(ValueError): + provenance.create_source( + conn, profile_id=first, kind="remote_collection", display_name=bad_name + ) + assert conn.execute("SELECT COUNT(*) FROM station_sources").fetchone()[0] == 0 + + +def test_caller_controls_commit(library): + conn, first, _, _ = library + provenance.create_source(conn, profile_id=first, kind="local_collection") + assert conn.in_transaction diff --git a/tests/test_station_provenance_manual_integration.py b/tests/test_station_provenance_manual_integration.py new file mode 100644 index 0000000..6590c6f --- /dev/null +++ b/tests/test_station_provenance_manual_integration.py @@ -0,0 +1,148 @@ +# SPDX-License-Identifier: MIT +"""Manual provenance, explicit reuse and privacy integration.""" + +from __future__ import annotations + +import pytest + +from fluxtuner.core import db +from fluxtuner.core import station_provenance as provenance +from fluxtuner.core.custom_stations import add_custom_station +from fluxtuner.core.m3u_import import import_m3u, parse_station_collection +from fluxtuner.core.profiles import get_or_create_profile + + +@pytest.fixture +def library(tmp_path, monkeypatch): + monkeypatch.setattr(db, "DB_FILE", tmp_path / "manual-provenance.db") + db.init_db() + with db.connect() as conn: + first = db.ensure_default_profile(conn) + second = get_or_create_profile(conn, "Second") + conn.commit() + return first, second + + +def test_manual_creation_adds_one_source_and_observation(library): + first, _ = library + with db.connect() as conn: + assert add_custom_station( + conn, + profile_id=first, + name="Personal", + url="https://radio.example/live?token=private", + record_manual_provenance=True, + ) + sources = provenance.list_sources(conn, profile_id=first) + assert len(sources) == 1 + assert sources[0]["kind"] == "manual_entry" + assert sources[0]["display_name"] is None + stations = provenance.list_source_stations( + conn, profile_id=first, source_id=sources[0]["id"] + ) + assert len(stations) == 1 + assert "station_key" not in stations[0] + assert "token=private" not in str(stations) + conn.commit() + + +def test_duplicate_manual_entry_does_not_create_second_source(library): + first, _ = library + with db.connect() as conn: + args = { + "profile_id": first, + "name": "Personal", + "url": "https://radio.example/live", + "record_manual_provenance": True, + } + assert add_custom_station(conn, **args) + assert not add_custom_station(conn, **args) + assert len(provenance.list_sources(conn, profile_id=first)) == 1 + assert conn.execute("SELECT COUNT(*) FROM station_source_memberships").fetchone()[0] == 1 + + +def test_manual_sources_are_isolated_by_profile(library): + first, second = library + with db.connect() as conn: + for owner in (first, second): + assert add_custom_station( + conn, + profile_id=owner, + name="Radio", + url="https://radio.example/live", + record_manual_provenance=True, + ) + assert len(provenance.list_sources(conn, profile_id=first)) == 1 + assert len(provenance.list_sources(conn, profile_id=second)) == 1 + assert conn.execute("SELECT COUNT(*) FROM stations").fetchone()[0] == 1 + + +def test_failure_rolls_back_manual_station_and_source(library, monkeypatch): + first, _ = library + + def fail(*args, **kwargs): + raise RuntimeError("injected observation failure") + + monkeypatch.setattr(provenance, "record_observation", fail) + + with db.connect() as conn: + with pytest.raises(RuntimeError, match="injected observation"): + add_custom_station( + conn, + profile_id=first, + name="Radio", + url="https://radio.example/live", + record_manual_provenance=True, + ) + for table in ("stations", "favorites", "station_sources", "station_source_memberships"): + assert conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0] == 0 + + +def test_legacy_import_does_not_create_manual_source(library): + first, _ = library + entries = parse_station_collection( + b"#EXTM3U\n#EXTINF:-1,Radio\nhttps://radio.example/live\n" + ).entries + with db.connect() as conn: + result = import_m3u(conn, profile_id=first, playlist_name="Imported", entries=entries) + assert result.added_to_playlist == 1 + assert provenance.list_sources(conn, profile_id=first) == [] + + +def test_reuse_existing_source_retains_one_identity(library): + first, _ = library + entries = parse_station_collection( + b"#EXTM3U\n#EXTINF:-1,Radio\nhttps://radio.example/live\n" + ).entries + with db.connect() as conn: + source_id = provenance.create_source(conn, profile_id=first, kind="local_collection") + for playlist_name in ("Morning", "Evening"): + import_m3u( + conn, + profile_id=first, + playlist_name=playlist_name, + entries=entries, + source_id=source_id, + ) + assert [src["id"] for src in provenance.list_sources(conn, profile_id=first)] == [source_id] + assert ( + len(provenance.list_source_stations(conn, profile_id=first, source_id=source_id)) == 1 + ) + + +def test_foreign_source_cannot_be_reused(library): + first, second = library + entries = parse_station_collection( + b"#EXTM3U\n#EXTINF:-1,Radio\nhttps://radio.example/live\n" + ).entries + with db.connect() as conn: + foreign = provenance.create_source(conn, profile_id=second, kind="local_collection") + with pytest.raises(ValueError, match="Source not found"): + import_m3u( + conn, + profile_id=first, + playlist_name="Denied", + entries=entries, + source_id=foreign, + ) + assert provenance.list_sources(conn, profile_id=first) == [] diff --git a/tests/test_station_provenance_migration.py b/tests/test_station_provenance_migration.py new file mode 100644 index 0000000..a3d7766 --- /dev/null +++ b/tests/test_station_provenance_migration.py @@ -0,0 +1,157 @@ +# SPDX-License-Identifier: MIT +"""Regression coverage for station provenance schema migration.""" + +from concurrent.futures import ThreadPoolExecutor + +import pytest + +from fluxtuner.core import db + + +def _make_v3_database(path): + """Create a representative v3 schema without provenance tables.""" + db.init_db(path) + + with db.connect(path) as conn: + conn.execute("DROP TABLE station_source_memberships") + conn.execute("DROP TABLE station_sources") + conn.execute("DROP TABLE remote_import_receipts") + conn.execute("PRAGMA user_version = 3") + conn.commit() + + +def _schema_objects(conn): + return { + (row["type"], row["name"], row["sql"]) + for row in conn.execute( + """ + SELECT type, name, sql + FROM sqlite_master + WHERE type IN ('table', 'index') + AND name NOT LIKE 'sqlite_%' + """ + ) + } + + +def test_v3_upgrade_preserves_existing_library_data(tmp_path): + path = tmp_path / "upgrade.db" + _make_v3_database(path) + + with db.connect(path) as conn: + profile_id = db.ensure_default_profile(conn) + now = db.utc_now() + + conn.execute( + """ + INSERT INTO stations ( + station_key, name, url, url_resolved, + metadata_json, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?) + """, + ( + "https://example.test/live", + "Existing station", + "https://example.test/live", + "https://example.test/live", + "{}", + now, + now, + ), + ) + station_id = conn.execute( + "SELECT id FROM stations WHERE station_key = ?", + ("https://example.test/live",), + ).fetchone()["id"] + + conn.execute( + """ + INSERT INTO favorites ( + profile_id, station_id, favorite_tags_json, + created_at, updated_at + ) VALUES (?, ?, ?, ?, ?) + """, + (profile_id, station_id, '["Keep"]', now, now), + ) + conn.commit() + + station_before = tuple( + conn.execute("SELECT * FROM stations WHERE id = ?", (station_id,)).fetchone() + ) + favorite_before = tuple( + conn.execute("SELECT * FROM favorites WHERE station_id = ?", (station_id,)).fetchone() + ) + + db.init_db(path) + + with db.connect(path) as conn: + assert conn.execute("PRAGMA user_version").fetchone()[0] == 4 + assert ( + tuple(conn.execute("SELECT * FROM stations WHERE id = ?", (station_id,)).fetchone()) + == station_before + ) + assert ( + tuple( + conn.execute( + "SELECT * FROM favorites WHERE station_id = ?", (station_id,) + ).fetchone() + ) + == favorite_before + ) + assert conn.execute("SELECT COUNT(*) FROM station_sources").fetchone()[0] == 0 + assert conn.execute("SELECT COUNT(*) FROM station_source_memberships").fetchone()[0] == 0 + assert conn.execute("PRAGMA foreign_key_check").fetchall() == [] + + +def test_v3_upgrade_matches_fresh_v4_schema(tmp_path): + legacy = tmp_path / "legacy.db" + fresh = tmp_path / "fresh.db" + + _make_v3_database(legacy) + db.init_db(fresh) + db.init_db(legacy) + + with db.connect(legacy) as old_conn, db.connect(fresh) as fresh_conn: + assert _schema_objects(old_conn) == _schema_objects(fresh_conn) + assert old_conn.execute("PRAGMA user_version").fetchone()[0] == 4 + assert fresh_conn.execute("PRAGMA user_version").fetchone()[0] == 4 + + +def test_failed_v3_upgrade_rolls_back_ddl_and_version(tmp_path, monkeypatch): + path = tmp_path / "failure.db" + _make_v3_database(path) + + original = db._create_station_provenance_schema + + def fail_after_ddl(conn): + original(conn) + raise RuntimeError("injected provenance schema failure") + + monkeypatch.setattr(db, "_create_station_provenance_schema", fail_after_ddl) + + with pytest.raises(RuntimeError, match="injected provenance schema failure"): + db.init_db(path) + + with db.connect(path) as conn: + assert conn.execute("PRAGMA user_version").fetchone()[0] == 3 + tables = db.table_names(conn) + assert "station_sources" not in tables + assert "station_source_memberships" not in tables + assert "remote_import_receipts" not in tables + assert conn.execute("PRAGMA foreign_key_check").fetchall() == [] + + +def test_concurrent_v3_initializers_create_one_consistent_schema(tmp_path): + path = tmp_path / "concurrent.db" + _make_v3_database(path) + + with ThreadPoolExecutor(max_workers=4) as pool: + results = list(pool.map(lambda _: db.init_db(path), range(4))) + + assert results == [None] * 4 + + with db.connect(path) as conn: + assert conn.execute("PRAGMA user_version").fetchone()[0] == 4 + assert "station_sources" in db.table_names(conn) + assert "station_source_memberships" in db.table_names(conn) + assert conn.execute("PRAGMA foreign_key_check").fetchall() == [] diff --git a/tests/test_station_provenance_remote_integration.py b/tests/test_station_provenance_remote_integration.py new file mode 100644 index 0000000..2834d9c --- /dev/null +++ b/tests/test_station_provenance_remote_integration.py @@ -0,0 +1,98 @@ +# SPDX-License-Identifier: MIT +"""Remote provenance dispatch and profile-owned source catalog.""" + +import pytest + +from fluxtuner import tui_m3u_import as importer +from fluxtuner.core import db +from fluxtuner.core import station_provenance as provenance +from fluxtuner.core.m3u_import import ( + import_m3u_with_new_source, + parse_station_collection, +) +from fluxtuner.core.profiles import get_or_create_profile + + +@pytest.fixture +def library(tmp_path, monkeypatch): + monkeypatch.setattr(db, "DB_FILE", tmp_path / "remote-provenance.db") + db.init_db() + with db.connect() as conn: + first = db.ensure_default_profile(conn) + second = get_or_create_profile(conn, "Second") + conn.commit() + return first, second + + +def _preview(): + return parse_station_collection(b"#EXTM3U\n#EXTINF:-1,Radio\nhttps://radio.example/live\n") + + +def test_tui_remote_dispatch_records_remote_source(library): + first, second = library + outcome = importer._save_import( + first, + _preview(), + "Remote collection", + False, + lambda: True, + remote_source=True, + ) + assert outcome.result.added_to_playlist == 1 + + with db.connect() as conn: + sources = provenance.list_sources(conn, profile_id=first) + assert len(sources) == 1 + assert sources[0]["kind"] == "remote_collection" + assert sources[0]["display_name"] is None + assert ( + len(provenance.list_source_stations(conn, profile_id=first, source_id=sources[0]["id"])) + == 1 + ) + assert provenance.list_sources(conn, profile_id=second) == [] + + +def test_catalog_is_profile_scoped_and_hides_refs(library): + first, second = library + with db.connect() as conn: + first_id = provenance.create_source(conn, profile_id=first, kind="local_collection") + second_id = provenance.create_source(conn, profile_id=second, kind="remote_collection") + conn.commit() + + first_sources = provenance.list_sources(conn, profile_id=first) + second_sources = provenance.list_sources(conn, profile_id=second) + assert [row["id"] for row in first_sources] == [first_id] + assert [row["id"] for row in second_sources] == [second_id] + assert "source_ref" not in first_sources[0] + assert "station_key" not in first_sources[0] + + +def test_conflicting_source_context_rejected(library): + first, _ = library + with pytest.raises(ValueError, match="Conflicting"): + importer._save_import( + first, + _preview(), + "Conflict", + False, + lambda: True, + local_source=True, + remote_source=True, + ) + with db.connect() as conn: + assert provenance.list_sources(conn, profile_id=first) == [] + + +def test_remote_source_creation_obeys_caller_rollback(library): + first, _ = library + with db.connect() as conn: + outcome = import_m3u_with_new_source( + conn, + profile_id=first, + playlist_name="Rollback", + entries=_preview().entries, + kind="remote_collection", + ) + assert outcome.result.added_to_playlist == 1 + conn.rollback() + assert provenance.list_sources(conn, profile_id=first) == [] diff --git a/tests/test_station_provenance_security.py b/tests/test_station_provenance_security.py new file mode 100644 index 0000000..e323c59 --- /dev/null +++ b/tests/test_station_provenance_security.py @@ -0,0 +1,172 @@ +# SPDX-License-Identifier: MIT +"""Security and ownership regressions for station provenance.""" + +from __future__ import annotations + +import pytest + +from fluxtuner.core import db +from fluxtuner.core import station_provenance as provenance +from fluxtuner.core.custom_stations import add_custom_station +from fluxtuner.core.profiles import get_or_create_profile + + +@pytest.fixture +def library(tmp_path): + path = tmp_path / "provenance-security.db" + db.init_db(path) + with db.connect(path) as conn: + first = db.ensure_default_profile(conn) + second = get_or_create_profile(conn, "Second") + add_custom_station( + conn, + profile_id=first, + name="Radio", + url="https://example.test/live", + ) + station_id = int(conn.execute("SELECT id FROM stations").fetchone()["id"]) + conn.commit() + yield conn, first, second, station_id + + +@pytest.mark.parametrize("station_id", [None, 0, -1, True, 999999]) +def test_invalid_station_cannot_create_observation(library, station_id): + conn, first, _, _ = library + source_id = provenance.create_source(conn, profile_id=first, kind="local_collection") + with pytest.raises(ValueError): + provenance.record_observation( + conn, + profile_id=first, + source_id=source_id, + station_id=station_id, + ) + assert conn.execute("SELECT COUNT(*) FROM station_source_memberships").fetchone()[0] == 0 + + +def test_shared_station_can_have_independent_profile_sources(library): + conn, first, second, station_id = library + local = provenance.create_source(conn, profile_id=first, kind="local_collection") + remote = provenance.create_source(conn, profile_id=first, kind="remote_collection") + other = provenance.create_source(conn, profile_id=second, kind="local_collection") + for owner, source in ((first, local), (first, remote), (second, other)): + provenance.record_observation( + conn, profile_id=owner, source_id=source, station_id=station_id + ) + + first_sources = provenance.list_station_sources(conn, profile_id=first, station_id=station_id) + second_sources = provenance.list_station_sources(conn, profile_id=second, station_id=station_id) + assert {item["id"] for item in first_sources} == {local, remote} + assert {item["id"] for item in second_sources} == {other} + assert conn.execute("SELECT COUNT(*) FROM stations").fetchone()[0] == 1 + + +def test_profile_deletion_cascades_only_owned_provenance(library): + conn, first, second, station_id = library + first_source = provenance.create_source(conn, profile_id=first, kind="local_collection") + second_source = provenance.create_source(conn, profile_id=second, kind="remote_collection") + provenance.record_observation( + conn, profile_id=first, source_id=first_source, station_id=station_id + ) + provenance.record_observation( + conn, profile_id=second, source_id=second_source, station_id=station_id + ) + + conn.execute("DELETE FROM profiles WHERE id = ?", (first,)) + remaining = conn.execute("SELECT id, profile_id FROM station_sources").fetchall() + assert [(row["id"], row["profile_id"]) for row in remaining] == [(second_source, second)] + assert ( + provenance.list_station_sources(conn, profile_id=second, station_id=station_id)[0]["id"] + == second_source + ) + assert conn.execute("SELECT COUNT(*) FROM stations").fetchone()[0] == 1 + assert conn.execute("PRAGMA foreign_key_check").fetchall() == [] + + +def test_foreign_source_id_is_not_usable_as_capability(library): + conn, first, second, station_id = library + source_id = provenance.create_source(conn, profile_id=first, kind="local_collection") + provenance.record_observation( + conn, profile_id=first, source_id=source_id, station_id=station_id + ) + + for action in ( + lambda: provenance.get_source(conn, profile_id=second, source_id=source_id), + lambda: provenance.list_source_stations(conn, profile_id=second, source_id=source_id), + lambda: provenance.record_observation( + conn, profile_id=second, source_id=source_id, station_id=station_id + ), + lambda: provenance.delete_source(conn, profile_id=second, source_id=source_id), + ): + with pytest.raises(ValueError, match="Source not found"): + action() + + assert conn.execute("SELECT COUNT(*) FROM station_source_memberships").fetchone()[0] == 1 + + +def test_observation_timestamp_update_can_be_rolled_back(library): + conn, first, _, station_id = library + source_id = provenance.create_source(conn, profile_id=first, kind="local_collection") + provenance.record_observation( + conn, profile_id=first, source_id=source_id, station_id=station_id + ) + conn.commit() + + conn.execute( + """ + UPDATE station_source_memberships + SET last_observed_at = ? + WHERE source_id = ? AND station_id = ? + """, + ("2000-01-01T00:00:00+00:00", source_id, station_id), + ) + conn.commit() + + conn.execute("SAVEPOINT observe_rollback") + provenance.record_observation( + conn, profile_id=first, source_id=source_id, station_id=station_id + ) + assert ( + conn.execute( + """ + SELECT last_observed_at FROM station_source_memberships + WHERE source_id = ? AND station_id = ? + """, + (source_id, station_id), + ).fetchone()[0] + != "2000-01-01T00:00:00+00:00" + ) + + conn.execute("ROLLBACK TO SAVEPOINT observe_rollback") + conn.execute("RELEASE SAVEPOINT observe_rollback") + assert ( + conn.execute( + """ + SELECT last_observed_at FROM station_source_memberships + WHERE source_id = ? AND station_id = ? + """, + (source_id, station_id), + ).fetchone()[0] + == "2000-01-01T00:00:00+00:00" + ) + + +def test_public_source_data_does_not_expose_opaque_ref(library): + conn, first, _, station_id = library + source_id = provenance.create_source( + conn, profile_id=first, kind="remote_collection", display_name="Jazz" + ) + provenance.record_observation( + conn, profile_id=first, source_id=source_id, station_id=station_id + ) + + private_ref = conn.execute( + "SELECT source_ref FROM station_sources WHERE id = ?", (source_id,) + ).fetchone()[0] + + source = provenance.get_source(conn, profile_id=first, source_id=source_id) + observed = provenance.list_station_sources(conn, profile_id=first, station_id=station_id) + + assert "source_ref" not in source + assert all("source_ref" not in item for item in observed) + assert private_ref not in str(source) + assert private_ref not in str(observed) diff --git a/tests/test_tui_m3u_import.py b/tests/test_tui_m3u_import.py index 20e2a1e..862f9a0 100644 --- a/tests/test_tui_m3u_import.py +++ b/tests/test_tui_m3u_import.py @@ -298,3 +298,90 @@ def test_application_opens_destination_only_after_confirmation(monkeypatch, conf harness.show_persistent_playlist_stations.assert_awaited_once_with("Imported") else: harness.show_persistent_playlist_stations.assert_not_awaited() + + +def test_tui_local_import_records_provenance(library): + async def scenario(): + app, screen = form_app(library) + async with app.run_test(size=(100, 44)) as pilot: + await preview(app, screen, pilot, library) + assert screen._local_source is True + with db.connect() as conn: + assert conn.execute("SELECT COUNT(*) FROM station_sources").fetchone()[0] == 0 + + screen.query_one("#m3u-name", Input).value = "Local provenance" + await save(app, screen, pilot) + assert len(app.results) == 1 + assert app.results[0].result.added_to_playlist == 2 + + with db.connect() as conn: + sources = conn.execute( + "SELECT profile_id, kind, display_name FROM station_sources" + ).fetchall() + assert len(sources) == 1 + assert sources[0]["profile_id"] == library["first"] + assert sources[0]["kind"] == "local_collection" + assert sources[0]["display_name"] is None + assert ( + conn.execute("SELECT COUNT(*) FROM station_source_memberships").fetchone()[0] == 2 + ) + + asyncio.run(scenario()) + + +def test_tui_path_edit_clears_provenance_context(library): + async def scenario(): + app, screen = form_app(library) + async with app.run_test(size=(100, 44)) as pilot: + await preview(app, screen, pilot, library) + assert screen._local_source is True + screen.query_one("#m3u-path", Input).value += ".changed" + await pilot.pause() + assert screen._preview is None + assert screen._local_source is False + assert screen.query_one("#m3u-save", Button).disabled + assert not app.results + + with db.connect() as conn: + assert conn.execute("SELECT COUNT(*) FROM station_sources").fetchone()[0] == 0 + + asyncio.run(scenario()) + + +def test_tui_cancelled_preview_creates_no_provenance(library): + async def scenario(): + app, screen = form_app(library) + async with app.run_test(size=(100, 44)) as pilot: + await preview(app, screen, pilot, library) + assert screen._local_source is True + await pilot.press("escape") + await pilot.pause() + assert app.results == [None] + + with db.connect() as conn: + assert conn.execute("SELECT COUNT(*) FROM station_sources").fetchone()[0] == 0 + assert ( + conn.execute("SELECT COUNT(*) FROM station_source_memberships").fetchone()[0] == 0 + ) + + asyncio.run(scenario()) + + +def test_tui_remote_save_does_not_claim_local_provenance(library): + from fluxtuner.core.m3u_import import parse_station_collection + + captured = parse_station_collection( + b"#EXTM3U\n#EXTINF:-1,Remote\nhttps://remote.example/live\n" + ) + outcome = importer._save_import( + library["first"], + captured, + "Remote provenance guard", + False, + lambda: True, + ) + assert outcome.result.added_to_playlist == 1 + + with db.connect() as conn: + assert conn.execute("SELECT COUNT(*) FROM station_sources").fetchone()[0] == 0 + assert conn.execute("SELECT COUNT(*) FROM station_source_memberships").fetchone()[0] == 0 diff --git a/tests/test_web_reauth_storage.py b/tests/test_web_reauth_storage.py index 97cf138..8824920 100644 --- a/tests/test_web_reauth_storage.py +++ b/tests/test_web_reauth_storage.py @@ -14,7 +14,7 @@ def test_fresh_database_has_reauthentication_storage(tmp_path) -> None: version = conn.execute("PRAGMA user_version").fetchone()[0] columns = {row["name"] for row in conn.execute("PRAGMA table_info(web_reauth_attempts)")} - assert version == 3 + assert version == db.CURRENT_SCHEMA_VERSION assert { "id", "user_id", @@ -54,7 +54,7 @@ def test_v2_upgrade_preserves_existing_users(tmp_path) -> None: for row in conn.execute("SELECT name FROM sqlite_master WHERE type = 'index'") } - assert version == 3 + assert version == db.CURRENT_SCHEMA_VERSION assert user is not None assert user["password_hash"] == "test-hash-not-a-credential" assert "web_reauth_attempts" in tables diff --git a/tests/test_web_remote_provenance.py b/tests/test_web_remote_provenance.py new file mode 100644 index 0000000..0ff25bb --- /dev/null +++ b/tests/test_web_remote_provenance.py @@ -0,0 +1,164 @@ +# SPDX-License-Identifier: MIT +"""Persisted Web remote import receipts: integrity, isolation and one-time use.""" + +import pytest + +from fluxtuner.core import db +from fluxtuner.core.profiles import get_or_create_profile +from fluxtuner.core.station_provenance import list_sources +from fluxtuner.web import m3u_import +from fluxtuner.web import remote_import_receipts as receipts + +DATA = b"#EXTM3U\n#EXTINF:-1,Radio\nhttps://radio.example/live\n" + + +@pytest.fixture +def library(web_library): + _client, users = web_library + alice_id, first = users["alice"] + bob_id, _bob_profile = users["bob"] + with db.connect() as conn: + second = get_or_create_profile(conn, "Second", user_id=alice_id) + conn.commit() + return first, second, alice_id, bob_id + + +def _import(token, *, user, profile=None, session="session-one", data=DATA): + return m3u_import.import_payload( + data, + user_id=user, + profile_name=profile, + playlist_name="Imported", + existing=False, + remote_receipt=token, + session_token=session, + ) + + +def test_valid_receipt_creates_remote_source_once(library): + first, _, alice, _ = library + token = receipts.issue_for_default_profile( + user_id=alice, session_token="session-one", data=DATA + ) + result = _import(token, user=alice) + assert result["added_to_playlist"] == 1 + with db.connect() as conn: + sources = list_sources(conn, profile_id=first) + assert len(sources) == 1 + assert sources[0]["kind"] == "remote_collection" + assert sources[0]["display_name"] is None + stored = conn.execute( + "SELECT token_hash, session_hash, payload_hash FROM remote_import_receipts" + ).fetchone() + assert token not in str(tuple(stored)) + assert "session-one" not in str(tuple(stored)) + with pytest.raises(ValueError, match="receipt"): + m3u_import.import_payload( + DATA, + user_id=alice, + profile_name=None, + playlist_name="Replay attempt", + existing=False, + remote_receipt=token, + session_token="session-one", + ) + + +@pytest.mark.parametrize("change", ["payload", "session", "profile", "user", "forged"]) +def test_attestation_cannot_be_reassigned(library, change): + first, _, alice, bob = library + token = receipts.issue_for_default_profile( + user_id=alice, session_token="session-one", data=DATA + ) + args = {"user": alice} + if change == "payload": + args["data"] = DATA.replace(b"Radio", b"Altered") + elif change == "session": + args["session"] = "other-session" + elif change == "profile": + args["profile"] = "Second" + elif change == "user": + args["user"] = bob + else: + token = "fabricated-receipt" + with pytest.raises(ValueError, match="receipt"): + _import(token, **args) + with db.connect() as conn: + assert list_sources(conn, profile_id=first) == [] + assert conn.execute("SELECT count(*) FROM stations").fetchone()[0] == 0 + + +def test_failure_after_claim_restores_unconsumed_receipt(library, monkeypatch): + _, _, alice, _ = library + token = receipts.issue_for_default_profile( + user_id=alice, session_token="session-one", data=DATA + ) + + def fail(*args, **kwargs): + raise RuntimeError("injected persistence failure") + + monkeypatch.setattr(m3u_import, "import_m3u_with_new_source", fail) + with pytest.raises(RuntimeError, match="injected"): + _import(token, user=alice) + with db.connect() as conn: + assert conn.execute("SELECT consumed_at FROM remote_import_receipts").fetchone()[0] is None + + +def test_explicit_profile_receipt_imports_only_into_selected_profile(library): + first, second, alice, _ = library + token = receipts.issue_for_profile( + user_id=alice, profile_name="Second", session_token="session-one", data=DATA + ) + result = m3u_import.import_payload( + DATA, + user_id=alice, + profile_name="Second", + playlist_name="Second import", + existing=False, + remote_receipt=token, + session_token="session-one", + ) + assert result["added_to_playlist"] == 1 + with db.connect() as conn: + assert list_sources(conn, profile_id=first) == [] + assert [s["kind"] for s in list_sources(conn, profile_id=second)] == ["remote_collection"] + + +def test_explicit_profile_receipt_rejects_default_profile(library): + first, second, alice, _ = library + token = receipts.issue_for_profile( + user_id=alice, profile_name="Second", session_token="session-one", data=DATA + ) + with pytest.raises(ValueError, match="receipt"): + _import(token, user=alice) + with db.connect() as conn: + assert list_sources(conn, profile_id=first) == [] + assert list_sources(conn, profile_id=second) == [] + + +def test_explicit_profile_receipt_rejects_other_user(library): + _first, second, alice, bob = library + token = receipts.issue_for_profile( + user_id=alice, profile_name="Second", session_token="session-one", data=DATA + ) + with pytest.raises((ValueError, LookupError)): + _import(token, user=bob, profile="Second") + with db.connect() as conn: + assert list_sources(conn, profile_id=second) == [] + + +def test_receipt_expiration_rejected(library, monkeypatch): + from datetime import timedelta + + first, _, alice, _ = library + token = receipts.issue_for_profile( + user_id=alice, profile_name=None, session_token="session-one", data=DATA + ) + initial_now = receipts.utc_now() + monkeypatch.setattr( + receipts, "utc_now", lambda: initial_now + receipts.RECEIPT_LIFETIME + timedelta(seconds=5) + ) + with pytest.raises(ValueError, match="receipt"): + _import(token, user=alice) + with db.connect() as conn: + assert list_sources(conn, profile_id=first) == []