Files
provenance/backend/app/services/privacy.py
T
justin dffd05d303 Add layered service/API for tenancy and people with the privacy seam
Wires the data model through repository -> service -> API/v1. The privacy engine (app/services/privacy.py) is the single enforcement point: every read resolves visibility there (tree role, tree visibility, per-person override; living-person redaction is a marked Phase 2 TODO). All writes record an attributable AuditEntry.

Endpoints: POST /users (open dev bootstrap until auth), GET /users/me, POST/GET /trees, GET /trees/{id}, and POST/GET /trees/{id}/persons. Authn is a temporary X-User-Id header shim; authz is membership-based (owner/editor/viewer). Domain errors map to 401/403/404/409. Verified on the deploy target: private tree -> 403 for non-members, missing actor -> 401, audit log populated.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: Justin Paul <justin@jpaul.me>
2026-06-06 10:40:19 -04:00

63 lines
2.2 KiB
Python

"""The privacy engine — the single enforcement point for visibility.
INVARIANT (CLAUDE.md #2): every read path resolves visibility here. Do not add a
query path that returns rows to a caller without first passing through this
module. Effective visibility is a function of the viewer's role on the tree, the
tree's visibility, the per-person override, and (Phase 2) living-person status.
"""
import enum
import uuid
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.enums import MembershipRole, PersonPrivacy, TreeVisibility
from app.models.person import Person
from app.models.tree import Tree, TreeMembership
class Visibility(enum.StrEnum):
full = "full"
redacted = "redacted"
hidden = "hidden"
async def get_membership_role(
session: AsyncSession, user_id: uuid.UUID | None, tree_id: uuid.UUID
) -> MembershipRole | None:
if user_id is None:
return None
stmt = select(TreeMembership.role).where(
TreeMembership.tree_id == tree_id,
TreeMembership.user_id == user_id,
)
return (await session.execute(stmt)).scalar_one_or_none()
async def can_view_tree(session: AsyncSession, *, user_id: uuid.UUID | None, tree: Tree) -> bool:
if tree.deleted_at is not None:
return False
if await get_membership_role(session, user_id, tree.id) is not None:
return True
return tree.visibility in (TreeVisibility.public, TreeVisibility.unlisted)
async def can_edit_tree(session: AsyncSession, *, user_id: uuid.UUID | None, tree: Tree) -> bool:
role = await get_membership_role(session, user_id, tree.id)
return role in (MembershipRole.owner, MembershipRole.editor)
async def person_visibility(
session: AsyncSession, *, user_id: uuid.UUID | None, tree: Tree, person: Person
) -> Visibility:
if not await can_view_tree(session, user_id=user_id, tree=tree):
return Visibility.hidden
if await get_membership_role(session, user_id, tree.id) is not None:
return Visibility.full
# Non-member viewing a public/unlisted tree:
if person.privacy == PersonPrivacy.private:
return Visibility.hidden
# TODO(Phase 2): redact living people for non-members (ARCHITECTURE §6).
return Visibility.full