"""External API meta surface — credential verify + machine-readable spec. GET /api/v1/open/me verify credential (any valid api token) GET /api/v1/open/openapi.json curated OpenAPI (credential-gated) `/me` is the "make your first call" smoke test so an integrator can confirm a key works before touching data endpoints. `/openapi.json` is the machine-readable spec (import into Postman / client codegen) — the global `/docs` `/openapi.json` are dev-only, and the human-readable reference is rendered in-app at ``/admin/api-clients/docs`` (the SPA, where CSP allows our own JS to run — a backend-served Redoc page can't, since the API CSP is ``default-src 'none'``). """ from __future__ import annotations import logging from fastapi import APIRouter, Depends, Query, Request from fastapi.responses import JSONResponse from sqlalchemy.ext.asyncio import AsyncSession from app.api_clients.auth import ApiClientContext, require_api_client from app.api_clients.tokens import API_TOKEN_PREFIX, ApiClientTokenService from app.core.database import get_db from app.core.exceptions import UnauthorizedError from app.core.response import success_response logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/v1/open", tags=["open-meta"]) _ME_EXAMPLE = { "success": True, "data": { "client": {"name": "BI Platform", "slug": "bi-platform"}, "scopes": ["cloud.subscriptions:read"], "rate_limit_per_minute": 300, "token_prefix": "helix_api_x…", "cloud_visibility": { "restricted": True, "providers": ["aws"], "subscription_count": 3, }, "person_visibility": {"restricted": False}, }, "meta": {"trace_id": "…"}, "errors": [], } @router.get( "/me", summary="Verify credential", responses={200: {"content": {"application/json": {"example": _ME_EXAMPLE}}}}, ) async def whoami( request: Request, ctx: ApiClientContext = Depends(require_api_client()), db: AsyncSession = Depends(get_db), ) -> dict: """Echo the authenticated API client + its granted scopes. Any valid ``helix_api_`` token (no scope required) — an integrator's first call to confirm the credential works. ``cloud_visibility`` reports whether the client is narrowed to a subset of cloud subscriptions. Without it, "the cloud endpoints return fewer accounts than I expect" is indistinguishable from a filter mistake, and only a Helix admin could tell the difference. A whitelist the data plane cannot parse is reported as ``misconfigured`` rather than ``restricted: false`` — that plane fails closed on it, so "unrestricted" would be the exact opposite of what the client is about to experience. ``person_visibility`` is the same self-diagnosis for the people surface — the population ceiling is deliberately indistinguishable from "no such person" in the data responses, so this is the ONLY place a client can learn its own reach is narrowed. """ from app.api_clients.person_visibility import ( person_visibility_audit_detail, person_visibility_is_unrestricted, ) from app.domains.cmdb.open_visibility import resolve_client_cloud_visibility visibility = await resolve_client_cloud_visibility(db, ctx) # A ceiling the data plane cannot parse makes that plane return # NOBODY (fail-closed). Reporting `restricted: false` here — which the # coerced-to-None context value would — would be a lie in the one # place built to explain an unexpectedly narrow result set: the client # would read "unrestricted", get zero rows, and have nothing to go on. if ctx.person_visibility_malformed: person_meta: dict = { "restricted": True, "misconfigured": True, "detail": ( "The stored people-scope filter is not readable, so this " "credential currently resolves to no people at all. Ask a " "Helix admin to re-save it." ), } else: person_restricted = not person_visibility_is_unrestricted( ctx.person_visibility, ) person_meta = {"restricted": person_restricted} if person_restricted: # The same shape the audit trail records: which dimensions # bound the client. detail = person_visibility_audit_detail(ctx.person_visibility) person_meta.update( {k: v for k, v in detail.items() if k != "person_visibility"}, ) return success_response( data={ "client": {"name": ctx.name, "slug": ctx.slug}, "scopes": sorted(ctx.scopes), "token_prefix": ctx.token_prefix, "cloud_visibility": visibility.as_meta(), "person_visibility": person_meta, }, request=request, ) # ── Hosted, credential-gated API reference ───────────────────────────── async def _authorize_reference( request: Request, db: AsyncSession, key: str | None, ) -> None: """Gate the reference (spec) on a valid credential: an ``helix_api_`` API-client token (header or ``?key=``) OR a Helix access JWT. The spec carries no secrets — this just keeps the API surface from being fully public, per the access-control decision.""" # 1) explicit ?key= or Authorization bearer token = key if not token: auth = request.headers.get("authorization") or "" if auth.lower().startswith("bearer "): token = auth.split(" ", 1)[1].strip() if not token: raise UnauthorizedError("Provide your API key to view the reference.") if token.startswith(API_TOKEN_PREFIX): resolved = await ApiClientTokenService(db).authenticate(token) if resolved is not None: return raise UnauthorizedError("Invalid or expired API key.") # 2) fall back to a Helix access JWT (logged-in admin viewing in-console) try: from app.core.security import decode_token payload = decode_token(token) if payload.get("type") == "access": return except Exception: # noqa: BLE001 pass raise UnauthorizedError("Invalid credential.") def _curated_spec() -> dict: """Build an OpenAPI spec containing only the ``/api/v1/open`` data routes (excludes these meta endpoints themselves).""" from fastapi.openapi.utils import get_openapi from app.domains.asset.open_feishu_options_router import router as feishu_router from app.domains.cmdb.open_billing_router import ( router as cloud_billing_router, ) from app.domains.cmdb.open_cloud_router import router as cloud_router from app.domains.hr.open_holiday_router import router as holiday_router from app.domains.hr.open_leave_router import router as leave_router from app.identity.open_people_router import router as people_router # Use the source routers rather than ``app.routes``. FastAPI 0.141 keeps # included routers as deferred ``_IncludedRouter`` objects, so filtering # the application's top-level routes by ``.path`` silently produced an # empty spec. Listing the External API routers explicitly also keeps the # separate /open/directory service-credential surface out by construction. _excluded = {"/api/v1/open/openapi.json", "/api/v1/open/docs"} routes = [ route for source_router in ( router, cloud_router, cloud_billing_router, feishu_router, leave_router, holiday_router, people_router, ) for route in source_router.routes if getattr(route, "path", "") not in _excluded ] spec = get_openapi( title="Helix External API", version="v1", description=( "Scoped, machine-to-machine access to Helix resources.\n\n" "**Auth:** `Authorization: Bearer helix_api_…` (issued by a " "Helix admin). Verify with `GET /api/v1/open/me`." ), routes=routes, ) spec.setdefault("servers", [{"url": ""}]) return spec @router.get("/openapi.json", include_in_schema=False) async def open_openapi( request: Request, db: AsyncSession = Depends(get_db), key: str | None = Query(default=None), ) -> JSONResponse: await _authorize_reference(request, db, key) return JSONResponse(_curated_spec()) __all__ = ["router"]