"""HTTP API for the identity-consumer self-service request flow. * ``request_router`` — authenticated self-service + admin governance: POST /api/v1/identity/consumer-requests submit (any active employee) GET /api/v1/identity/consumer-requests list (mine; admin: ?all=true) GET /api/v1/identity/consumer-requests/{id} detail (requester or admin) POST /api/v1/identity/consumer-requests/{id}/cancel cancel (requester) * ``issue_router`` — the no-login one-time token-issue link delivered to the applicant after approval (action-token authorised, like the provisioning quick-complete page): GET /api/v1/open/consumer-token/issue?token=… render-only page POST /api/v1/open/consumer-token/issue mint the first token (once) """ from __future__ import annotations import html from datetime import UTC, datetime from uuid import UUID import jwt from fastapi import APIRouter, Depends, Form, Query, Request from fastapi.responses import HTMLResponse from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from app.auth.models import AuthUser from app.core.database import get_db from app.core.dependencies import ( enforce_action_token_subject, get_current_user, ) from app.core.dependencies import ( require_active_employee_for_mutations as _leaver_gate, ) from app.core.exceptions import AppException, NotFoundError from app.core.response import success_response from app.core.security import decode_action_token from app.identity.consumers.models import ( IdentityConsumer, IdentityConsumerRequest, ) from app.identity.consumers.request_schemas import ( ConsumerRequestCreate, ConsumerRequestResponse, ) from app.identity.consumers.request_service import ( ACTION_TOKEN_PURPOSE_ISSUE, ConsumerRequestService, ) from app.identity.consumers.token_service import ConsumerTokenService, token_is_active from app.identity.person_models import IdentityPerson request_router = APIRouter( prefix="/api/v1/identity/consumer-requests", tags=["identity-consumers"], ) issue_router = APIRouter( prefix="/api/v1/open/consumer-token", tags=["open-directory"], ) async def _person(db: AsyncSession, user: AuthUser) -> IdentityPerson: person = await db.get(IdentityPerson, user.id) if person is None: raise NotFoundError("Person", str(user.id)) return person def _to_response( req: IdentityConsumerRequest, people: dict[UUID, IdentityPerson], ) -> ConsumerRequestResponse: requester = people.get(req.requester_person_id) owner = people.get(req.owner_person_id) base = ConsumerRequestResponse.model_validate(req) base.requester_display_name = requester.display_name if requester else None base.owner_display_name = owner.display_name if owner else None return base @request_router.post("", summary="Apply to register an identity consumer") async def submit_request( body: ConsumerRequestCreate, request: Request, db: AsyncSession = Depends(get_db), user: AuthUser = Depends(get_current_user), _gate: None = Depends(_leaver_gate), ): requester = await _person(db, user) req = await ConsumerRequestService(db).submit( requester=requester, body=body, trace_id=str(getattr(request.state, "trace_id", "") or ""), ) people = await _load_people(db, [req]) return success_response( data=_to_response(req, people).model_dump(mode="json"), request=request, ) @request_router.get("", summary="List identity-consumer requests") async def list_requests( request: Request, all_requests: bool = Query(default=False, alias="all"), db: AsyncSession = Depends(get_db), user: AuthUser = Depends(get_current_user), ): stmt = ( select(IdentityConsumerRequest) .where(IdentityConsumerRequest.deleted_at.is_(None)) .order_by(IdentityConsumerRequest.created_at.desc()) ) if all_requests: # Admin-only firehose; non-admins silently get their own rows. if not _is_admin_reader(user): stmt = stmt.where(IdentityConsumerRequest.requester_person_id == user.id) else: stmt = stmt.where(IdentityConsumerRequest.requester_person_id == user.id) rows = list((await db.execute(stmt)).scalars().all()) people = await _load_people(db, rows) return success_response( data=[_to_response(r, people).model_dump(mode="json") for r in rows], request=request, ) @request_router.get("/catalog", summary="Scope/field catalog for the apply form") async def request_catalog( request: Request, _user: AuthUser = Depends(get_current_user), ): """Read-only metadata any authenticated employee needs to render the self-service apply form (the admin ``/scopes`` endpoint is admin-only). No consumer data — just the bundle/field/expiry catalog.""" from app.identity.consumers.field_catalog import catalog_payload from app.identity.consumers.models import KNOWN_SCOPES from app.identity.consumers.token_service import ALLOWED_EXPIRY_DAYS return success_response( data={ "scopes": sorted(KNOWN_SCOPES), "fields": catalog_payload(), "token_expiry_days": sorted(ALLOWED_EXPIRY_DAYS), }, request=request, ) @request_router.get("/{request_id}", summary="Identity-consumer request detail") async def get_request( request_id: UUID, request: Request, db: AsyncSession = Depends(get_db), user: AuthUser = Depends(get_current_user), ): row = await db.get(IdentityConsumerRequest, request_id) if row is None or row.deleted_at is not None: raise NotFoundError("IdentityConsumerRequest", str(request_id)) if row.requester_person_id != user.id and not _is_admin_reader(user): raise AppException( code="CONSUMER_REQUEST_FORBIDDEN", message="You can only view your own requests.", status_code=403, ) people = await _load_people(db, [row]) return success_response( data=_to_response(row, people).model_dump(mode="json"), request=request, ) @request_router.post("/{request_id}/cancel", summary="Cancel a pending request") async def cancel_request( request_id: UUID, request: Request, db: AsyncSession = Depends(get_db), user: AuthUser = Depends(get_current_user), ): actor = await _person(db, user) req = await ConsumerRequestService(db).cancel( request_id=request_id, actor=actor, trace_id=str(getattr(request.state, "trace_id", "") or ""), ) people = await _load_people(db, [req]) return success_response( data=_to_response(req, people).model_dump(mode="json"), request=request, ) def _is_admin_reader(user: AuthUser) -> bool: from app.core.rbac import ( AUDITOR_ROLE, GLOBAL_ADMIN_ROLE, MANAGEMENT_ROLE, has_role, ) # ``role_names`` is already canonicalised (legacy ``it_admin`` / # ``security_auditor`` aliases mapped to their modern names). return has_role(user.role_names, GLOBAL_ADMIN_ROLE, AUDITOR_ROLE, MANAGEMENT_ROLE) async def _load_people( db: AsyncSession, rows: list[IdentityConsumerRequest], ) -> dict[UUID, IdentityPerson]: ids = {r.requester_person_id for r in rows} | {r.owner_person_id for r in rows} if not ids: return {} result = await db.execute( select(IdentityPerson).where(IdentityPerson.id.in_(ids)) ) return {p.id: p for p in result.scalars().all()} # ── One-time token-issue link (no login; action-token authorised) ────── def _issue_page(*, title: str, heading: str, body_html: str, tone: str = "") -> HTMLResponse: color = {"ok": "#16a34a", "err": "#dc2626"}.get(tone, "#111827") return HTMLResponse( f""" {html.escape(title)}
{html.escape(heading)}
{body_html}""", ) def _issue_invalid() -> HTMLResponse: return _issue_page( title="链接无效", heading="链接无效或已过期", tone="err", body_html="

该领取链接无效或已过期。请联系 IT 管理员重新发放令牌。

", ) @issue_router.get("/issue", include_in_schema=False) async def issue_page( token: str = Query(...), db: AsyncSession = Depends(get_db), ): """Render-only: show the consumer + a confirm button. No side effects (email clients prefetch GETs).""" try: payload = decode_action_token(token, expected_purpose=ACTION_TOKEN_PURPOSE_ISSUE) consumer_id = UUID(payload["res"]) except (jwt.InvalidTokenError, ValueError, KeyError): return _issue_invalid() consumer = await _load_consumer(db, consumer_id) if consumer is None: return _issue_invalid() # Single-use: the request's token_issued_at marker is authoritative # (survives a later token revocation, so revoking does NOT re-open # this no-login link). Fall back to the active-token check for any # pre-288 row with a null marker. req = await _latest_request(db, consumer_id) if (req is not None and req.token_issued_at is not None) or any( token_is_active(t) for t in consumer.tokens ): return _issue_page( title="令牌已发放", heading="服务令牌已发放", tone="ok", body_html=( f"

消费方 {html.escape(consumer.slug)} " "已经领取过服务令牌。如需轮换,请联系 IT 管理员在控制台铸造新令牌。

" ), ) return _issue_page( title="领取服务令牌", heading=f"领取「{html.escape(consumer.name)}」的服务令牌", body_html=( f"

消费方标识:{html.escape(consumer.slug)}

" "

点击下方按钮生成服务令牌。令牌仅展示一次,请立即复制保存。

" f"
" f"" "
" ), ) @issue_router.post("/issue", include_in_schema=False) async def issue_submit( token: str = Form(...), db: AsyncSession = Depends(get_db), ): """Mint the consumer's first token (once). The action token is the sole authorisation — no login.""" try: payload = decode_action_token(token, expected_purpose=ACTION_TOKEN_PURPOSE_ISSUE) consumer_id = UUID(payload["res"]) subject_id = UUID(payload["sub"]) # A departed / force-logged-out requester's stale link must not still # mint a consumer token — raises ValueError → friendly invalid page. await enforce_action_token_subject(db, payload) except (jwt.InvalidTokenError, ValueError, KeyError): return _issue_invalid() consumer = await _load_consumer(db, consumer_id) if consumer is None: return _issue_invalid() # Single-use, race-free: lock the originating request row, then gate # on its token_issued_at marker (authoritative — survives token # revocation) AND the active-token check (covers any pre-288 row). # Two concurrent POSTs serialize on the lock; the second sees the # marker set and refuses. req = await _latest_request(db, consumer_id, lock=True) already = (req is not None and req.token_issued_at is not None) or any( token_is_active(t) for t in consumer.tokens ) if already: await db.rollback() return _issue_page( title="令牌已发放", heading="服务令牌已发放", tone="ok", body_html="

