Getting Started with FastAPI-Topaz¶

from fastapi import Request
from fastapi_topaz import (
    AuditLogger,
    AuthorizerOptions,
    CircuitBreaker,
    DecisionCache,
    Identity,
    IdentityType,
    TopazConfig,
)


def identity_provider(request: Request) -> Identity:
    """
    Extract user identity from request.

    In a real app, this would decode a JWT, validate a session,
    or use your authentication system.
    """
    user_id = request.headers.get("X-User-ID")

    if not user_id:
        return Identity(type=IdentityType.IDENTITY_TYPE_NONE)

    return Identity(
        type=IdentityType.IDENTITY_TYPE_SUB,
        value=user_id,
    )


def resource_context_provider(request: Request) -> dict:
    """
    Provide additional context for policy evaluation.

    This context is available in Rego as `input.resource.*`
    """
    return {
        **request.path_params,
        "method": request.method,
    }


# Create the configuration once at startup
topaz_config = TopazConfig(
    authorizer_options=AuthorizerOptions(
        url="localhost:8282",
    ),
    policy_path_root="myapp",
    identity_provider=identity_provider,
    policy_instance_name="myapp",
    resource_context_provider=resource_context_provider,
    # Cache decisions for 30s to reduce Topaz calls
    decision_cache=DecisionCache(ttl_seconds=30, max_size=500),
    # Graceful degradation when Topaz is unavailable
    circuit_breaker=CircuitBreaker(
        failure_threshold=5,
        recovery_timeout=30,
        fallback="cache_then_deny",
    ),
    # Structured JSON audit logging for all decisions
    audit_logger=AuditLogger(),
)

from fastapi import Depends, FastAPI, HTTPException, Request
from fastapi_topaz import (
    TopazMiddleware,
    filter_authorized_resources,
    get_authorized_resource,
    require_policy_auto,
    require_rebac_allowed,
    require_rebac_hierarchy,
    skip_middleware,
)

from auth import topaz_config

app = FastAPI(title="Document API")

# --- Global Middleware ---
# Auto-protects ALL routes; excluded paths skip authorization
app.add_middleware(
    TopazMiddleware,
    config=topaz_config,
    exclude_paths=[r"^/health$", r"^/docs.*", r"^/openapi.json$"],
)

# Sample data
DOCUMENTS = {
    1: {"id": 1, "title": "Public Doc", "owner": "alice"},
    2: {"id": 2, "title": "Private Doc", "owner": "bob"},
    3: {"id": 3, "title": "Shared Doc", "owner": "alice"},
}


# --- 1. Public endpoint (excluded from middleware) ---
@app.get("/health")
@skip_middleware
async def health():
    """No authorization required."""
    return {"status": "ok"}


# --- 2. Policy-based with auto-generated path ---
@app.get("/documents")
async def list_documents(
    request: Request,
    _: None = Depends(require_policy_auto(topaz_config)),
):
    """
    Auto-generates policy path: myapp.GET.documents
    Checks: "Can this user list documents?"
    """
    return {"documents": list(DOCUMENTS.values())}


# --- 3. ReBAC: relationship-based access ---
@app.get("/documents/{id}")
async def get_document(
    id: int,
    request: Request,
    _: None = Depends(
        require_rebac_allowed(topaz_config, "document", "can_read")
    ),
):
    """
    Checks: "Can this user read THIS specific document?"
    Raises 403 if no can_read relation exists.
    """
    doc = DOCUMENTS.get(id)
    if not doc:
        raise HTTPException(status_code=404, detail="Not found")
    return doc


# --- 4. Non-raising permission check for UI ---
@app.get("/documents/{id}/details")
async def document_details(id: int, request: Request):
    """
    Uses check_relations() to return permissions without raising.
    Useful for showing/hiding UI buttons.
    """
    doc = DOCUMENTS.get(id)
    if not doc:
        raise HTTPException(status_code=404, detail="Not found")

    permissions = await topaz_config.check_relations(
        request,
        object_type="document",
        object_id=str(id),
        relations=["can_read", "can_write", "can_delete"],
    )
    return {"document": doc, "permissions": permissions}


# --- 5. Fetch-then-authorize pattern ---
def fetch_document(request: Request):
    """Fetch resource; returns None if not found."""
    doc_id = int(request.path_params.get("id", 0))
    return DOCUMENTS.get(doc_id)


@app.put("/documents/{id}")
async def update_document(
    document=Depends(
        get_authorized_resource(
            topaz_config, fetch_document, "document", "can_write"
        )
    ),
):
    """
    Fetches the document first, then checks can_write.
    Returns 404 if not found, 403 if not authorized.
    """
    return {"updated": document["title"]}


# --- 6. Batch filtering ---
@app.get("/my-documents")
async def my_documents(
    request: Request,
    filter_fn=Depends(
        filter_authorized_resources(topaz_config, "document", "can_read")
    ),
):
    """
    Checks can_read on each document concurrently.
    Returns only authorized documents.
    """
    authorized = await filter_fn(list(DOCUMENTS.values()))
    return {"documents": authorized}


# --- 7. Hierarchical ReBAC ---
@app.get("/orgs/{org_id}/projects/{proj_id}/docs/{doc_id}")
async def get_nested_doc(
    org_id: str,
    proj_id: str,
    doc_id: str,
    request: Request,
    _=Depends(
        require_rebac_hierarchy(
            topaz_config,
            [
                ("organization", "org_id", "member"),
                ("project", "proj_id", "viewer"),
                ("document", "doc_id", "can_read"),
            ],
        )
    ),
):
    """
    Checks organization membership, project access, AND document read.
    All three must pass (mode="all" is the default).
    """
    return {"org": org_id, "project": proj_id, "doc": doc_id}


# --- Lifecycle ---
@app.on_event("shutdown")
async def shutdown():
    """Release cache, circuit breaker, and pool resources."""
    await topaz_config.close()