# FastAPI Migration Templates

FastAPI services do not use `verify_grass_access`, `flaskify`, or `oto.response`. The legacy
auth mechanism is `check_vendor_access_for_profiles` (or `assert_access` which wraps it). The
PP check is `is_authorized_for_tenant`. Migration does not require a new `auth.py` module —
the logic layer already has `assert_authorization` as the target state.

PP migration spans **Phase 1 (Baseline)**, **Phase 2 (Shadow)** and **Phase 3 (Enforce)** — 

- **Phase 1 - Baseline integration tests**: see `SKILL.md`
- **Phase 2 — Shadow** *(ship first; additive, zero behavior change)*: leave the existing legacy
  auth (`assert_access` / `check_vendor_access_for_profiles`) **untouched** and bolt on a
  side-effect-only `is_authorized()` call via `MigrationAuthorizationBackend` from
  `python-pdp-sdk[migration]==6.6.0`. It always returns allow and emits the
  `pp_auth.rollout.would_deny` Datadog metric whenever the real PP decision *would* have denied.
- **Phase 3 — Enforce** *(follow-up PR, after metrics are clean)*: swap
  `MigrationAuthorizationBackend` for the real `PdpAuthorizationBackend`, then apply Template
  FA / FB / FC to make the PP decision enforcing.

> **Why additive shadow rather than a config-driven backend swap?** An always-allow
> `is_authorized()` that the handler short-circuits on would bypass the existing legacy check during
> the shadow phase — an enforcement gap, not a shadow. Keeping legacy auth untouched and adding a
> side-effect-only PP call avoids that. See
> `_tasks/migration-authorization-backend-shadow-approach.md`.

---

# Phase 2 — Shadow

FastAPI has no global request proxy, so the `extra_tags_getter` reads the live request from a
`ContextVar` set by middleware. Keep the middleware, the accessors, and `request_tags()` together.

## `context.py` — request accessors + `request_tags()`

```python
# PP TODO: Move this to python-owscontext
from contextvars import ContextVar

from owscontext import get_request_context
from starlette.requests import Request
from starlette.types import ASGIApp, Receive, Scope, Send

_current_request: ContextVar[Request | None] = ContextVar("current_request", default=None)


class CurrentRequestMiddleware:
    """Store the active Starlette Request in a ContextVar for request_tags()."""

    def __init__(self, app: ASGIApp) -> None:
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return
        token = _current_request.set(Request(scope, receive))
        try:
            await self.app(scope, receive, send)
        finally:
            _current_request.reset(token)


def get_route_template() -> str | None:
    """Matched route template (e.g. /tracks/{track_id}); falls back to the raw path
    for unmatched routes, keeping metric tags low-cardinality.

    Read lazily: Starlette only populates scope["route"] during routing, after every
    middleware's setup runs — so this resolves at auth-check time, inside the endpoint.
    """
    request = _current_request.get()
    if request is None:
        return None
    route = request.scope.get("route")
    return route.path if route is not None else request.url.path


def request_tags() -> list[str]:
    """Datadog tags describing the active request, for use as extra_tags_getter."""
    request = _current_request.get()
    if request is None:
        return []
    context = get_request_context()
    has_auth_header = bool(context and context.authorization)
    return [
        f"method:{request.method}",
        f"endpoint:{get_route_template()}",
        f"has_authorization_header:{str(has_auth_header).lower()}",
    ]
```

## `main.py` — middleware stack

```python
from owscontext.context.asgi.middleware import CorrelationIdMiddleware, RequestContextMiddleware

from <service>.context import CurrentRequestMiddleware

app = FastAPI(...)
app.add_middleware(CurrentRequestMiddleware)
app.add_middleware(RequestContextMiddleware)
app.add_middleware(CorrelationIdMiddleware)
```

## Startup wiring *(once at startup)*

```python
import config
from python_pdp_sdk import (
    MigrationAuthorizationBackend,
    OwsPdpClient,
    PdpAuthorizationBackend,
)

from <service>.context import request_tags

ows_pdp_client = OwsPdpClient(ows_client=ows_client)
pdp_backend = PdpAuthorizationBackend(ows_pdp_client)

# Always-allow wrapper that emits pp_auth.rollout.would_deny on a would-deny.
# Fetches DD_API_KEY from `<env>/datadog/DD_API_KEY`; pass dd_api_key=... to override.
authorization_backend = MigrationAuthorizationBackend(
    inner_backend=pdp_backend,
    service_name=config.SERVICE_NAME,
    environment=config.environment,
    extra_tags_getter=request_tags,
)
```

## Handler shadow call *(same for all postures)*

Leave the existing legacy `assert_access` / `check_vendor_access_for_profiles` **exactly as it is**
and add the side-effect-only PP call. The return value is ignored — the backend always allows, so
traffic is unaffected; the call exists only to emit `pp_auth.rollout.would_deny`.

```python
from python_pdp_sdk import ForwardKwargsGetter

# existing legacy auth stays untouched, e.g.:
await check_vendor_access_for_profiles(
    identity_id=identity_uuid, vendor_uuid=request.vendor_uuid,
    profiles=profiles, allowed_types=header.BULK_SESSION_ALLOWED_PROFILE_TYPES,
)

# shadow-only PP measurement (always returns True; return value ignored):
tenant = await get_tenant(request.vendor_uuid, request.subaccount_id)
if tenant:
    authorization_backend.is_authorized(
        action="view", resource_id=0, resource_type="account",
        resource_getter=ForwardKwargsGetter(),
        tenant={"tenant_type": tenant.tenant_type, "tenant_uuid": str(tenant.tenant_uuid)},
    )
```

