patristic-be / tests /test_reader_access.py
Mario33333's picture
deploy: reader-access security hardening (feature/audiobook@06a5ed8)
7c2d250 verified
Raw
History Blame Contribute Delete
27.4 kB
"""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.)