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>
This commit is contained in:
@@ -0,0 +1,62 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user