feat: implement authorization with RBAC and relationship-based access

[rorybyrne]

Implement authorization for OSA using a hybrid model: Role-Based Access Control (RBAC) for coarse-grained permissions and Relationship-Based Access Control (ReBAC) for resource-level access.

Background

OSA needs authorization to protect:

• Archive configuration - Only admins should manage schemas, traits, conventions, vocabularies
• Depositions - Owners can edit their own drafts, curators can review assigned submissions
• Published records - Public read access

Design Decision: RBAC + ReBAC (not ABAC)

Model	Use Case	Decision
RBAC	Role-based permissions (admin, curator, depositor)	✅ Use for coarse-grained access
ReBAC	Relationship-based (ownership, assignment)	✅ Use for resource-level access
ABAC	Attribute-based (status, timestamps, etc.)	❌ Handle in domain logic instead

Why not ABAC for lifecycle rules? Rules like "can only edit if status == draft" are domain invariants, not access policy. They apply regardless of who's calling (even system processes). Keeping them in domain logic centralises business rules and keeps authorization focused on identity.

Separation of Concerns

Authorization answers: "Is this user allowed to attempt this action on this resource?"
Domain logic answers:  "Is this action valid given the current state?"

Authorization Model

Roles (RBAC)

SuperAdmin (node operator)
    └── Admin (manages registries)
            └── Curator (reviews depositions)
                    └── Depositor (submits data)
                            └── Public (reads published records)

Relationships (ReBAC)

• Ownership: User owns depositions they created
• Assignment: Curator is assigned to review specific depositions (future)

Permission Matrix

Resource	Public	Depositor	Curator	Admin	SuperAdmin
Published Records	read	read	read	read	read
Own Depositions	-	CRUD	read	read	read
Others' Depositions	-	-	read, approve/reject	read	read
Schemas	read	read	read	CRUD	CRUD
Traits	read	read	read	CRUD	CRUD
Conventions	read	read	read	CRUD	CRUD
Vocabularies	read	read	read	CRUD	CRUD
User Management	-	-	-	-	CRUD
Node Configuration	-	-	-	-	CRUD

Key Abstractions

Two complementary patterns enforce authorization at different layers, both designed so that forgetting to authorise is harder than remembering.

1. Declarative __auth__ gate on handlers (role check, no I/O)

Handlers declare a composable Policy via __auth__. The CommandHandler metaclass wraps run() to evaluate the policy before the handler body executes. A startup validator rejects any handler for a non-__public__ command that lacks __auth__ — missing policies are boot failures, not production vulnerabilities.

# Composable policy constructors
requires_role(Role.DEPOSITOR)
requires_role(Role.CURATOR) | requires_role(Role.ADMIN)
requires_any_role(Role.CURATOR, Role.ADMIN)

# Handler declaration
class SubmitDepositionHandler(CommandHandler[SubmitDeposition, SubmitResult]):
    __auth__ = requires_role(Role.DEPOSITOR)
    service: DepositionService

    async def run(self, cmd: SubmitDeposition) -> SubmitResult:
        # Only reached if principal has the Depositor role
        return await self.service.submit(self._principal, cmd.deposition_id)

Policies compose with & (all must pass), | (any must pass), and ~ (negate):

class Policy:
    async def evaluate(self, ctx: AuthContext) -> Verdict: ...
    def __and__(self, other: Policy) -> AllOf: ...
    def __or__(self, other: Policy) -> AnyOf: ...
    def __invert__(self) -> Not: ...

Public commands opt out explicitly — silence means protected:

class SearchRecords(Command):
    __public__: ClassVar = True

What this gives us: Cheap, I/O-free role check that runs before any DB access. Auditable by grepping __auth__. Forgetting it is a startup crash.

2. AuthorizedRepo + Guarded[T] (ownership/relationship check, at resource load)

For relationship-based checks (ownership, assignment), the resource must be loaded first. Rather than relying on manual authz.guard() calls in services that are easy to forget, the repository itself returns resources wrapped in Guarded[T] — a type that cannot be used without an explicit authorization check.

