"""HTTP governance API for External API clients (global_admin). GET /api/v1/api-clients list POST /api/v1/api-clients register GET /api/v1/api-clients/scopes scope catalog + expiry presets GET /api/v1/api-clients/{id} detail (+ tokens / usage) PATCH /api/v1/api-clients/{id} update / kill switch DELETE /api/v1/api-clients/{id} soft delete (+ revoke all) POST /api/v1/api-clients/{id}/restore undo soft delete (frees the slug) POST /api/v1/api-clients/{id}/tokens mint (plaintext once) POST /api/v1/api-clients/{id}/tokens/{tid}/revoke Distinct from the identity-consumer governance surface — this manages API access to Helix business resources, not directory data distribution. """ from __future__ import annotations from datetime import UTC, datetime, timedelta from uuid import UUID from fastapi import APIRouter, Depends, Query, Request from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from app.api_clients.models import ( CLIENT_STATUS_ACTIVE, ApiClient, ApiClientToken, ApiClientUsageDaily, ) from app.api_clients.person_visibility import person_visibility_audit_detail from app.api_clients.schemas import ( ApiClientCreateRequest, ApiClientDetailResponse, ApiClientResponse, ApiClientUpdateRequest, TokenMintRequest, TokenMintResponse, TokenResponse, UsageDayResponse, ) from app.api_clients.scopes import catalog_payload from app.api_clients.tokens import ( ALLOWED_EXPIRY_DAYS, ApiClientTokenService, token_is_active, ) from app.api_clients.visibility import ( cloud_visibility_audit_detail, cloud_visibility_is_malformed, ) from app.audit.service import AuditService from app.auth.models import AuthUser from app.core.database import get_db from app.core.dependencies import ( require_admin, require_admin_read, require_browser_session, ) from app.core.exceptions import ConflictError, NotFoundError, ValidationError from app.core.response import success_response from app.identity.person_models import ( IdentityOrgMembership, IdentityOrgNode, IdentityPerson, ) router = APIRouter(prefix="/api/v1/api-clients", tags=["api-clients"]) async def _audit( db: AsyncSession, request: Request, admin: AuthUser, *, action: str, resource_id: str, detail: dict, ) -> None: await AuditService(db).log( trace_id=getattr(request.state, "trace_id", "") or "", actor_type="user", actor_user_id=admin.id, action=action, status="success", resource_type="api_client", resource_id=resource_id, detail=detail, ip_address=request.client.host if request.client else None, user_agent=request.headers.get("user-agent"), ) async def _require_owner(db: AsyncSession, owner_person_id: UUID) -> IdentityPerson: owner = await db.get(IdentityPerson, owner_person_id) if owner is None or owner.deleted_at is not None: raise ValidationError("owner_person_id does not reference a known person.") if owner.employment_status != "active": raise ValidationError( "API client owner must be an active employee — pick the person " "accountable for this integration." ) return owner async def _validated_visibility( db: AsyncSession, normalized: dict | None, ) -> dict: """Existence-check a normalized whitelist and return its label map. Lazy import: the shape lives in ``app.api_clients`` (platform), but resolving it against ``integrations`` / ``cmdb_cloud_scopes`` is cmdb's data. Same platform→domain lazy-import pattern the curated OpenAPI spec builder uses. """ from app.domains.cmdb.open_visibility import validate_cloud_visibility_targets try: return await validate_cloud_visibility_targets(db, normalized) except ValueError as exc: raise ValidationError(str(exc)) from exc async def _validate_person_visibility_targets( db: AsyncSession, normalized: dict | None, ) -> None: """Existence-check the identifiers in a person ceiling. Same reasoning as ``_validated_visibility`` for cloud: a ceiling that names a department or a person that does not exist is not an error at read time — it is a filter that silently matches nobody, which the admin then debugs for a day. ``departments`` holds HRIS org-node **source ids** (not Helix UUIDs), which is exactly the mix-up worth catching here. """ if not isinstance(normalized, dict): return source_ids = [str(d) for d in (normalized.get("departments") or [])] if source_ids: # Checked against BOTH tables, because the runtime filter matches # ``IdentityOrgMembership.org_node_source_id`` and # ``expand_department_subtree`` keeps seed ids whether or not an # org-node row exists. Validating against the node table alone # would reject a ceiling that demonstrably works — e.g. one naming # a node that an org restructure has since soft-deleted while # memberships still carry it — and would do so on ANY later PATCH, # including one that narrows the grant. known = set((await db.execute( select(IdentityOrgNode.org_node_source_id).where( IdentityOrgNode.org_node_source_id.in_(source_ids), IdentityOrgNode.deleted_at.is_(None), ) )).scalars().all()) still_unknown = [s for s in source_ids if s not in known] if still_unknown: known |= set((await db.execute( select(IdentityOrgMembership.org_node_source_id).where( IdentityOrgMembership.org_node_source_id.in_(still_unknown), IdentityOrgMembership.deleted_at.is_(None), # Mirrors the runtime resolver, which reads primary # memberships only: accepting an id that appears # solely on a secondary membership would bless a # ceiling that still matches nobody. IdentityOrgMembership.is_primary.is_(True), ).distinct() )).scalars().all()) missing = sorted(set(source_ids) - known) if missing: raise ValidationError( "person_visibility.departments: unknown org node id(s) " f"{', '.join(missing)}. Use the HRIS org-node source id " "(what the department picker stores), not a Helix row UUID.", ) person_ids = [str(p) for p in (normalized.get("person_ids") or [])] if person_ids: # Partitioned first: the id column is a UUID TypeDecorator whose # bind step raises on a non-UUID string, which SQLAlchemy wraps # into a StatementError and the global handler turns into a 500. # A malformed id is a caller mistake — it belongs in the same # actionable 422 as an id that simply doesn't exist. well_formed: list[str] = [] missing_people = [] for raw in person_ids: try: well_formed.append(str(UUID(raw))) except (ValueError, AttributeError, TypeError): missing_people.append(raw) known_people = set() if well_formed: known_people = set(str(p) for p in (await db.execute( select(IdentityPerson.id).where( IdentityPerson.id.in_(well_formed), IdentityPerson.deleted_at.is_(None), ) )).scalars().all()) missing_people = sorted( [*missing_people, *(set(well_formed) - known_people)], ) if missing_people: raise ValidationError( "person_visibility.person_ids: unknown person id(s) " f"{', '.join(missing_people)}.", ) async def _get_client(db: AsyncSession, client_id: UUID) -> ApiClient: # Explicit selectinload + populate_existing so the sync ``row.tokens`` # access in the response builders can't trip MissingGreenlet on a # ``db.get`` identity-map hit. row = (await db.execute( select(ApiClient) .where(ApiClient.id == client_id) .options(selectinload(ApiClient.tokens)) .execution_options(populate_existing=True) )).scalar_one_or_none() if row is None or row.deleted_at is not None: raise NotFoundError("ApiClient", str(client_id)) return row def _token_response(t: ApiClientToken) -> TokenResponse: return TokenResponse( id=t.id, name=t.name, token_prefix=t.token_prefix, expires_at=t.expires_at, revoked_at=t.revoked_at, last_used_at=t.last_used_at, last_used_ip=t.last_used_ip, created_at=t.created_at, is_active=token_is_active(t), ) def _client_response(c: ApiClient, owner: IdentityPerson | None) -> ApiClientResponse: live = [t for t in c.tokens if token_is_active(t)] last_used = [t.last_used_at for t in c.tokens if t.last_used_at] return ApiClientResponse( id=c.id, name=c.name, slug=c.slug, description=c.description, owner_person_id=c.owner_person_id, owner_display_name=owner.display_name if owner else None, owner_work_email=owner.work_email if owner else None, owner_employee_no=owner.employee_no if owner else None, owner_employment_status=owner.employment_status if owner else None, status=c.status, scopes=list(c.scopes or []), rate_limit_per_minute=c.rate_limit_per_minute, created_at=c.created_at, updated_at=c.updated_at, deleted_at=c.deleted_at, active_token_count=len(live), last_used_at=max(last_used) if last_used else None, cloud_visibility=( c.cloud_visibility if isinstance(c.cloud_visibility, dict) else None ), # Same reasoning as ``person_visibility_misconfigured`` below: an # unreadable whitelist resolves to NO subscriptions on the data # plane, so rendering the row as `cloud_visibility: null` alone # would tell the admin "unrestricted" about a client that can # currently read nothing. cloud_visibility_misconfigured=cloud_visibility_is_malformed( c.cloud_visibility, ), person_visibility=( c.person_visibility if isinstance(c.person_visibility, dict) else None ), # A stored value that isn't a JSON object cannot be rendered as a # ceiling, and the data plane reads it as "nobody". Saying only # `person_visibility: null` here would tell the admin "this client # is unrestricted" about a client that can currently read no one — # the pre-fix lie, on the surface the admin actually looks at. person_visibility_misconfigured=( c.person_visibility is not None and not isinstance(c.person_visibility, dict) ), ) @router.get("/scopes", summary="Scope catalog + token expiry presets") async def list_known_scopes( request: Request, db: AsyncSession = Depends(get_db), _admin: AuthUser = Depends(require_admin_read()), ): """Everything the register/edit drawer needs to render a grant: the scope catalog, the token-expiry presets, and the cloud providers a visibility whitelist can name (derived from the actual cloud integrations, so the picker can never offer a provider that resolves to zero subscriptions).""" from app.integrations.models import Integration providers = sorted({ p for p in (await db.execute( select(Integration.provider).where(Integration.kind == "cloud"), )).scalars().all() if p }) return success_response( data={ "scopes": catalog_payload(), "token_expiry_days": sorted(ALLOWED_EXPIRY_DAYS), "cloud_providers": providers, }, request=request, ) @router.get("", summary="List API clients") async def list_clients( request: Request, include_deleted: bool = Query(default=False), db: AsyncSession = Depends(get_db), _admin: AuthUser = Depends(require_admin_read()), ): stmt = select(ApiClient).order_by(ApiClient.created_at.desc()) if not include_deleted: stmt = stmt.where(ApiClient.deleted_at.is_(None)) rows = list((await db.execute(stmt)).scalars().all()) owner_ids = {c.owner_person_id for c in rows} owners: dict[UUID, IdentityPerson] = {} if owner_ids: result = await db.execute( select(IdentityPerson).where(IdentityPerson.id.in_(owner_ids)) ) owners = {p.id: p for p in result.scalars().all()} data = [ _client_response(c, owners.get(c.owner_person_id)).model_dump(mode="json") for c in rows ] return success_response(data=data, request=request) @router.post("", summary="Register an API client") async def create_client( body: ApiClientCreateRequest, request: Request, db: AsyncSession = Depends(get_db), admin: AuthUser = Depends(require_admin()), ): existing = (await db.execute( select(ApiClient.id, ApiClient.deleted_at).where(ApiClient.slug == body.slug) )).one_or_none() if existing is not None: # Includes soft-deleted rows on purpose: the slug lands in audit # rows / logs, so handing it to a DIFFERENT row would make old trails # ambiguous. The way out is to restore the original row — same id, # same slug, continuous trail — not to recycle the identifier. # # Say WHICH of the two it is, and name the escape hatch: before the # restore endpoint existed this 409 was a genuine dead end (the list # page hides deleted clients, so the holder was invisible and the # slug was burned forever). if existing[1] is not None: raise ConflictError( f"API client slug '{body.slug}' belongs to a DELETED client " f"(id {existing[0]}). Slugs are never handed to a different " "client because they appear in audit trails and request logs " "— restore that client instead " f"(POST /api/v1/api-clients/{existing[0]}/restore, or the " "console's \"show deleted\" filter → Restore), or register " "the new client under a different slug." ) raise ConflictError(f"API client slug '{body.slug}' already exists.") owner = await _require_owner(db, body.owner_person_id) await _validated_visibility(db, body.cloud_visibility) await _validate_person_visibility_targets(db, body.person_visibility) row = ApiClient( name=body.name.strip(), slug=body.slug, description=(body.description or "").strip() or None, owner_person_id=body.owner_person_id, scopes=body.scopes, rate_limit_per_minute=body.rate_limit_per_minute, cloud_visibility=body.cloud_visibility, person_visibility=body.person_visibility, created_by=admin.id, ) db.add(row) await db.flush() await _audit( db, request, admin, action="api_client.create", resource_id=str(row.id), detail={ "slug": row.slug, "scopes": body.scopes, "owner": owner.work_email, **cloud_visibility_audit_detail(body.cloud_visibility), **person_visibility_audit_detail(body.person_visibility), }, ) await db.commit() row = await _get_client(db, row.id) return success_response( data=_client_response(row, owner).model_dump(mode="json"), request=request, ) @router.get("/{client_id}", summary="API client detail (tokens + usage)") async def get_client( client_id: UUID, request: Request, db: AsyncSession = Depends(get_db), _admin: AuthUser = Depends(require_admin_read()), ): row = await _get_client(db, client_id) owner = await db.get(IdentityPerson, row.owner_person_id) since = (datetime.now(UTC) - timedelta(days=30)).date() usage_rows = list((await db.execute( select(ApiClientUsageDaily) .where( ApiClientUsageDaily.client_id == client_id, ApiClientUsageDaily.usage_date >= since, ) .order_by(ApiClientUsageDaily.usage_date.asc()) )).scalars().all()) # Label the whitelisted subscriptions so the console shows account # names, not UUIDs. Read-path resolution is best-effort: a since- # deleted subscription must not 400 the whole detail view (the admin # opened this drawer precisely to fix such a grant). labels: dict = {} if isinstance(row.cloud_visibility, dict) and row.cloud_visibility: from app.domains.cmdb.open_visibility import validate_cloud_visibility_targets try: labels = await validate_cloud_visibility_targets(db, row.cloud_visibility) except ValueError: labels = {} base = _client_response(row, owner) detail = ApiClientDetailResponse( **base.model_dump(), tokens=[_token_response(t) for t in row.tokens], usage_recent_days=[UsageDayResponse.model_validate(u) for u in usage_rows], requests_30d=sum(u.request_count for u in usage_rows), cloud_visibility_labels=labels, ) return success_response(data=detail.model_dump(mode="json"), request=request) @router.patch("/{client_id}", summary="Update API client / kill switch") async def update_client( client_id: UUID, body: ApiClientUpdateRequest, request: Request, db: AsyncSession = Depends(get_db), admin: AuthUser = Depends(require_admin()), ): row = await _get_client(db, client_id) changes: dict = {} if body.name is not None and body.name.strip() != row.name: changes["name"] = {"from": row.name, "to": body.name.strip()} row.name = body.name.strip() if body.description is not None: desc = body.description.strip() or None if desc != row.description: changes["description"] = "updated" row.description = desc if body.owner_person_id is not None and body.owner_person_id != row.owner_person_id: owner = await _require_owner(db, body.owner_person_id) changes["owner"] = {"from": str(row.owner_person_id), "to": owner.work_email} row.owner_person_id = body.owner_person_id if body.scopes is not None and sorted(body.scopes) != sorted(row.scopes or []): changes["scopes"] = {"from": row.scopes, "to": body.scopes} row.scopes = body.scopes if ( "rate_limit_per_minute" in body.model_fields_set and body.rate_limit_per_minute != row.rate_limit_per_minute ): # model_fields_set distinguishes "omitted" from explicit null (the # latter resets to the platform-default limit). changes["rate_limit_per_minute"] = { "from": row.rate_limit_per_minute, "to": body.rate_limit_per_minute, } row.rate_limit_per_minute = body.rate_limit_per_minute if body.status is not None and body.status != row.status: changes["status"] = {"from": row.status, "to": body.status} row.status = body.status if "cloud_visibility" in body.model_fields_set: # Omitted = unchanged; explicit null = clear the whitelist (widen # to the whole inventory). Both directions are audited with the # before/after shape — a widening is the single most consequential # edit on this row and must never be inferable only from silence. # # Compared against the RAW stored value, not a dict-coerced one. # Coercing first made a malformed row (anything that isn't a JSON # object) read as ``None``, so `PATCH {"cloud_visibility": null}` # — the exact repair `/open/me` and the console tell the admin to # make — compared ``None != None``, skipped the assignment, and # left the unreadable value in place with the client still # resolving to no subscriptions. Same fix as person_visibility # below. current = row.cloud_visibility if body.cloud_visibility != current: await _validated_visibility(db, body.cloud_visibility) changes["cloud_visibility"] = { "from": cloud_visibility_audit_detail(current), "to": cloud_visibility_audit_detail(body.cloud_visibility), } row.cloud_visibility = body.cloud_visibility if "person_visibility" in body.model_fields_set: # Same contract as cloud_visibility above: omitted = unchanged, # explicit null = clear the ceiling (widen to the whole roster). # # Compared against the RAW stored value, not a dict-coerced one. # Coercing first made a malformed row (anything that isn't a JSON # object) read as ``None``, so `PATCH {"person_visibility": null}` # — the exact repair `/me` tells the integrator to ask for — was a # no-op that left the unreadable value in place and the client # resolving to nobody. current_pv = row.person_visibility if body.person_visibility != current_pv: await _validate_person_visibility_targets(db, body.person_visibility) changes["person_visibility"] = { "from": person_visibility_audit_detail(current_pv), "to": person_visibility_audit_detail(body.person_visibility), } row.person_visibility = body.person_visibility if changes: await db.flush() action = "api_client.update" if "status" in changes: action = ( "api_client.disable" if row.status != CLIENT_STATUS_ACTIVE else "api_client.enable" ) await _audit( db, request, admin, action=action, resource_id=str(row.id), detail={"slug": row.slug, "changes": changes}, ) await db.commit() row = await _get_client(db, client_id) owner = await db.get(IdentityPerson, row.owner_person_id) return success_response( data=_client_response(row, owner).model_dump(mode="json"), request=request, ) @router.delete("/{client_id}", summary="Soft-delete API client (revokes all tokens)") async def delete_client( client_id: UUID, request: Request, db: AsyncSession = Depends(get_db), admin: AuthUser = Depends(require_admin()), ): row = await _get_client(db, client_id) now = datetime.now(UTC) row.deleted_at = now revoked = 0 for t in row.tokens: if t.revoked_at is None: t.revoked_at = now t.revoked_by_user_id = admin.id revoked += 1 await db.flush() await _audit( db, request, admin, action="api_client.delete", resource_id=str(row.id), detail={"slug": row.slug, "tokens_revoked": revoked}, ) await db.commit() return success_response( data={"deleted": True, "tokens_revoked": revoked}, request=request, ) @router.post("/{client_id}/restore", summary="Restore a soft-deleted API client") async def restore_client( client_id: UUID, request: Request, db: AsyncSession = Depends(get_db), admin: AuthUser = Depends(require_admin()), ): """Undo a soft delete. Deletion is soft *and* the slug unique constraint counts deleted rows, so without this the pair was a deadlock: delete a client, then neither re-register its slug (409) nor see the row that holds it (the list hides deleted clients by default). Restoring the original row keeps audit trails unambiguous — same id, same slug, same history — which is exactly why recycling the slug onto a new row stays forbidden. Tokens revoked by the delete stay revoked: a restored client has no live credential until one is minted, so the restore itself grants nothing. Idempotent: restoring a live client is a no-op success. The owner is deliberately NOT re-validated — a client whose owner has since left must still be restorable, otherwise the slug is burned forever again; the console shows the departed owner so it can be re-assigned. """ row = (await db.execute( select(ApiClient) .where(ApiClient.id == client_id) .options(selectinload(ApiClient.tokens)) .execution_options(populate_existing=True) )).scalar_one_or_none() if row is None: raise NotFoundError("ApiClient", str(client_id)) if row.deleted_at is not None: row.deleted_at = None await db.flush() await _audit( db, request, admin, action="api_client.restore", resource_id=str(row.id), detail={"slug": row.slug, "status": row.status, "live_tokens": 0, "note": "tokens stay revoked"}, ) await db.commit() row = await _get_client(db, client_id) owner = await db.get(IdentityPerson, row.owner_person_id) return success_response( data=_client_response(row, owner).model_dump(mode="json"), request=request, ) @router.post("/{client_id}/tokens", summary="Mint an API token") async def mint_token( client_id: UUID, body: TokenMintRequest, request: Request, db: AsyncSession = Depends(get_db), admin: AuthUser = Depends(require_admin()), _gate: None = Depends(require_browser_session), ): # Step-up: minting returns a fresh plaintext ``helix_api_`` credential # in the response body, so — like every other secret-revealing admin # endpoint (PAT mint, API-key reveal) — it must originate from a real # Google-SSO browser session, never from a stolen admin PAT that could # otherwise mint + exfiltrate long-lived partner-API credentials. row = await _get_client(db, client_id) token_row, plaintext = await ApiClientTokenService(db).mint( client=row, name=body.name, expires_in_days=body.expires_in_days, created_by=admin.id, ) await _audit( db, request, admin, action="api_client.token_mint", resource_id=str(row.id), detail={"slug": row.slug, "token_id": str(token_row.id), "token_prefix": token_row.token_prefix, "expires_in_days": body.expires_in_days}, ) await db.commit() return success_response( data=TokenMintResponse( token=_token_response(token_row), plaintext=plaintext, ).model_dump(mode="json"), request=request, ) @router.post("/{client_id}/tokens/{token_id}/revoke", summary="Revoke an API token") async def revoke_token( client_id: UUID, token_id: UUID, request: Request, db: AsyncSession = Depends(get_db), admin: AuthUser = Depends(require_admin()), _gate: None = Depends(require_browser_session), ): # Step-up to match the mint pattern above: token lifecycle management # for external API clients is a browser-console admin action, not # something a PAT should be able to drive (revoke = partner-integration # DoS via a stolen credential). row = await _get_client(db, client_id) token_row = await ApiClientTokenService(db).revoke( client_id=client_id, token_id=token_id, actor_user_id=admin.id, ) await _audit( db, request, admin, action="api_client.token_revoke", resource_id=str(row.id), detail={"slug": row.slug, "token_id": str(token_id), "token_prefix": token_row.token_prefix}, ) await db.commit() return success_response(data={"revoked": True}, request=request) __all__ = ["router"]