"""Self-service identity-consumer request flow. An owner applies in Helix (rich form: scope/field grant + population filter + token validity). On submit Helix creates a Feishu approval instance pre-filled with the request and stores the correlation. The security/data-governance approver decides in Feishu; the callback (``handle_feishu_callback``) materialises the live ``IdentityConsumer`` on approval and hands the applicant a **one-time token-issue link** (action-token, no plaintext ever stored). The Feishu template (flow_key ``identity_consumer_request``, seeded by migration 286) is filled by matching control NAME keywords — the same technique as ``domain_registration`` — extended to resolve single-choice (radioV2) controls against their registered option set (Feishu rejects free-form text on radioV2 with API 1390001). """ from __future__ import annotations import json import logging import uuid from datetime import UTC, datetime, timedelta from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.approval.cancel_desync import notify_feishu_cancel_desync from app.approval.feishu_client import FeishuApprovalClient from app.approval.models import ApprovalRequest, ApprovalSubmissionIntent from app.approval.submission_intents import ( compact_terminal_intent_payload, defer_submission_intent_retry, reconciliation_due_predicate, reconciliation_order, schedule_submission_intent_retry, submission_fingerprint, submit_with_durable_intent, terminal_intent_resource_id, ) from app.audit.service import AuditService from app.core.exceptions import ( AppException, ConflictError, NotFoundError, ValidationError, ) from app.core.person_filters import FilterConfig, FilterError from app.core.security import create_action_token from app.identity.consumers.field_catalog import ( PII_FIELD_KEYS, validate_granted_fields, ) from app.identity.consumers.models import ( CONSUMER_REQUEST_FLOW_KEY, REQUEST_STATUS_APPROVED, REQUEST_STATUS_CANCELLED, REQUEST_STATUS_FAILED, REQUEST_STATUS_PENDING, REQUEST_STATUS_REJECTED, IdentityConsumer, IdentityConsumerRequest, ) from app.identity.consumers.request_schemas import ConsumerRequestCreate from app.identity.person_models import IdentityPerson from app.integrations.compat import get_feishu_config from app.integrations.service import get_approval_flow logger = logging.getLogger(__name__) _INTENT_RECOVERABLE_STATES = ( "prepared", "submitting", "unknown", "provider_created", "compensation_pending", ) #: Action-token purpose for the post-approval one-time token-issue link. ACTION_TOKEN_PURPOSE_ISSUE = "consumer_token_issue" #: Canonical pending value for the *platform* ``ApprovalRequest`` row. #: This MUST be ``"pending"`` to match every sibling approval flow #: (DNS / VPN / SSO / domain / scope / api-key …) AND the lost-webhook #: safety net ``feishu.approval_pending_reaper``, which filters #: ``ApprovalRequest.status == "pending"``. The consumer-request row's #: own status vocabulary (``IdentityConsumerRequest.status``) keeps the #: domain-specific ``REQUEST_STATUS_PENDING = "pending_approval"`` — the #: two are different tables and must not be conflated, or a #: consumer-request whose Feishu decision is never delivered is never #: reaped and the applicant is never told it expired. APPROVAL_REQUEST_PENDING = "pending" #: Control-name keywords → logical field. Disambiguated so "Requested #: Data Scope" and "Population Scope" don't collide (no bare "scope"). _FIELD_KEYWORDS: dict[str, tuple[str, ...]] = { "name": ("system name", "consumer system", "系统名称", "名称"), "slug": ("slug", "标识"), "purpose": ("access purpose", "purpose", "用途", "接入用途"), "owner": ("owner", "负责人"), "scope": ("requested data", "data scope", "数据范围", "申请数据"), "pii_flag": ("plaintext pii", "pii", "明文"), "pii_justification": ("justification", "理由"), "population": ("population", "人群"), "rate": ("rate", "频率", "限流"), "validity": ("validity", "有效期", "token validity"), } _SKIP_WIDGET_TYPES = frozenset({"text", "section", "image", "attachmentV2"}) def _match_field(widget_name: str) -> str | None: name = (widget_name or "").lower() for field_key, kws in _FIELD_KEYWORDS.items(): if any(kw.lower() in name for kw in kws): return field_key return None def _resolve_option(widget: dict[str, Any], logical_value: str) -> str | None: """Resolve a logical single-choice value to the widget's option machine key. Feishu radioV2 accepts ONLY a registered option value, not its display text — match against option text/value and return the option's ``value`` (machine key).""" want = (logical_value or "").strip().lower() options = widget.get("options") or [] for opt in options: text = str(opt.get("text") or "").strip().lower() val = str(opt.get("value") or "").strip().lower() if want and (want == text or want == val or want in text): return str(opt.get("value") or opt.get("text")) return None class ConsumerRequestService: def __init__(self, db: AsyncSession) -> None: self.db = db self.audit = AuditService(db) self._feishu_client: FeishuApprovalClient | None = None async def _client(self) -> FeishuApprovalClient: if self._feishu_client is None: cfg = await get_feishu_config(session=self.db) self._feishu_client = FeishuApprovalClient( app_id=str(cfg.get("app_id") or ""), app_secret=str(cfg.get("app_secret") or ""), ) return self._feishu_client # ── Submit ────────────────────────────────────────────────── async def submit( self, *, requester: IdentityPerson, body: ConsumerRequestCreate, trace_id: str, ) -> IdentityConsumerRequest: from app.core.distributed_lock import ( # noqa: PLC0415 LockBusy, distributed_lock, ) requester_id = requester.id try: async with distributed_lock( f"identity_consumer_request_submit:{requester_id}", ttl=120, ): await self.db.rollback() current_requester = await self.db.scalar( select(IdentityPerson) .where(IdentityPerson.id == requester_id) .with_for_update() .execution_options(populate_existing=True) ) if current_requester is None: raise AppException( code="CONSUMER_REQUEST_REQUESTER_NOT_ACTIVE", message="The requester no longer exists.", status_code=403, ) return await self._submit_locked( requester=current_requester, body=body, trace_id=trace_id, ) except LockBusy as exc: raise AppException( code="CONSUMER_REQUEST_SUBMIT_IN_FLIGHT", message=( "Another identity-consumer request is being submitted. " "Please retry in a few seconds." ), status_code=409, ) from exc async def _submit_locked( self, *, requester: IdentityPerson, body: ConsumerRequestCreate, trace_id: str, ) -> IdentityConsumerRequest: # Active-only — positive check so this service gate matches the # HTTP route's ``require_active_employee_for_mutations`` # (``employment_status == 'active'``) exactly. A bare # ``in ("leaver_pending", "left")`` would let a ``pre_hire`` # requester through the MCP path (which has no leaver dependency) # while the HTTP path rejects them — two surfaces disagreeing on # authorization. if requester.employment_status != "active": raise AppException( code="CONSUMER_REQUEST_REQUESTER_NOT_ACTIVE", message=( "Your account is not an active employee; new " "identity-consumer requests are not accepted." ), status_code=403, ) owner = await self._resolve_owner(body, requester) granted_fields = self._validate_grant(body) filter_config = self._validate_filter(body.filter_config) self._check_pii_justification(body, granted_fields) await self._reject_if_slug_taken(body.slug) existing_pending = await self.db.scalar( select(IdentityConsumerRequest.id).where( IdentityConsumerRequest.requester_person_id == requester.id, IdentityConsumerRequest.slug == body.slug, IdentityConsumerRequest.status == REQUEST_STATUS_PENDING, ).limit(1).with_for_update() ) if existing_pending is not None: raise ConflictError( f"A request for consumer slug '{body.slug}' is already pending." ) req = IdentityConsumerRequest( id=uuid.uuid4(), requester_person_id=requester.id, name=body.name.strip(), slug=body.slug, description=(body.description or "").strip() or None, owner_person_id=owner.id, scopes=body.scopes, granted_fields=granted_fields, filter_config=filter_config, rate_limit_per_minute=body.rate_limit_per_minute, token_validity_days=body.token_validity_days, pii_justification=(body.pii_justification or "").strip() or None, status=REQUEST_STATUS_PENDING, ) intent = await self._open_approval( req, requester, owner, granted_fields, trace_id=trace_id, ) finalized = await self._finalize_submission_intent( intent.id, recovered=( intent.attempt_count > 1 or str((intent.payload or {}).get("trace_id") or "") != trace_id ), ) if finalized.status == REQUEST_STATUS_CANCELLED: raise AppException( code="CONSUMER_REQUEST_REQUESTER_NOT_ACTIVE", message=( "The requester or proposed owner is no longer active; " "the Feishu approval has been withdrawn." ), status_code=403, ) return finalized async def _resolve_owner( self, body: ConsumerRequestCreate, requester: IdentityPerson, ) -> IdentityPerson: owner_id = body.owner_person_id or requester.id owner = await self.db.get(IdentityPerson, owner_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( "Consumer owner must be an active employee." ) return owner def _validate_grant(self, body: ConsumerRequestCreate) -> list[str]: try: return validate_granted_fields(body.granted_fields, body.scopes) except ValueError as e: raise ValidationError(f"granted_fields invalid: {e}") from e def _validate_filter(self, raw: dict) -> dict: try: return FilterConfig.from_dict(raw or {}).to_dict() except FilterError as e: raise ValidationError(f"filter_config invalid: {e}") from e def _check_pii_justification( self, body: ConsumerRequestCreate, granted_fields: list[str], ) -> None: wants_pii = bool(set(granted_fields) & PII_FIELD_KEYS) if wants_pii and not (body.pii_justification or "").strip(): raise ValidationError( "A PII justification is required when the grant includes " "plaintext PII fields (mobile_phone / personal_email)." ) async def _reject_if_slug_taken(self, slug: str) -> None: # Best-effort at submit; provisioning re-checks authoritatively. existing = await self.db.execute( select(IdentityConsumer.id).where(IdentityConsumer.slug == slug) ) if existing.scalar_one_or_none() is not None: raise ConflictError(f"Consumer slug '{slug}' already exists.") async def _open_approval( self, req: IdentityConsumerRequest, requester: IdentityPerson, owner: IdentityPerson, granted_fields: list[str], *, trace_id: str, ) -> ApprovalSubmissionIntent: flow = await get_approval_flow(CONSUMER_REQUEST_FLOW_KEY, session=self.db) approval_code = flow.external_code client = await self._client() definition = await client.get_approval_definition(approval_code) widgets = _parse_widgets(definition.get("form", "[]")) values = self._form_values(req, owner, granted_fields) form_payload = self._build_form_payload(widgets, values) feishu_user_id = await client.resolve_user_id_by_email(requester.work_email) recovery_key = submission_fingerprint({"slug": req.slug}) return await submit_with_durable_intent( self.db, client=client, scope=CONSUMER_REQUEST_FLOW_KEY, applicant_id=requester.id, recovery_key=recovery_key, semantic_payload={ "name": req.name, "slug": req.slug, "description": req.description, "owner_person_id": str(owner.id), "scopes": req.scopes, "granted_fields": granted_fields, "filter_config": req.filter_config, "rate_limit_per_minute": req.rate_limit_per_minute, "token_validity_days": req.token_validity_days, "pii_justification": req.pii_justification, }, provider_payload={ "approval_code": approval_code, "identity_type": "user_id", "identity_value": feishu_user_id, "form": form_payload, }, finalization_payload={ "consumer_request_id": str(req.id), "approval_request_id": str(uuid.uuid4()), "name": req.name, "slug": req.slug, "description": req.description, "owner_person_id": str(owner.id), "owner_work_email": owner.work_email, "scopes": req.scopes, "granted_fields": granted_fields, "filter_config": req.filter_config, "rate_limit_per_minute": req.rate_limit_per_minute, "token_validity_days": req.token_validity_days, "pii_justification": req.pii_justification, "approval_code": approval_code, }, trace_id=trace_id, ) async def _finalize_submission_intent( self, intent_id: uuid.UUID, *, recovered: bool, ) -> IdentityConsumerRequest: intent = await self.db.scalar( select(ApprovalSubmissionIntent) .where(ApprovalSubmissionIntent.id == intent_id) .with_for_update() .execution_options(populate_existing=True) ) if intent is None or intent.scope != CONSUMER_REQUEST_FLOW_KEY: raise RuntimeError("identity-consumer submission intent disappeared") if intent.state in {"completed", "compensated"}: request_id = uuid.UUID(terminal_intent_resource_id( intent, resource_type="identity_consumer_request", )) existing = await self.db.get(IdentityConsumerRequest, request_id) if existing is None: raise RuntimeError("terminal identity-consumer intent has no request") return existing finalization = dict((intent.payload or {}).get("finalization") or {}) request_id = uuid.UUID(str(finalization["consumer_request_id"])) existing = await self.db.get(IdentityConsumerRequest, request_id) if intent.state == "compensation_pending": if existing is None: raise RuntimeError("identity-consumer compensation has no request") await self.db.commit() await self._compensate_provider_intent(intent_id) await self.db.refresh(existing) return existing if intent.state != "provider_created" or not intent.external_id: raise RuntimeError( f"cannot finalize identity-consumer intent in {intent.state!r}" ) person_ids = sorted({ intent.applicant_id, uuid.UUID(str(finalization["owner_person_id"])), }, key=str) people = list((await self.db.scalars( select(IdentityPerson) .where(IdentityPerson.id.in_(person_ids)) .order_by(IdentityPerson.id) .with_for_update() .execution_options(populate_existing=True) )).all()) people_by_id = {person.id: person for person in people} requester = people_by_id.get(intent.applicant_id) owner_id = uuid.UUID(str(finalization["owner_person_id"])) owner = people_by_id.get(owner_id) compensate = ( requester is None or requester.deleted_at is not None or requester.employment_status != "active" or owner is None or owner.deleted_at is not None or owner.employment_status != "active" ) local_status = ( REQUEST_STATUS_CANCELLED if compensate else REQUEST_STATUS_PENDING ) if existing is not None: raise RuntimeError( "provider-created identity-consumer intent already has a request" ) now = datetime.now(UTC) request = IdentityConsumerRequest( id=request_id, requester_person_id=intent.applicant_id, name=str(finalization["name"]), slug=str(finalization["slug"]), description=finalization.get("description"), owner_person_id=owner_id, scopes=list(finalization.get("scopes") or []), granted_fields=list(finalization.get("granted_fields") or []), filter_config=dict(finalization.get("filter_config") or {}), rate_limit_per_minute=finalization.get("rate_limit_per_minute"), token_validity_days=finalization.get("token_validity_days"), pii_justification=finalization.get("pii_justification"), status=local_status, feishu_instance_code=intent.external_id, decided_at=now if compensate else None, created_at=now, updated_at=now, ) self.db.add(request) await self.db.flush() approval_request_id = uuid.UUID(str(finalization["approval_request_id"])) approval_request = ApprovalRequest( id=approval_request_id, external_id=intent.external_id, approval_type=CONSUMER_REQUEST_FLOW_KEY, applicant_id=intent.applicant_id, status=(REQUEST_STATUS_CANCELLED if compensate else APPROVAL_REQUEST_PENDING), source_payload={ "consumer_request_id": str(request_id), "approval_code": finalization["approval_code"], "slug": finalization["slug"], "submission_intent_id": str(intent.id), }, ) self.db.add(approval_request) await self.db.flush() request.approval_request_id = approval_request_id intent.approval_request_id = approval_request_id intent.state = "compensation_pending" if compensate else "completed" intent.completed_at = None if compensate else now intent.last_error = None await self.audit.log( trace_id=( f"approval-intent-reconcile:{intent.id}" if recovered else str((intent.payload or {}).get("trace_id")) ), actor_type="system" if recovered or compensate else "user", actor_user_id=None if recovered or compensate else intent.applicant_id, action=( "identity_consumer_request.submit_compensating_ineligible" if compensate else ( "approval.submission_intent.reconciled" if recovered else "identity_consumer_request.submit" ) ), status="failure" if compensate else "success", resource_type="identity_consumer_request", resource_id=str(request_id), detail={ "intent_id": str(intent.id), "provider_request_uuid": intent.provider_request_uuid, "feishu_instance_code": intent.external_id, "slug": finalization["slug"], "owner": finalization.get("owner_work_email"), "scopes": finalization.get("scopes") or [], "recovered": recovered, "compensation_required": compensate, }, ) if not compensate: compact_terminal_intent_payload( intent, resource_type="identity_consumer_request", resource_id=request_id, ) await self.db.commit() await self.db.refresh(request) if compensate: await self._compensate_provider_intent(intent_id) await self.db.refresh(request) return request async def _compensate_provider_intent(self, intent_id: uuid.UUID) -> bool: intent = await self.db.get(ApprovalSubmissionIntent, intent_id) if intent is None: raise RuntimeError("identity-consumer compensation intent disappeared") if intent.state == "compensated": return True if intent.state != "compensation_pending" or not intent.external_id: raise RuntimeError( f"cannot compensate identity-consumer intent in {intent.state!r}" ) payload = dict(intent.payload or {}) stored_provider = dict(payload.get("provider") or {}) request_id = str( (payload.get("finalization") or {}).get("consumer_request_id") ) instance_code = intent.external_id await self.db.rollback() client = await self._client() try: detail = await client.get_instance_detail(instance_code) except Exception: # noqa: BLE001 detail = {} provider_status = str(detail.get("status") or "").upper() already_closed = provider_status in { "CANCELED", "CANCELLED", "REJECTED", "REFUSED", "DELETED", "TIMEOUT", "RECALLED", "REVERTED", } try: if not already_closed: identity_type = str(stored_provider["identity_type"]) identity_value = str(stored_provider["identity_value"]) await client.cancel_instance( instance_code=instance_code, approval_code=str(stored_provider["approval_code"]), user_id=identity_value if identity_type == "user_id" else None, open_id=identity_value if identity_type == "open_id" else None, ) except Exception as exc: # noqa: BLE001 await self.db.rollback() intent = await self.db.get(ApprovalSubmissionIntent, intent_id) if intent is not None and intent.state == "compensation_pending": schedule_submission_intent_retry(intent, exc) await self.audit.log( trace_id=f"approval-intent-compensate:{intent_id}", actor_type="system", actor_user_id=None, action="approval.submission_intent.compensation_pending", status="failure", resource_type="identity_consumer_request", resource_id=request_id, detail={ "intent_id": str(intent_id), "instance_code": instance_code, "error_type": type(exc).__name__, }, ) await notify_feishu_cancel_desync( self.db, flow_label="身份消费方申请", resource_type="identity_consumer_submission_compensation", resource_id=request_id, instance_code=instance_code, error=exc, trace_id=f"approval-intent-compensate:{intent_id}", ) await self.db.commit() return False await self.db.rollback() intent = await self.db.scalar( select(ApprovalSubmissionIntent) .where(ApprovalSubmissionIntent.id == intent_id) .with_for_update() .execution_options(populate_existing=True) ) if intent is None: raise RuntimeError("identity-consumer compensation intent disappeared") if intent.state == "compensated": return True if intent.state != "compensation_pending": raise RuntimeError("identity-consumer compensation state changed") intent.state = "compensated" intent.completed_at = datetime.now(UTC) intent.last_error = None await self.audit.log( trace_id=f"approval-intent-compensate:{intent_id}", actor_type="system", actor_user_id=None, action="approval.submission_intent.compensated", status="success", resource_type="identity_consumer_request", resource_id=request_id, detail={ "intent_id": str(intent_id), "instance_code": instance_code, "provider_status_before_cancel": provider_status or None, "already_closed": already_closed, }, ) compact_terminal_intent_payload( intent, resource_type="identity_consumer_request", resource_id=request_id, ) await self.db.commit() return True async def reconcile_submission_intents( self, *, min_age_minutes: int = 2, batch_limit: int = 100, ) -> dict[str, int]: cutoff = datetime.now(UTC) - timedelta(minutes=min_age_minutes) refs = list((await self.db.execute( select( ApprovalSubmissionIntent.id, ApprovalSubmissionIntent.applicant_id, ) .where( ApprovalSubmissionIntent.scope == CONSUMER_REQUEST_FLOW_KEY, ApprovalSubmissionIntent.state.in_(_INTENT_RECOVERABLE_STATES), ( ApprovalSubmissionIntent.state.in_(( "provider_created", "compensation_pending", )) | (ApprovalSubmissionIntent.updated_at < cutoff) ), reconciliation_due_predicate(datetime.now(UTC)), ) .order_by(*reconciliation_order()) .limit(batch_limit) )).all()) await self.db.rollback() summary = { "checked": len(refs), "reconciled": 0, "compensated": 0, "still_unknown": 0, "failed": 0, } from app.core.distributed_lock import ( # noqa: PLC0415 LockBusy, distributed_lock, ) for intent_id, applicant_id in refs: try: async with distributed_lock( f"identity_consumer_request_submit:{applicant_id}", ttl=120, ): intent = await self.db.get(ApprovalSubmissionIntent, intent_id) if intent is None or intent.state not in _INTENT_RECOVERABLE_STATES: continue payload = dict(intent.payload or {}) if intent.state != "compensation_pending": intent = await submit_with_durable_intent( self.db, client=await self._client(), scope=intent.scope, applicant_id=intent.applicant_id, recovery_key=payload.get("recovery_key"), semantic_payload=dict(payload.get("semantic") or {}), provider_payload=dict(payload.get("provider") or {}), finalization_payload=dict( payload.get("finalization") or {} ), trace_id=str( payload.get("trace_id") or f"approval-intent:{intent.id}" ), ) row = await self._finalize_submission_intent( intent.id, recovered=True, ) refreshed = await self.db.get( ApprovalSubmissionIntent, intent.id, ) if refreshed is not None and refreshed.state == "compensated": summary["compensated"] += 1 elif ( refreshed is not None and refreshed.state == "compensation_pending" ): summary["still_unknown"] += 1 elif row.status == REQUEST_STATUS_PENDING: summary["reconciled"] += 1 except LockBusy as exc: await defer_submission_intent_retry(self.db, intent_id, exc) summary["still_unknown"] += 1 except AppException as exc: if exc.code == "APPROVAL_SUBMISSION_RESULT_UNKNOWN": await self.db.rollback() summary["still_unknown"] += 1 else: await defer_submission_intent_retry(self.db, intent_id, exc) summary["failed"] += 1 except Exception as exc: # noqa: BLE001 await defer_submission_intent_retry(self.db, intent_id, exc) summary["failed"] += 1 return summary def _form_values( self, req: IdentityConsumerRequest, owner: IdentityPerson, granted_fields: list[str], ) -> dict[str, str]: owner_label = ( f"{owner.display_name or owner.work_email} " f"({owner.employee_no or owner.work_email})" ) wants_pii = bool(set(granted_fields) & PII_FIELD_KEYS) return { "name": req.name, "slug": req.slug, "purpose": req.description or "(未填写)", "owner": owner_label, "scope": _render_scope_summary(req.scopes, granted_fields), "pii_flag": "Yes" if wants_pii else "No", "pii_justification": req.pii_justification or "", "population": _render_filter_summary(req.filter_config), "rate": str(req.rate_limit_per_minute) if req.rate_limit_per_minute else "", "validity": _render_validity(req.token_validity_days), } def _build_form_payload( self, widgets: list[dict[str, Any]], values: dict[str, str], ) -> str: from app.integrations.feishu_form_visibility import ( evaluate_display_conditions, ) visibility = evaluate_display_conditions( widgets, {}, assume_visible_when_answers_unsourced=True, ) serialized: list[dict[str, Any]] = [] for widget in widgets: wtype = widget.get("type", "input") if wtype in _SKIP_WIDGET_TYPES: continue wid = widget.get("id") if not wid or not visibility.get(wid, True): continue field_key = _match_field(widget.get("name") or "") if field_key is None or field_key not in values: continue raw_value = values[field_key] if raw_value in (None, "", [], {}): continue if wtype in ("radioV2", "select"): resolved = _resolve_option(widget, raw_value) if resolved is None: # Can't safely map to an option — skip rather than # ship a value Feishu will 1390001-reject. logger.warning( "consumer-request: no option match for widget '%s' " "value '%s' — skipping", widget.get("name"), raw_value, ) continue serialized.append({"id": wid, "type": wtype, "value": resolved}) else: serialized.append({"id": wid, "type": wtype, "value": str(raw_value)}) return json.dumps(serialized, ensure_ascii=False) # ── Cancel ────────────────────────────────────────────────── async def cancel( self, *, request_id: uuid.UUID, actor: IdentityPerson, trace_id: str, ) -> IdentityConsumerRequest: req = await self._get(request_id) if req.requester_person_id != actor.id: raise AppException( code="CONSUMER_REQUEST_NOT_OWNER", message="Only the requester can cancel this request.", status_code=403, ) if req.status != REQUEST_STATUS_PENDING: raise ValidationError("Only a pending request can be cancelled.") req.status = REQUEST_STATUS_CANCELLED req.decided_at = datetime.now(UTC) if req.approval_request_id: ar = await self.db.get(ApprovalRequest, req.approval_request_id) # ``ar`` is the platform ApprovalRequest — compare against the # canonical "pending", not the consumer-row vocabulary. if ar is not None and ar.status == APPROVAL_REQUEST_PENDING: ar.status = REQUEST_STATUS_CANCELLED await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor.id, action="identity_consumer_request.cancel", status="success", resource_type="identity_consumer_request", resource_id=str(req.id), detail={"slug": req.slug}, ) # Best-effort Feishu side-cancel (mirrors other flows). Needs the # Feishu user_id (not email) + the template's approval_code. if req.feishu_instance_code: try: client = await self._client() flow = await get_approval_flow( CONSUMER_REQUEST_FLOW_KEY, session=self.db, ) feishu_user_id = await client.resolve_user_id_by_email( actor.work_email, ) await client.cancel_instance( instance_code=req.feishu_instance_code, user_id=feishu_user_id, approval_code=flow.external_code, ) except Exception as exc: # noqa: BLE001 logger.warning("consumer-request Feishu cancel failed: %s", exc) await self.db.commit() await self.db.refresh(req) return req # ── Feishu callback (dispatched by approval/service.py) ────── async def handle_feishu_callback( self, *, approval_request: ApprovalRequest, instance_detail: dict[str, Any], ) -> None: source = approval_request.source_payload or {} req_id = source.get("consumer_request_id") if not req_id: return req = await self.db.get(IdentityConsumerRequest, uuid.UUID(str(req_id))) if req is None or req.status != REQUEST_STATUS_PENDING: return # already processed / unknown feishu_status = (instance_detail.get("status") or "").upper() now = datetime.now(UTC) req.decided_at = now # COMPLETED is Feishu's terminal "approved" state for some # templates; TIMEOUT/DELETED are terminal non-approvals. The # dispatcher (approval/service.py) already maps these to terminal # statuses on the parent row and won't redeliver — so this # handler MUST also settle, or the request would be stuck pending # forever while the dispatcher considers it done (review fix # 2026-06-15; mirrors _map_feishu_status semantics). if feishu_status in ("APPROVED", "PASS", "COMPLETED"): await self._provision(req) elif feishu_status in ("REJECTED", "REFUSED", "TIMEOUT"): req.status = REQUEST_STATUS_REJECTED req.reject_reason = "Rejected in Feishu approval." elif feishu_status in ("CANCELED", "CANCELLED", "DELETED"): req.status = REQUEST_STATUS_CANCELLED else: req.decided_at = None # genuinely unknown status — leave pending return async def _provision(self, req: IdentityConsumerRequest) -> None: """Materialise the live consumer + notify the applicant with a one-time token-issue link. Re-validates slug + owner at this authoritative moment; a conflict marks the request FAILED with a reason (no half-built consumer).""" owner = await self.db.get(IdentityPerson, req.owner_person_id) if owner is None or owner.employment_status != "active": req.status = REQUEST_STATUS_FAILED req.reject_reason = ( "Owner is no longer an active employee at approval time; " "re-apply with a current owner." ) return taken = (await self.db.execute( select(IdentityConsumer.id).where(IdentityConsumer.slug == req.slug) )).scalar_one_or_none() if taken is not None: req.status = REQUEST_STATUS_FAILED req.reject_reason = ( f"Slug '{req.slug}' was registered by another consumer before " "approval; re-apply with a different slug." ) return consumer = IdentityConsumer( name=req.name, slug=req.slug, description=req.description, owner_person_id=req.owner_person_id, scopes=list(req.scopes or []), granted_fields=list(req.granted_fields or []), filter_config=dict(req.filter_config or {}), rate_limit_per_minute=req.rate_limit_per_minute, created_by=req.requester_person_id, ) self.db.add(consumer) await self.db.flush() req.consumer_id = consumer.id req.status = REQUEST_STATUS_APPROVED await self.audit.log( trace_id=f"consumer_request_provision:{req.id}", actor_type="integration", actor_user_id=None, action="identity_consumer.create", status="success", resource_type="identity_consumer", resource_id=str(consumer.id), detail={"slug": consumer.slug, "via": "self_service_request", "request_id": str(req.id)}, ) await self._notify_issue_link(req, consumer) async def _notify_issue_link( self, req: IdentityConsumerRequest, consumer: IdentityConsumer, ) -> None: try: from app.integrations.service import get_platform_setting # noqa: PLC0415 from app.notifications.service import ( # noqa: PLC0415 CardAction, LocalizedText, NotificationEnvelope, NotificationService, ) requester = await self.db.get(IdentityPerson, req.requester_person_id) if requester is None: return token = create_action_token( purpose=ACTION_TOKEN_PURPOSE_ISSUE, subject_id=str(req.requester_person_id), resource_id=str(consumer.id), ) portal = str(await get_platform_setting( "app.frontend_base_url", default="", session=self.db, ) or "").rstrip("/") issue_url = ( f"{portal}/api/v1/open/consumer-token/issue?token={token}" if portal else None ) actions = ( [CardAction( label=LocalizedText( zh="领取服务令牌", en="Get service token", ja="サービストークンを取得", ), url=issue_url, style="primary", )] if issue_url else [] ) await NotificationService(self.db).dispatch(NotificationEnvelope( recipient=requester, title=f"身份消费方「{consumer.name}」已获批", body_text=( f"你申请接入的「{consumer.name}」({consumer.slug}) 已通过审批。" "点击领取一次性服务令牌(仅展示一次,请妥善保存)。" "令牌领取链接 14 天内有效,过期请联系 IT 管理员。" ), channels=["feishu", "email"], url="/my/requests", related_resource_type="identity_consumer", related_resource_id=consumer.id, card_actions=actions, )) except Exception: # noqa: BLE001 — notification must not block provisioning logger.exception("consumer-request issue-link notification failed") # ── Helpers ───────────────────────────────────────────────── async def _get(self, request_id: uuid.UUID) -> IdentityConsumerRequest: row = await self.db.get(IdentityConsumerRequest, request_id) if row is None or row.deleted_at is not None: raise NotFoundError("IdentityConsumerRequest", str(request_id)) return row def _parse_widgets(raw_form: Any) -> list[dict[str, Any]]: if isinstance(raw_form, str): return json.loads(raw_form or "[]") if isinstance(raw_form, list): return raw_form return [] def _render_scope_summary(scopes: list[str], granted_fields: list[str]) -> str: parts = [", ".join(sorted(scopes))] if granted_fields: parts.append(f"字段({len(granted_fields)}): {', '.join(granted_fields)}") return "\n".join(parts) def _render_filter_summary(filter_config: dict) -> str: if not filter_config: return "全体在职员工(默认)" bits: list[str] = [] if filter_config.get("departments"): n = len(filter_config["departments"]) sub = "(含下属部门)" if filter_config.get("include_subtree") else "" bits.append(f"部门 {n} 个{sub}") if filter_config.get("employment_statuses"): bits.append("在职状态: " + ", ".join(filter_config["employment_statuses"])) if filter_config.get("countries"): bits.append("国家: " + ", ".join(filter_config["countries"])) return "；".join(bits) if bits else "全体在职员工(默认)" def _render_validity(days: int | None) -> str: if days is None: return "Never" return f"{days}d" __all__ = ["ACTION_TOKEN_PURPOSE_ISSUE", "ConsumerRequestService"]