"""Reader-role access-control tests (feature/reader-access). Proves the security-critical access surface introduced for the low-privilege ``reader`` role (preview / Reading-Room-only accounts): (a) a reader gets 404 on a PRIVATE book's content endpoint (no oracle), and 200 on a PUBLIC book's content; (b) GET /books returns ONLY public books to a reader; (c) admins + viewers still see every book (public and private); (d) authenticate() resolves a DB-backed app_user (and rejects disabled ones); (e) the is_public PATCH toggles visibility; plus the dependency-level gates (require_book_read_access, forbid_readers, require_query_access) and the /admin/users management endpoints. DB isolation: the project's configured inventory backend is Turso, but these tests MUST NOT touch live data. Every DB-touching test points the inventory layer at a fresh temp SQLite file via the same monkeypatch pattern as tests/test_stage1_inventory.py (a config double that exposes only ``.paths``, so ``_inventory_backend()`` falls back to 'local'). No paid API calls are made. """ from __future__ import annotations from pathlib import Path import pytest from src.api.deps import ( forbid_readers, require_book_read_access, require_query_access, ) from src.api.errors import BookNotFound, Forbidden from src.lib.auth.models import User _READER = User(username="t_reader", password="x", role="reader") _VIEWER = User(username="t_viewer", password="x", role="viewer") _ADMIN = User(username="t_admin", password="x", role="admin") # ---------- Temp local-DB harness ----------------------------------------- @pytest.fixture def temp_inventory(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): """Point the inventory layer at a fresh temp SQLite DB and seed two books. Yields the temp db path. Seeds one public + one private book so the visibility logic has both cases to discriminate. """ from src.config import load_config load_config.cache_clear() db_path = tmp_path / "inv.db" monkeypatch.setattr( "src.stage1_inventory.db.load_config", lambda: type( "C", (), { "paths": type( "P", (), {"inventory_db": db_path, "books_csv": tmp_path / "books.csv"}, )() }, )(), ) # Force a clean schema-init against THIS temp file (the once-guard is keyed # by "local:{db_path}", so a fresh path inits fresh — but clear to be safe). import src.stage1_inventory.db as dbmod dbmod._SCHEMA_INITIALIZED.clear() from src.lib.books.repo import insert_book, update_book insert_book({"book_id": "pub_book", "title_en": "Public", "status": "indexed"}) insert_book({"book_id": "priv_book", "title_en": "Private", "status": "indexed"}) update_book("pub_book", {"is_public": True}) # priv_book stays default 0 yield db_path dbmod._SCHEMA_INITIALIZED.clear() # ---------- (e) is_public PATCH / repo round-trip ------------------------- def test_is_public_defaults_private_and_patch_toggles(temp_inventory) -> None: from src.lib.books.repo import get_book, get_book_is_public, update_book # Default seed: pub_book public, priv_book private. assert get_book_is_public("pub_book") is True assert get_book_is_public("priv_book") is False assert get_book_is_public("does_not_exist") is None # → 404 signal pub = get_book("pub_book") assert pub is not None and pub.is_public is True priv = get_book("priv_book") assert priv is not None and priv.is_public is False # Toggle a private book public, then back to private. update_book("priv_book", {"is_public": True}) assert get_book_is_public("priv_book") is True update_book("priv_book", {"is_public": False}) assert get_book_is_public("priv_book") is False # ---------- (b)+(c) GET /books visibility at the repo layer --------------- def test_list_books_public_only_filters_for_readers(temp_inventory) -> None: from src.lib.books.repo import list_books # Admin/viewer path (public_only=False): sees BOTH books. all_ids = {b.book_id for b in list_books()} assert {"pub_book", "priv_book"}.issubset(all_ids) # Reader path (public_only=True): ONLY the public book. reader_ids = {b.book_id for b in list_books(public_only=True)} assert reader_ids == {"pub_book"} assert "priv_book" not in reader_ids # ---------- (a) require_book_read_access (the critical dependency) --------- def test_admin_and_viewer_bypass_visibility(temp_inventory) -> None: # Admin + viewer get full access to a PRIVATE book with no DB lookup at all # (the dep short-circuits before touching is_public). Returns the user. assert require_book_read_access(book_id="priv_book", user=_ADMIN) is _ADMIN assert require_book_read_access(book_id="priv_book", user=_VIEWER) is _VIEWER # And of course on a public book too. assert require_book_read_access(book_id="pub_book", user=_VIEWER) is _VIEWER def test_reader_allowed_on_public_book(temp_inventory) -> None: assert require_book_read_access(book_id="pub_book", user=_READER) is _READER def test_reader_404s_on_private_book_no_oracle(temp_inventory) -> None: # A reader hitting a PRIVATE book must get BOOK_NOT_FOUND — identical to a # missing book, so the private id can't be enumerated. with pytest.raises(BookNotFound) as e1: require_book_read_access(book_id="priv_book", user=_READER) with pytest.raises(BookNotFound) as e2: require_book_read_access(book_id="does_not_exist", user=_READER) # Same code + same shape for both (no distinguishing 403 for private). assert e1.value.code == "BOOK_NOT_FOUND" assert e2.value.code == "BOOK_NOT_FOUND" # ---------- forbid_readers + require_query_access gates -------------------- def test_forbid_readers_blocks_reader_allows_others() -> None: with pytest.raises(Forbidden): forbid_readers(_READER) assert forbid_readers(_VIEWER) is _VIEWER assert forbid_readers(_ADMIN) is _ADMIN def test_reader_cannot_run_queries_even_if_viewer_can( monkeypatch: pytest.MonkeyPatch, ) -> None: # Even if the operator opens queries to viewers via config, a reader must # NEVER reach a paid query path. import src.api.deps as deps monkeypatch.setattr( deps, "load_config", lambda: type("C", (), {"section": lambda self, n: {"viewer_can_query": True}})(), ) with pytest.raises(Forbidden): require_query_access(_READER) # The viewer, however, is now allowed (config opened it). assert require_query_access(_VIEWER) is _VIEWER # Admin always allowed. assert require_query_access(_ADMIN) is _ADMIN # ---------- (d) authenticate() resolves a DB-backed app_user -------------- def test_authenticate_resolves_db_app_user(temp_inventory) -> None: from src.lib.auth.repo import authenticate, user_for_username from src.lib.auth.user_store import ( create_app_user, set_app_user_disabled, ) created = create_app_user( username="DbReader", # mixed case → stored lowercased password="readerpass123", display_name="DB Reader", ) assert created.username == "dbreader" assert created.role == "reader" # forced reader, never admin # Right password → resolves a reader User; wrong password → None. u = authenticate("dbreader", "readerpass123") assert u is not None assert u.username == "dbreader" assert u.role == "reader" assert u.is_admin is False assert authenticate("dbreader", "wrong") is None # The session-resolution path (current_user) also resolves the DB user. assert user_for_username("dbreader") is not None # Disabling the account refuses login AND fails the session resolution # (an admin-disabled reader is locked out on the next request, not just # the next login). set_app_user_disabled("dbreader", True) assert authenticate("dbreader", "readerpass123") is None assert user_for_username("dbreader") is None def test_create_app_user_rejects_dupes_and_empties(temp_inventory) -> None: from src.lib.auth.user_store import create_app_user create_app_user("solo", "password1", None) with pytest.raises(ValueError): create_app_user("solo", "password2", None) # duplicate with pytest.raises(ValueError): create_app_user("", "password1", None) # empty username with pytest.raises(ValueError): create_app_user("noempty", "", None) # empty password def test_create_app_user_handles_libsql_unique_violation( temp_inventory, monkeypatch: pytest.MonkeyPatch ) -> None: """Regression: libsql raises ValueError not sqlite3.IntegrityError on UNIQUE collision. The production backend (Turso/libsql) does NOT subclass sqlite3 — a duplicate primary-key INSERT raises a bare builtins.ValueError with message "UNIQUE constraint failed: app_users.username", never sqlite3.IntegrityError. The pre-fix except clause caught only sqlite3.IntegrityError, so on Turso the raw SQL error propagated to the router's ``except ValueError`` handler, where ``"already exists" not in str(e)`` → 400 BadRequest AND leaked the SQL text. Fix: the broader except branch matches the UNIQUE constraint message and converts it to the canonical "already exists" ValueError so the router takes the 409 USER_EXISTS path with no SQL internals on the wire. Method (b): monkeypatch the store so the INSERT raises the exact libsql exception — no live Turso connection required. """ import src.lib.auth.user_store as us from src.lib.auth.user_store import create_app_user # We want to exercise the except-block race path: # (1) app_user_exists(uname) pre-check returns False → proceeds to INSERT # (2) INSERT raises libsql-style ValueError → except branch fires # (3) app_user_exists(uname) re-check returns True → raise "already exists" # # Monkeypatch app_user_exists to return False on the first call (pre-check) # and True on subsequent calls (re-check inside except). Patch connect() so # the INSERT raises the libsql UNIQUE-constraint ValueError. from contextlib import contextmanager libsql_unique_error = ValueError("UNIQUE constraint failed: app_users.username") exists_calls: list[bool] = [] def _fake_exists(uname: str) -> bool: # First call = pre-check (must return False to pass the guard). # Subsequent calls = re-check inside the except block (True = dup confirmed). first = len(exists_calls) == 0 exists_calls.append(True) return False if first else True monkeypatch.setattr(us, "app_user_exists", _fake_exists) @contextmanager def _connect_that_fails_insert(): # Raise the libsql-style UNIQUE ValueError immediately, simulating the # moment the INSERT lands after a concurrent INSERT already committed. raise libsql_unique_error yield # noqa: unreachable — required by @contextmanager monkeypatch.setattr(us, "connect", _connect_that_fails_insert) with pytest.raises(ValueError) as exc_info: create_app_user("raceuser", "password99", None) msg = str(exc_info.value) assert "already exists" in msg, ( f"Expected 'already exists' in the converted ValueError, got: {msg!r}. " "The raw UNIQUE-constraint SQL text must never reach the caller." ) assert "UNIQUE constraint failed" not in msg, ( f"Raw SQL text must be scrubbed from the converted ValueError, got: {msg!r}" ) def test_create_app_user_libsql_unique_recheck_false_scrubbed( temp_inventory, monkeypatch: pytest.MonkeyPatch ) -> None: """Hardening: concurrent-delete race — recheck returns False after UNIQUE signal. When the libsql UNIQUE constraint fires but the row has vanished by re-check time (concurrent delete between INSERT and re-check), the raw SQL error text ("UNIQUE constraint failed: app_users.username") must still NOT reach the caller. The except branch must raise a scrubbed generic conflict message instead of re-raising the original ValueError. """ from contextlib import contextmanager import src.lib.auth.user_store as us from src.lib.auth.user_store import create_app_user libsql_unique_error = ValueError("UNIQUE constraint failed: app_users.username") # app_user_exists: pre-check → False (pass the guard), re-check → False # (simulate concurrent delete — the row is gone by the time we recheck). exists_calls: list[bool] = [] def _fake_exists_always_false(uname: str) -> bool: exists_calls.append(uname) return False # always False: pre-check passes, re-check also False monkeypatch.setattr(us, "app_user_exists", _fake_exists_always_false) @contextmanager def _connect_that_fails_insert(): raise libsql_unique_error yield # noqa: unreachable — required by @contextmanager monkeypatch.setattr(us, "connect", _connect_that_fails_insert) with pytest.raises(ValueError) as exc_info: create_app_user("raceuser2", "password99", None) msg = str(exc_info.value) # The raw SQL internals must be scrubbed regardless of the recheck outcome. assert "UNIQUE constraint failed" not in msg, ( f"Raw SQL text leaked in recheck-False path: {msg!r}" ) assert "app_users" not in msg, ( f"Table name leaked in recheck-False path: {msg!r}" ) assert "username" not in msg or "already exists" in msg, ( f"Column name leaked without a clean message in recheck-False path: {msg!r}" ) # The message should be the scrubbed generic conflict notice. assert "conflict" in msg, ( f"Expected a scrubbed conflict message, got: {msg!r}" ) # ---------- /admin/users endpoints (contract for the FE) ------------------ def test_admin_users_crud_roundtrip(temp_inventory) -> None: """Drive the admin_users router functions directly (admin already gated). Exercises create (forces reader) → list (no hash) → patch disable → patch password → delete, plus the 409/404 typed errors. """ from src.api.dto.admin_users import ( CreateAppUserRequest, UpdateAppUserRequest, ) from src.api.errors import BadRequest, UserExists, UserNotFound from src.api.routers import admin_users as au # create dto = au.create_user( CreateAppUserRequest(username="PreviewBob", password="bobpass12"), _=_ADMIN, ) assert dto.username == "previewbob" assert dto.role == "reader" assert not hasattr(dto, "password_hash") # never exposes the hash # list (no password hash on the wire DTO) listed = au.list_users(_=_ADMIN) names = {u.username for u in listed.items} assert "previewbob" in names for u in listed.items: assert "passwordHash" not in u.model_dump(by_alias=True) # duplicate → 409 USER_EXISTS with pytest.raises(UserExists): au.create_user( CreateAppUserRequest(username="previewbob", password="bobpass12"), _=_ADMIN, ) # short password → rejected at the DTO layer (pydantic min_length=8) # or at the router layer (BadRequest) — either way must not succeed. import pydantic with pytest.raises((BadRequest, pydantic.ValidationError)): au.create_user( CreateAppUserRequest(username="shorty", password="x"), _=_ADMIN ) # patch: disable + reset password patched = au.update_user( UpdateAppUserRequest(disabled=True, password="newbobpass1"), username="previewbob", _=_ADMIN, ) assert patched.disabled is True # disabled reader can't authenticate even with the new password from src.lib.auth.repo import authenticate assert authenticate("previewbob", "newbobpass1") is None # patch on missing user → 404 USER_NOT_FOUND with pytest.raises(UserNotFound): au.update_user( UpdateAppUserRequest(disabled=False), username="ghost", _=_ADMIN ) # delete (idempotent) au.delete_user(username="previewbob", _=_ADMIN) au.delete_user(username="previewbob", _=_ADMIN) # no raise assert "previewbob" not in {u.username for u in au.list_users(_=_ADMIN).items} # ---------- BookDTO redaction for readers (security: detail endpoint) -------- @pytest.fixture def temp_inventory_with_sensitive_book(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): """Temp SQLite DB with one public book carrying sensitive fields. Seeds a public book with notes, source_url, local_path, and costs so the redaction tests have values to check against. Isolated from the live inventory (same monkeypatch pattern as ``temp_inventory``). """ from src.config import load_config load_config.cache_clear() db_path = tmp_path / "inv_redact.db" monkeypatch.setattr( "src.stage1_inventory.db.load_config", lambda: type( "C", (), { "paths": type( "P", (), {"inventory_db": db_path, "books_csv": tmp_path / "books.csv"}, )() }, )(), ) import src.stage1_inventory.db as dbmod dbmod._SCHEMA_INITIALIZED.clear() from src.lib.books.repo import insert_book, update_book insert_book({ "book_id": "pub_sensitive", "title_en": "Sensitive Public Book", "status": "indexed", "source_url": "https://example.com/sensitive.pdf", "local_path": "data/raw/pub_sensitive.pdf", "notes": "Admin-only private notes.", "pages_total": 42, }) update_book("pub_sensitive", {"is_public": True}) yield db_path dbmod._SCHEMA_INITIALIZED.clear() def test_reader_get_book_detail_redacts_sensitive_fields( temp_inventory_with_sensitive_book, ) -> None: """A reader hitting GET /books/{id} must NOT receive sensitive fields. The shared cache holds the full BookDTO; the redacted copy returned to the reader is never stored, so admin/viewer access is unaffected. """ from src.api.dto.books import BookCosts, BookDTO, ExtractionModeEnum from src.api.routers.books import _redact_book_for_reader # Build a representative full BookDTO (mirrors what _book_dto would return) from src.api.dto.books import ( BookStatusEnum, ReligionPair, Titles, ) full_dto = BookDTO( id="pub_sensitive", titles=Titles(en="Sensitive Public Book", display="Sensitive Public Book"), religion=ReligionPair(), status=BookStatusEnum.indexed, extraction_mode=ExtractionModeEnum.ocr, cleanup_enabled=True, is_public=True, labels=[], indexes=[], costs=BookCosts(total_usd=1.23, by_stage={"ocr": 1.23}), last_processing_per_stage={}, source_url="https://example.com/sensitive.pdf", local_path="pub_sensitive.pdf", notes="Admin-only private notes.", pages_total=42, ) redacted = _redact_book_for_reader(full_dto) # --- Sensitive fields must be blanked --- assert redacted.notes is None, "notes must be redacted for reader" assert redacted.source_url is None, "source_url must be redacted for reader" assert redacted.local_path is None, "local_path must be redacted for reader" assert redacted.costs == BookCosts(total_usd=0.0, by_stage={}), ( "costs must be zero-filled for reader" ) assert redacted.indexes == [], "indexes must be cleared for reader" assert redacted.last_processing_per_stage == {}, ( "last_processing_per_stage must be cleared for reader" ) # --- Reader-visible fields must be kept intact --- assert redacted.id == "pub_sensitive" assert redacted.titles.display == "Sensitive Public Book" assert redacted.pages_total == 42 assert redacted.is_public is True assert redacted.status == BookStatusEnum.indexed # --- Original DTO must be UNCHANGED (model_copy is immutable) --- assert full_dto.notes == "Admin-only private notes.", ( "model_copy must not mutate the cached DTO" ) assert full_dto.costs.total_usd == 1.23, ( "cached DTO costs must be unaffected by reader redaction" ) assert full_dto.source_url == "https://example.com/sensitive.pdf", ( "cached DTO source_url must be unaffected by reader redaction" ) def test_admin_and_viewer_get_book_detail_receives_full_dto( temp_inventory_with_sensitive_book, ) -> None: """Admins and viewers must receive the unredacted full BookDTO. Verifies that _redact_book_for_reader is NOT called for non-reader roles, which is enforced by the ``if user.role == "reader"`` guard in ``get_book_endpoint``. """ from src.api.dto.books import BookCosts, BookDTO, ExtractionModeEnum from src.api.routers.books import _redact_book_for_reader from src.api.dto.books import BookStatusEnum, ReligionPair, Titles full_dto = BookDTO( id="pub_sensitive", titles=Titles(en="Sensitive Public Book", display="Sensitive Public Book"), religion=ReligionPair(), status=BookStatusEnum.indexed, extraction_mode=ExtractionModeEnum.ocr, cleanup_enabled=True, is_public=True, labels=[], indexes=[], costs=BookCosts(total_usd=1.23, by_stage={"ocr": 1.23}), last_processing_per_stage={}, source_url="https://example.com/sensitive.pdf", local_path="pub_sensitive.pdf", notes="Admin-only private notes.", pages_total=42, ) # Simulate the endpoint guard: only readers are redacted. for role_user in (_ADMIN, _VIEWER): # Non-reader path: dto returned directly (no call to _redact_book_for_reader) assert role_user.role != "reader" # The full DTO is untouched — notes and costs intact. assert full_dto.notes == "Admin-only private notes." assert full_dto.costs.total_usd == 1.23 assert full_dto.source_url == "https://example.com/sensitive.pdf" # ---------- BE-m4 regression tests ---------------------------------------- def test_tampered_role_in_db_still_authenticates_as_reader( temp_inventory, ) -> None: """BE-m4(a): a row manually storing role='admin' still resolves as reader. Directly inserts a row with ``role='admin'`` into app_users (bypassing create_app_user) to simulate a tampered/hand-edited DB, then verifies that ``get_app_user_for_auth`` unconditionally returns role='reader' and is_admin=False. """ from src.lib.auth.passwords import hash_password from src.lib.auth.user_store import get_app_user_for_auth from src.stage1_inventory.db import connect pw_hash = hash_password("tamperpass1") with connect() as conn: conn.execute( """ INSERT INTO app_users (username, password_hash, role, display_name, disabled) VALUES ('tampered_admin', ?, 'admin', NULL, 0) """, (pw_hash,), ) conn.commit() user = get_app_user_for_auth("tampered_admin") assert user is not None, "tampered row should still resolve (it's enabled)" assert user.role == "reader", ( f"tampered role='admin' must be forced to 'reader', got {user.role!r}" ) assert user.is_admin is False, "is_admin must be False for a DB-backed account" def test_reader_redaction_holds_after_admin_fills_cache( temp_inventory_with_sensitive_book, ) -> None: """BE-m4(b): reader redaction is applied after the shared cache, never before. Calls get_book_endpoint as admin first (populates the cache with the full DTO), then calls as reader and asserts that sensitive fields are stripped from the reader's response while the cache entry itself remains full. The test seeds ``notes`` and ``source_url`` on the book row; costs come from the (empty) llm_calls table so they are 0.0 for all roles. What matters here is that the reader's response has notes/source_url redacted, while the cached (admin-visible) entry retains them. """ import src.api.cache as api_cache from src.api.routers.books import get_book_endpoint # Clear any leftover cache entries from prior tests. api_cache.clear() # Admin call — populates the cache with the full (unredacted) DTO. admin_dto = get_book_endpoint(book_id="pub_sensitive", user=_ADMIN) assert admin_dto.notes == "Admin-only private notes." assert admin_dto.source_url == "https://example.com/sensitive.pdf" # Reader call — cache is already warm; the function must still redact. reader_dto = get_book_endpoint(book_id="pub_sensitive", user=_READER) assert reader_dto.notes is None, "notes must be redacted for reader even from cache" assert reader_dto.source_url is None, "source_url must be redacted for reader" assert reader_dto.local_path is None, "local_path must be redacted for reader" # The cached entry is still the full (unredacted) DTO. cached = api_cache.get_or_set("books:detail:pub_sensitive", lambda: None) assert cached is not None and cached.notes == "Admin-only private notes.", ( "cache must hold the full DTO — reader redaction must not mutate it" ) assert cached.source_url == "https://example.com/sensitive.pdf", ( "cached source_url must be intact after reader access" ) def test_reader_private_and_nonexistent_book_return_identical_404( temp_inventory, ) -> None: """BE-m4(c): private book and nonexistent book return identical 404 shapes. A reader must not be able to distinguish 'private book exists' from 'book does not exist' — both must raise BookNotFound with the same code and no extra detail. """ with pytest.raises(BookNotFound) as exc_private: require_book_read_access(book_id="priv_book", user=_READER) with pytest.raises(BookNotFound) as exc_missing: require_book_read_access(book_id="no_such_book_xyz", user=_READER) assert exc_private.value.code == exc_missing.value.code == "BOOK_NOT_FOUND", ( "private and nonexistent ids must produce identical error codes for readers" ) # Neither exception must carry the book_id as an informative detail that # distinguishes a real (private) book from a missing one. # (Both raise BookNotFound — same shape by construction.)