该消费方已领取过令牌。如需轮换请联系 IT 管理员。

", ) # Token expiry preset comes from the originating request. Guard # against an off-preset value (mint rejects anything not in # ALLOWED_EXPIRY_DAYS) so a bad request can't strand the consumer at # zero tokens on a spent link. from app.identity.consumers.token_service import ALLOWED_EXPIRY_DAYS expires_in_days = req.token_validity_days if req else None if expires_in_days is not None and expires_in_days not in ALLOWED_EXPIRY_DAYS: expires_in_days = None # fall back to never-expire rather than crash token_row, plaintext = await ConsumerTokenService(db).mint( consumer=consumer, name="initial (self-service)", expires_in_days=expires_in_days, created_by=subject_id, ) if req is not None: req.token_issued_at = datetime.now(UTC) await db.commit() return _issue_page( title="令牌已生成", heading="服务令牌已生成", tone="ok", body_html=( "

请立即复制并妥善保存 — 此令牌仅展示这一次。

" f"
{html.escape(plaintext)}
" f"

前缀:{html.escape(token_row.token_prefix)}

" + _issue_quickstart_html( consumer=consumer, plaintext=plaintext, base_url=await _frontend_base_url(db), ) ), ) #: One-line, applicant-facing description of each scope, shown on the #: token-issue page so a key-holder sees what their credential unlocks #: without first reading the docs. Kept in lockstep with the catalog #: bundles (``app.identity.consumers.field_catalog``). _SCOPE_BLURB: dict[str, str] = { "directory.persons:read": "人员核心字段(手机号/个人邮箱默认脱敏)", "directory.org:read": "组织树与人员-部门归属", "directory.pii:read": "手机号/个人邮箱明文(需配合人员读取)", } async def _frontend_base_url(db: AsyncSession) -> str: """Portal/API base (e.g. ``https://helix.pingcap.net``) for the copy-ready examples on the issue page. Empty string when unset — the quickstart then renders a relative path placeholder.""" from app.integrations.service import get_platform_setting # noqa: PLC0415 return str(await get_platform_setting( "app.frontend_base_url", default="", session=db, ) or "").rstrip("/") def _issue_quickstart_html( *, consumer: IdentityConsumer, plaintext: str, base_url: str, ) -> str: """The "how to use this credential" block appended after the minted token — granted scopes, a copy-ready ``GET /me`` smoke test, the next step, and where to read the full contract. Self-service applicants rarely know the API surface, so we hand it to them inline rather than leaving them with a bare token string.""" base = html.escape(base_url) if base_url else "https://helix.pingcap.net" scopes = "".join( f"
  • {html.escape(s)} — {html.escape(_SCOPE_BLURB.get(s, ''))}
    • " for s in (consumer.scopes or []) ) curl = ( f"curl -H 'Authorization: Bearer {html.escape(plaintext)}' \\\n" f" {base}/api/v1/open/directory/me" ) return ( "
    " "
    怎么用这个令牌
    " "

    每次调用都带请求头 " "Authorization: Bearer <令牌>。先用下面这条命令验证凭证" "(返回你的消费方标识、已授权范围与可见人群即为成功):

    " f"
    {html.escape(curl)}
    " f"

    已授权范围:

    " "

    下一步:验证通过后,循环拉 " f"{base}/api/v1/open/directory/persons 完成首次全量," "之后用响应里的 next_cursor 分钟级轮询做增量。组织数据见 " "/org/nodes 与 /org/memberships;" "支持 SCIM 的系统可直接对接 /scim/v2/Users 与 /Groups。

    " f"

    完整说明:登录 Helix → 顶部「API 文档」→「人员 / 组织目录」标签页" f"({base}/api-docs;含端点契约、增量协议、错误语义、SCIM)," "或帮助中心 →「我要从 Helix 拉取人员/组织数据」。

    " ) async def _latest_request( db: AsyncSession, consumer_id: UUID, *, lock: bool = False, ) -> IdentityConsumerRequest | None: """The request that provisioned this consumer (newest first). When ``lock`` is set, take a row lock so the check-then-stamp of the single-use ``token_issued_at`` marker is race-free.""" stmt = ( select(IdentityConsumerRequest) .where(IdentityConsumerRequest.consumer_id == consumer_id) .order_by(IdentityConsumerRequest.created_at.desc()) .limit(1) # Force a refresh on an identity-map hit so the single-use # token_issued_at marker is read fresh, not from a stale cached # instance (a second request in the same session would otherwise # see the pre-mint value and bypass the guard). .execution_options(populate_existing=True) ) if lock: stmt = stmt.with_for_update() return (await db.execute(stmt)).scalars().first() async def _load_consumer(db: AsyncSession, consumer_id: UUID) -> IdentityConsumer | None: # populate_existing forces the token relationship to reload even on # an identity-map hit — without it a second issue attempt could read # a stale (empty) tokens collection and bypass the replay guard # (mirrors _get_consumer in router.py). row = (await db.execute( select(IdentityConsumer) .where(IdentityConsumer.id == consumer_id) .options(selectinload(IdentityConsumer.tokens)) .execution_options(populate_existing=True) )).scalar_one_or_none() if row is None or row.deleted_at is not None: return None return row __all__ = ["issue_router", "request_router"]