class Guarded(Generic[T]):
    """A loaded resource that requires authorization before use."""

    def check(self, action: str) -> T:
        """Authorise and unwrap. Raises AuthorizationError if denied."""
        self._authz.guard(self._principal, action, self._resource)
        return self._resource

    # No __getattr__ — you MUST .check() to get the inner value

AuthorizedRepo wraps a raw repository to return Guarded resources:

class AuthorizedRepo(Generic[T, ID]):
    _inner: Repository[T, ID]
    _principal: Principal
    _authz: AuthorizationService

    async def get(self, id: ID) -> Guarded[T]:
        resource = await self._inner.get(id)
        return Guarded(resource, self._principal, self._authz)

DI wiring: Principal is request-scoped (resolved from JWT + role lookup, same pattern as CurrentUser today). Dishka injects it into AuthorizedRepo alongside the real repo and AuthorizationService. Services declare which repo type they want:

# User-facing write service — gets authorization-enforcing repo
class DepositionService(Service):
    _repo: AuthorizedRepo[Deposition, DepositionSRN]

    async def submit(self, dep_id: DepositionSRN) -> SubmitResult:
        dep = (await self._repo.get(dep_id)).check("deposition:submit")
        #      forgot .check()? ^^^ pyright error: Guarded[Deposition] has no .submit()
        dep.submit()
        ...

# Internal event handler — uses raw repo, no auth wrapping
class ConvertDepositionToRecord(EventHandler[DepositionApproved]):
    _dep_repo: DepositionRepository  # raw port, no Guarded wrapper

What this gives us: Relationship checks happen at the point the resource is loaded (no extra query). Forgetting .check() is a type error, not a security hole. Domain ports stay pure — wrapping is done at the DI/application layer.

How the layers compose

Request
  → DI resolves Principal (JWT + role lookup, request-scoped)
  → Handler.__auth__ gate               (role check, no I/O)
  → Handler.run()
    → Service calls repo.get()          (returns Guarded[T])
    → Service calls .check(action)      (ownership check, no extra query)
    → Domain logic on unwrapped T       (state invariants)

Layer	Concern	Enforced by	Cost
Startup validation	Every non-public handler has __auth__	Metaclass / boot check	Zero (one-time)
Gate	Role check	Handler metaclass wraps run()	Cheap (in-memory)
Guard	Ownership / relationship	Guarded[T] + type system	Free (resource already loaded)
Domain	State invariants	Aggregate methods	Already exists

Principal

Enriched version of CurrentUser — resolved via DI from JWT + DB role lookup:

@dataclass(frozen=True)
class Principal:
    user_id: UserId
    identity: ProviderIdentity
    roles: frozenset[Role]

    def has_role(self, role: Role) -> bool: ...
    def has_any_role(self, *roles: Role) -> bool: ...

Implementation Scope

In Scope

• Role enum, assignment model, and persistence
• Principal resolution via DI
• Policy type with composable constructors (requires_role, etc.)
• Handler metaclass enforcement (__auth__ / __public__ startup validation)
• Guarded[T] and AuthorizedRepo wrapper
• AuthorizationService (evaluates ownership, role hierarchy)
• owner_id / created_by on relevant aggregates
• Audit logging of authorization decisions

Out of Scope (Separate Issues)

• Curation assignment workflow
• Delegation
• Organizational/institutional boundaries
• Time-limited elevated access

Acceptance Criteria

• Unauthenticated users can only read published records
• Depositors can only modify their own depositions
• Curators can view and act on depositions pending review
• Only admins can create/modify schemas, traits, conventions
• Authorization failures return 403 with audit log entry
• App refuses to boot if a non-public handler lacks __auth__
• Forgetting .check() on a Guarded[T] is a type error

Dependencies

• Depends on feat: implement authentication with ORCiD as identity provider #28 (authentication) for user identity — ✅ complete

References

• Zanzibar: Google's Consistent, Global Authorization System