Visibility phase 3: redaction-safe public read API + leak test

Adds the anonymous read surface (/api/v1/public) — the privacy-critical core.

- CurrentUserOrNone dependency: optional auth that never 401s (anonymous OK).
- public_view_service: every projection passes through privacy.person_visibility.
  persons redacted (living → "Living person", hidden dropped); relationships
  only when both endpoints non-hidden; events only for FULL-visibility persons
  (partnership events only when both partners full); names only for FULL
  persons. Not-viewable trees raise 404 (not 403) so the surface can't probe
  for private trees. Media deferred (higher-sensitivity; own pass later).
- public router: read-only directory + tree + persons/relationships/events +
  person detail/names/events. Directory lists `public` to all and adds
  `site_members` for authenticated callers; never lists unlisted/private.
- PublicTreeRead omits owner_id.

Tests (ran locally — CI does not run pytest): an anonymous end-to-end leak test
asserting a living person's real name, alias, and birth year appear in NO public
response while the deceased person's data does; plus private=404, unlisted
viewable-by-link-but-unlisted, site_members requires login, and directory
visibility. Full suite: 70 passed. Regenerated openapi.json + TS client.

Note: the AUTHED list endpoints still leak per-person for non-members
(pre-existing) — fixed next, separately.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: Justin Paul <justin@jpaul.me>
This commit is contained in:
2026-06-09 09:17:41 -04:00
parent 3ff03b037b
commit 9820a77d25
8 changed files with 1478 additions and 0 deletions
+12
View File
@@ -40,6 +40,18 @@ async def get_current_user(request: Request, session: SessionDep) -> User:
CurrentUser = Annotated[User, Depends(get_current_user)]
async def get_current_user_or_none(request: Request, session: SessionDep) -> User | None:
"""Optional auth for public read endpoints — never raises. Returns the user
when a valid session is present, else None (anonymous viewer)."""
raw_token = extract_session_token(request)
if raw_token is None:
return None
return await auth_service.resolve_session_user(session, raw_token=raw_token)
CurrentUserOrNone = Annotated[User | None, Depends(get_current_user_or_none)]
def get_mailer() -> Mailer:
settings = get_settings()
if settings.mailer == "smtp" and settings.smtp_host:
+2
View File
@@ -11,6 +11,7 @@ from app.api.v1 import (
media,
names,
persons,
public,
relationships,
sources,
trees,
@@ -30,3 +31,4 @@ api_router.include_router(citations.router)
api_router.include_router(media.router)
api_router.include_router(gedcom.router)
api_router.include_router(cleanup.router)
api_router.include_router(public.router)
+135
View File
@@ -0,0 +1,135 @@
"""Public, read-only viewing surface.
Optional auth (anonymous allowed). Every response is built by
``public_view_service``, which routes through the privacy engine and redacts
possibly-living people. No create/update/delete here.
"""
import uuid
from fastapi import APIRouter
from app.api.deps import CurrentUserOrNone, SessionDep
from app.schemas.event import EventRead
from app.schemas.name import NameRead
from app.schemas.person import PersonRead
from app.schemas.relationship import RelationshipRead
from app.schemas.tree import PublicTreeRead
from app.services import public_view_service
router = APIRouter(prefix="/public", tags=["public"])
def _vid(viewer: CurrentUserOrNone) -> uuid.UUID | None:
return viewer.id if viewer else None
@router.get("/trees", response_model=list[PublicTreeRead])
async def public_directory(
session: SessionDep,
viewer: CurrentUserOrNone,
q: str | None = None,
limit: int = 50,
offset: int = 0,
) -> list[PublicTreeRead]:
trees = await public_view_service.list_public_trees(
session, viewer_id=_vid(viewer), q=q, limit=limit, offset=offset
)
return [PublicTreeRead.model_validate(t) for t in trees]
@router.get("/trees/{tree_id}", response_model=PublicTreeRead)
async def public_tree(
tree_id: uuid.UUID, session: SessionDep, viewer: CurrentUserOrNone
) -> PublicTreeRead:
tree = await public_view_service.get_public_tree(
session, viewer_id=_vid(viewer), tree_id=tree_id
)
return PublicTreeRead.model_validate(tree)
@router.get("/trees/{tree_id}/persons", response_model=list[PersonRead])
async def public_persons(
tree_id: uuid.UUID, session: SessionDep, viewer: CurrentUserOrNone
) -> list[PersonRead]:
tree = await public_view_service.get_public_tree(
session, viewer_id=_vid(viewer), tree_id=tree_id
)
persons = await public_view_service.list_public_persons(
session, viewer_id=_vid(viewer), tree=tree
)
return [PersonRead.model_validate(p) for p in persons]
@router.get("/trees/{tree_id}/relationships", response_model=list[RelationshipRead])
async def public_relationships(
tree_id: uuid.UUID, session: SessionDep, viewer: CurrentUserOrNone
) -> list[RelationshipRead]:
tree = await public_view_service.get_public_tree(
session, viewer_id=_vid(viewer), tree_id=tree_id
)
rels = await public_view_service.list_public_relationships(
session, viewer_id=_vid(viewer), tree=tree
)
return [RelationshipRead.model_validate(r) for r in rels]
@router.get("/trees/{tree_id}/events", response_model=list[EventRead])
async def public_events(
tree_id: uuid.UUID, session: SessionDep, viewer: CurrentUserOrNone
) -> list[EventRead]:
tree = await public_view_service.get_public_tree(
session, viewer_id=_vid(viewer), tree_id=tree_id
)
events = await public_view_service.list_public_events(
session, viewer_id=_vid(viewer), tree=tree
)
return [EventRead.model_validate(e) for e in events]
@router.get("/trees/{tree_id}/persons/{person_id}", response_model=PersonRead)
async def public_person(
tree_id: uuid.UUID,
person_id: uuid.UUID,
session: SessionDep,
viewer: CurrentUserOrNone,
) -> PersonRead:
tree = await public_view_service.get_public_tree(
session, viewer_id=_vid(viewer), tree_id=tree_id
)
person = await public_view_service.get_public_person(
session, viewer_id=_vid(viewer), tree=tree, person_id=person_id
)
return PersonRead.model_validate(person)
@router.get("/trees/{tree_id}/persons/{person_id}/names", response_model=list[NameRead])
async def public_person_names(
tree_id: uuid.UUID,
person_id: uuid.UUID,
session: SessionDep,
viewer: CurrentUserOrNone,
) -> list[NameRead]:
tree = await public_view_service.get_public_tree(
session, viewer_id=_vid(viewer), tree_id=tree_id
)
names = await public_view_service.list_public_person_names(
session, viewer_id=_vid(viewer), tree=tree, person_id=person_id
)
return [NameRead.model_validate(n) for n in names]
@router.get("/trees/{tree_id}/persons/{person_id}/events", response_model=list[EventRead])
async def public_person_events(
tree_id: uuid.UUID,
person_id: uuid.UUID,
session: SessionDep,
viewer: CurrentUserOrNone,
) -> list[EventRead]:
tree = await public_view_service.get_public_tree(
session, viewer_id=_vid(viewer), tree_id=tree_id
)
events = await public_view_service.list_public_person_events(
session, viewer_id=_vid(viewer), tree=tree, person_id=person_id
)
return [EventRead.model_validate(e) for e in events]