---

# Phase 3 — Enforce

Ship this only after the `pp_auth.rollout.would_deny` metric is clean for the endpoint (see
`references/caller-analysis-and-rollout.md`). Swap the startup wiring to the real backend, then
apply the template below and remove the shadow call:

```python
# replace the MigrationAuthorizationBackend wrapper with the raw backend
authorization_backend = PdpAuthorizationBackend(ows_pdp_client)
```

## Template FA — swap `assert_access` → `assert_authorization` *(simplest migration)*

The handler calls `assert_access` in the logic layer. Change it to `assert_authorization`:

```python
# Before (legacy only — no PP check):
await bulk_session_ingestion_logic.assert_access(
    identity_uuid=identity_uuid,
    profiles=profiles,
    bulk_session_ingestion_id=bulk_session_ingestion_id,
)

# After (PP first, falls back to check_vendor_access_for_profiles):
await bulk_session_ingestion_logic.assert_authorization(
    identity_uuid=identity_uuid,
    profiles=profiles,
    bulk_session_ingestion_id=bulk_session_ingestion_id,
)
```

No logic-layer changes needed if `assert_authorization` already exists for the resource type.
If it doesn't exist, create it following the pattern in `bulk_session_logic.assert_authorization`.

Decorate both methods with `@tracer.wrap()` so each call gets its own APM span:

```python
from ddtrace import tracer

@tracer.wrap()
async def assert_access(self, ...): ...

@tracer.wrap()
async def assert_authorization(self, ...): ...
```

---

## Template FB — inline PP check in handler *(when handler calls `check_vendor_access_for_profiles` directly)*

```python
# Before (legacy only):
await check_vendor_access_for_profiles(
    identity_id=identity_uuid,
    vendor_uuid=request.vendor_uuid,
    profiles=profiles,
    allowed_types=header.BULK_SESSION_ALLOWED_PROFILE_TYPES,
)

# After (PP first, falls back to legacy):
tenant = await get_tenant(request.vendor_uuid, request.subaccount_id)
is_pp_authorized = False
if tenant:
    is_pp_authorized = is_authorized_for_tenant(
        tenant_uuid=tenant.tenant_uuid,
        tenant_type=tenant.tenant_type,
        tenant_attributes=tenant.tenant_attributes,
    )
if not is_pp_authorized:
    await check_vendor_access_for_profiles(
        identity_id=identity_uuid,
        vendor_uuid=request.vendor_uuid,
        profiles=profiles,
        allowed_types=header.BULK_SESSION_ALLOWED_PROFILE_TYPES,
    )
```

---

## Template FC — add PP check to a no-resource-level-auth endpoint

For endpoints with no ownership check today (JWT required but no tenant check):

```python
# Add at the top of the handler:
tenant = await get_tenant(resource_vendor_uuid, resource_subaccount_id)
if not tenant or not is_authorized_for_tenant(
    tenant_uuid=tenant.tenant_uuid,
    tenant_type=tenant.tenant_type,
    tenant_attributes=tenant.tenant_attributes,
):
    raise HTTPException(status_code=403, detail="Unauthorized")
```

---

## Integration test personas

FastAPI services use two tiers of test personas defined in `tests/integration/conftest.py`:

**Tier 1 — Profile-based (legacy, `check_vendor_access_for_profiles`):**
- `client` — label user (e.g. Kitty Wizard); has legacy profiles
- `subaccount_client` — subaccount user (e.g. Mo Thugs Records); has legacy profiles
- `no_access_client` — user with no PP roles and no legacy profiles

**Tier 2 — PP role-based (`is_authorized_for_tenant`):**
- `<PP_role_name>_client` — label user with the specific cerbos derived role assigned
- `<PP_role_name>_subaccount_client` — subaccount user with the specific cerbos derived role assigned

The Tier 1 fixtures (`client`, `subaccount_client`, `no_access_client`) should already exist.
The Tier 2 fixtures must be added to `conftest.py` when migrating an endpoint to PP — one pair
per PP role being introduced. Each fixture follows the same `generate_bearer_token` +
`SecretLookupInfo` pattern as the existing fixtures, pointing to a new AWS Secrets Manager
secret for the PP-enabled test user.

### `pytest.mark.parametrize` pattern

```python
@pytest.mark.parametrize(
    "client_fixture_name, vendor_uuid_param, expected_status_code",
    [
        pytest.param("client", VENDOR_UUID_KITTY_WIZARD, 200,
                     id="label user with legacy profile can access their vendor"),
        pytest.param("client", VENDOR_UUID_NOT_KITTY_WIZARD, 403,
                     id="label user cannot access a different vendor"),
        pytest.param("<PP_role_name>_client", VENDOR_UUID_KITTY_WIZARD, 200,
                     id="PP role user can access their vendor"),
        pytest.param("<PP_role_name>_client", VENDOR_UUID_NOT_KITTY_WIZARD, 200,
                     id="PP role user can access any vendor (role grants cross-tenant access)"),
        pytest.param("subaccount_client", SUBACCOUNT_ID_MO_THUGS, 200,
                     id="subaccount user can access their subaccount"),
        pytest.param("no_access_client", VENDOR_UUID_KITTY_WIZARD, 403,
                     id="user with no roles or profiles is rejected"),
    ],
)
def test_<endpoint>(client_fixture_name, vendor_uuid_param, expected_status_code, request):
    client = request.getfixturevalue(client_fixture_name)
    ...
```