Getting started¶

import asyncio

from artifactr.workspace import ACTIONS as ARTIFACTR
from reflexr.workspace import ACTIONS as REFLEXR

from grantr import Grants, InMemoryStorage, Resource, User

ROOT, ALICE, BOB = User(id="root"), User(id="alice"), User(id="bob")
MAIN = Resource(kind="workspace", workspace="main")
PLAN = Resource(kind="artifact", workspace="main", id="plan")


async def main() -> None:
    grants = Grants(InMemoryStorage(), catalogs=[ARTIFACTR, REFLEXR])

    # The deployment's first operator, who may create tenants.
    await grants.first_operator(ROOT.id)
    # A tenant, owned by Alice, who may then do anything in it.
    await grants.create_tenant("acme", owner=ALICE.id, by=ROOT)

    # Bob joins acme, and may edit what is in its main workspace.
    await grants.add_member("acme", BOB, by=ALICE)
    await grants.grant("acme", "editor", on=MAIN, to=BOB, by=ALICE)

    assert await grants.check("acme", BOB, "artifactr:edit_artifact", PLAN)
    assert not await grants.check("acme", BOB, "grantr:grant", MAIN), "an admin grants roles"
    assert not await grants.check("globex", BOB, "artifactr:read_artifacts", PLAN)

    # Every change is in the tenant's audit log.
    log = await grants.audit("acme", by=ALICE)
    assert [entry.event.type for entry in log][-2:] == ["member_added", "granted"]


asyncio.run(main())

from artifactr import core as artifactr
from artifactr.workspace import InMemoryStorage, Workspaces

from grantr import Agent, Subject, User


def subject_of(tenant: str, workspace: str | None, actor: artifactr.Actor) -> Subject:
    match actor:
        case artifactr.UserActor(id=user):
            return User(id=user)
        # An agent's turn is checked as the person it acts for (ADR-0004).
        case artifactr.AgentActor(on_behalf_of=artifactr.UserActor(id=user)):
            return User(id=user)
        case _:  # an actor this application doesn't know holds nothing
            return Agent()


hooks = grants.hooks(subject_of)
workspaces = Workspaces(InMemoryStorage(), authorize=hooks.authorize, hidden=hooks.hidden)

from sqlalchemy.ext.asyncio import create_async_engine

from grantr import Grants
from grantr.sql import SqlStorage, migrate

engine = create_async_engine("postgresql+asyncpg://app:secret@localhost/app")
await migrate(engine)
grants = Grants(SqlStorage(engine), catalogs=[ARTIFACTR, REFLEXR])

from fastapi import FastAPI
from starlette.requests import HTTPConnection

from grantr.fastapi import Caller, Unauthorized, grantr_router


async def resolve_actor(connection: HTTPConnection) -> Caller:
    user = await verify_token(connection)  # the application's own sign-in
    if user is None:
        raise Unauthorized("no caller")
    return Caller(User(id=user))


app = FastAPI()
app.include_router(grantr_router(grants, resolve_actor=resolve_actor), prefix="/grantr/v1")