"""SSO grant request submission, callback, withdraw, and leaver handover. Lifecycle: submit() → Feishu approval (instance_code captured) ↓ Feishu callback "approved" handle_feishu_callback() → status=approved, granted_at=now ↓ requester offboarding starts pending_handovers_for_person() → admin hands over to a successor handover() → person_id rewritten, original archived in ``handover_from_person_id`` The form has four fields the Feishu template asks for verbatim: SSO (single choice: Google / Lark / Authing) System (text) System Owner (text — keeping it free-text so we don't need a Feishu contactList lookup the same way every other owner-picker widget in this codebase does) Reason (textarea) The submitter (``requester``) is the person *getting* the access. "System Owner" is whoever in our org is responsible for the system itself (often a different person), captured for audit / handover. """ from __future__ import annotations import hashlib 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, NotFoundError from app.core.grant_eligibility import is_blocked_from_new_grant from app.core.txn import best_effort from app.domains.sso.models import ( SSO_GRANT_STATUS_APPROVED, SSO_GRANT_STATUS_CANCELLED, SSO_GRANT_STATUS_PENDING, SSO_GRANT_STATUS_REJECTED, SSO_GRANT_STATUS_REVOKED, SsoGrant, ) 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__) _FEISHU_FLOW_KEY = "sso_grant_request" _INTENT_RECOVERABLE_STATES = ( "prepared", "submitting", "unknown", "provider_created", "compensation_pending", ) # The Feishu template's single-choice widget for SSO platform has a # fixed option list. Helix surfaces the same dropdown so we never send # a value the Feishu widget would reject. Keeping this mirrored # server-side so the validator can refuse a tampered request before it # gets to Feishu. # # Authing was deprecated 2026-05-07 — the company is consolidating on # Google + Lark and Helix should not surface a third choice that the # Feishu form no longer offers. Existing ``sso_grants`` rows with # ``sso_provider='Authing'`` are preserved (column is free-text) so # audit history and leaver flows still load; the validator below # blocks NEW Authing requests with a clear error pointing admins to # the deprecation note. SSO_PROVIDERS = ("Google", "Lark") SSO_PROVIDER_GOOGLE = "Google" # Widget keyword fragments — substring match, case-insensitive — that # map a Feishu form widget's display name to the structured field. Key # iteration order matters; the first hit wins. Cribbed from the same # pattern in ``cmdb/dns_request_service.py``. # # SAML-metadata keys (acs_url / entity_id / x509_cert / # saml_metadata_xml) added 2026-05-07 for the Google SAML-app # auto-create flow — the Feishu template grew four new widgets the # same day. Each tuple lists the exact field labels admins typed in # the Feishu builder PLUS fuzzy aliases so a label rename ("ACS URL" # → "ACS 地址") doesn't silently fall back to the reason field. # NOTE ordering matters: ``_build_sso_form_payload`` iterates this dict # and takes the first field_key whose keyword substring-matches the # widget name (first-hit-wins + break). ``system_owner`` MUST precede # ``system_name`` — a widget labelled "System Owner" / "系统负责人" # contains the substring "system" / "系统", so if ``system_name`` came # first the owner widget would be filled with the *system name* value # and the owner would never render on the approver's card. _SSO_FIELD_KEYWORDS: dict[str, tuple[str, ...]] = { "sso_provider": ("sso", "platform", "平台", "登录"), "system_owner": ("owner", "负责人", "归属人"), "system_name": ("system", "系统", "name"), "reason": ("reason", "原因", "理由", "用途", "why"), "acs_url": ("acs", "断言消费", "assertion consumer"), "entity_id": ("entity", "实体", "sp identifier", "audience"), "x509_cert": ("x.509", "x509", "证书", "certificate", "cert"), "saml_metadata_xml": ("metadata", "元数据", "saml xml"), } # attachmentV2 stays in the skip list for OUTBOUND form rendering # (we don't send the user's metadata XML BACK to Feishu — they # uploaded it; Feishu owns the file) but the INBOUND callback parses # attachment widgets via a separate handler. See # :func:`extract_attachment_urls_from_form` in approval/feishu_client.py. _SSO_SKIP_WIDGET_TYPES = frozenset( {"text", "section", "image", "attachmentV2"}, ) def _match_widget(widget_name: str, field_key: str) -> bool: name_lower = (widget_name or "").lower() return any( kw.lower() in name_lower for kw in _SSO_FIELD_KEYWORDS.get(field_key, ()) ) def _resolve_single_choice_value( widget: dict[str, Any], display_text: str, ) -> str | None: """For ``radioV2`` / single-choice widgets Feishu expects the option's ``value`` field, not its ``text``. Look up the matching option by case-insensitive text comparison and return its ``value``; ``None`` if no match (caller falls back to the literal text and Feishu will reject — admin needs to align option list).""" options = widget.get("option") or widget.get("options") or [] needle = display_text.strip().lower() for opt in options: if not isinstance(opt, dict): continue if str(opt.get("text", "")).strip().lower() == needle: return str(opt.get("value", "")) return None def _build_sso_form_payload( widgets: list[dict[str, Any]], values: dict[str, Any], ) -> str: # Honour Feishu ``display_condition`` — hidden widgets skipped. 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]] = [] # Which ``values`` keys actually made it onto the card. Used only for # the unrendered-field warning below. rendered_keys: set[str] = set() for widget in widgets: wtype = widget.get("type", "input") if wtype in _SSO_SKIP_WIDGET_TYPES: continue wid = widget.get("id") if not wid: continue if not visibility.get(wid, True): continue wname = widget.get("name") or "" matched_value: Any = None matched_key: str | None = None for field_key in _SSO_FIELD_KEYWORDS: if _match_widget(wname, field_key) and field_key in values: matched_value = values[field_key] matched_key = field_key break if matched_value is None and widget.get("required"): # Required free-text widgets we couldn't match fall back # to the reason so the approver doesn't see an empty card. if wtype in ("textarea", "input"): matched_value = values.get("reason", "") matched_key = "reason" if matched_value in (None, "", [], {}): continue text = ( str(matched_value) if not isinstance(matched_value, (dict, list)) else json.dumps(matched_value, ensure_ascii=False) ) # Single-choice widgets store the option's value, not its # display text. Resolve here so Helix surfaces a friendly # dropdown ("Google" / "Lark" / "Authing") and Feishu still # gets the internal ``value`` it expects. if wtype in ("radioV2", "radio"): resolved = _resolve_single_choice_value(widget, text) if resolved: text = resolved serialized.append({"id": wid, "type": wtype, "value": text}) if matched_key is not None: rendered_keys.add(matched_key) # This loop walks the TEMPLATE, so any value with no matching widget # is silently dropped — the approver then decides on a card missing # exactly the SAML detail they were meant to weigh, with nothing in # the logs to say so. The data still reaches our DB, so this is not # fatal, but it must not be invisible: a widget rename or a template # edit on the Feishu side is the likeliest cause and only shows up # here. Logged at WARNING rather than raised because a degraded card # beats refusing a request the requester filled in correctly. unrendered = sorted(set(values) - rendered_keys) if unrendered: logger.warning( "sso form payload: %d value(s) had no matching Feishu widget " "and were NOT rendered on the approval card: %s. Check the " "template's widget names against _SSO_FIELD_KEYWORDS.", len(unrendered), ", ".join(unrendered), ) return json.dumps(serialized, ensure_ascii=False) def _parse_template_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 [] class SsoGrantService: """Submit + cancel + Feishu-callback-execute + handover for the sso_grants domain. One instance per request. """ # Outcome of the most recent ``cancel()`` call's best-effort Feishu # side-cancel. Routers / MCP tools read it after ``cancel()`` to # surface ``feishu_cancel_synced`` to the requester. last_feishu_cancel_synced: bool = True 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, sso_provider: str, system_name: str, system_owner_person_id: str, reason: str, trace_id: str, acs_url: str | None = None, entity_id: str | None = None, x509_cert: str | None = None, saml_metadata_xml: str | None = None, ) -> SsoGrant: # Serialize concurrent submits by the SAME requester so a # double-click / MCP-agent retry cannot both pass the # ``existing_pending`` check and each create a Feishu approval # instance + a pending ``SsoGrant`` row (the in-method check is # not atomic on its own, and ``SsoGrant`` has no backing unique # constraint). Mirrors the canonical apikey / domain_registration # / announcement_request submit lock. from app.core.distributed_lock import ( # noqa: PLC0415 LockBusy, distributed_lock, ) lock_key = f"sso_submit:{requester.id}" try: async with distributed_lock(lock_key, ttl=120): return await self._submit_locked( requester=requester, sso_provider=sso_provider, system_name=system_name, system_owner_person_id=system_owner_person_id, reason=reason, trace_id=trace_id, acs_url=acs_url, entity_id=entity_id, x509_cert=x509_cert, saml_metadata_xml=saml_metadata_xml, ) except LockBusy as e: raise AppException( code="SSO_GRANT_ALREADY_PENDING", message=( "A submission for this request is already in progress; " "please wait for it to finish." ), status_code=409, ) from e async def _submit_locked( self, *, requester: IdentityPerson, sso_provider: str, system_name: str, system_owner_person_id: str, reason: str, trace_id: str, # SAML metadata (Google flow only) — see model docstring. # Either ``saml_metadata_xml`` (one file Helix parses) OR all # three of ``acs_url`` + ``entity_id`` + ``x509_cert`` is # required when ``sso_provider == 'Google'``. For Lark these # are silently ignored. acs_url: str | None = None, entity_id: str | None = None, x509_cert: str | None = None, saml_metadata_xml: str | None = None, ) -> SsoGrant: # Round 2b audit fix-up 2026-05-04: service-layer leaver gate # to mirror the router-level dependency for MCP-direct callers. # Also blocks hris_record_state='purged' (2026-07-12). if is_blocked_from_new_grant(requester): raise AppException( code="SSO_GRANT_REQUESTER_NOT_ACTIVE", message=( "Your account is in offboarding; new SSO grant " "requests are not accepted." ), status_code=403, ) if sso_provider not in SSO_PROVIDERS: raise AppException( code="SSO_PROVIDER_INVALID", message=( f"sso_provider must be one of {list(SSO_PROVIDERS)}; " f"got {sso_provider!r}." ), status_code=400, ) clean_system = (system_name or "").strip() if not clean_system: raise AppException( code="SSO_SYSTEM_REQUIRED", message="Please name the system you need access to.", status_code=400, ) clean_reason = (reason or "").strip() if not clean_reason: raise AppException( code="SSO_REASON_REQUIRED", message="Please explain why you need this access.", status_code=400, ) # SAML metadata validation — Google flow only. Validate AND # canonicalise BEFORE the Feishu round-trip so a malformed # request fails fast (no orphan Feishu instance to clean up). clean_acs = (acs_url or "").strip() or None clean_entity = (entity_id or "").strip() or None clean_cert = (x509_cert or "").strip() or None clean_xml = (saml_metadata_xml or "").strip() or None if sso_provider == SSO_PROVIDER_GOOGLE: # Accept either: (a) the metadata XML alone (Helix parses # ACS / Entity / cert at execute time), OR (b) ACS URL + # Entity ID. Mixed input is OK — the parser back-fills empty # discrete fields from the XML. # # ``x509_cert`` is OPTIONAL (2026-08-18). It used to be a # third mandatory field, which asked requesters for something # nothing downstream consumes: in this flow GOOGLE is the # IdP, so Google mints its own signing certificate, and the # "Add custom SAML app" screen IT fills in has no field for # the SP's cert unless encrypted assertions are turned on. # Requiring it turned away requests that had everything # actually needed. Still accepted and stored when supplied — # metadata XML carries one, and it matters if the SP later # wants encrypted assertions. has_xml = bool(clean_xml) has_pair = bool(clean_acs and clean_entity) if not (has_xml or has_pair): raise AppException( code="SSO_SAML_METADATA_REQUIRED", message=( "Google SSO requests need either a SAML " "metadata XML upload OR both of ACS URL and " "Entity ID." ), status_code=400, ) # Length sanity caps so the Feishu webhook can't smuggle # a multi-MB blob past the validator (existing field limits # in the model are 1024 chars / TEXT, so trim explicitly). if clean_acs and len(clean_acs) > 1024: raise AppException( code="SSO_ACS_URL_TOO_LONG", message="ACS URL must be ≤ 1024 characters.", status_code=400, ) if clean_entity and len(clean_entity) > 1024: raise AppException( code="SSO_ENTITY_ID_TOO_LONG", message="Entity ID must be ≤ 1024 characters.", status_code=400, ) if clean_cert and len(clean_cert) > 20_000: raise AppException( code="SSO_X509_CERT_TOO_LONG", message="X.509 certificate must be ≤ 20000 characters.", status_code=400, ) if clean_xml and len(clean_xml) > 1_000_000: # 1 MB hard cap — typical SAML metadata is < 10 KB; a # multi-MB blob is either an attached PDF (wrong file # type) or an attempted DoS via giant payload. raise AppException( code="SSO_SAML_XML_TOO_LARGE", message=( "SAML metadata XML must be ≤ 1 MB; ours is " "typically a few KB. Please re-export from " "your IdP and try again." ), status_code=400, ) else: # Non-Google providers don't carry SAML metadata. Drop any # values the caller accidentally supplied so we don't # persist misleading data. clean_acs = clean_entity = clean_cert = clean_xml = None # Pre-launch hardening 2026-05-04: reject duplicate pending # SSO requests for the SAME (provider, system) pair from the # same requester. Lets the user request access to multiple # systems in parallel, but blocks the double-click case where # the same Feishu approval would be created twice. existing_pending = await self.db.scalar( select(SsoGrant.id).where( SsoGrant.person_id == str(requester.id), SsoGrant.sso_provider == sso_provider, SsoGrant.system_name == clean_system, SsoGrant.status == SSO_GRANT_STATUS_PENDING, ).limit(1).with_for_update() ) if existing_pending: raise AppException( code="SSO_GRANT_ALREADY_PENDING", message=( "You already have a pending SSO request for this " "system. Cancel it first or wait for it to be processed." ), status_code=409, ) # Resolve the picked owner person. Frontend always sends a # ``system_owner_person_id`` (defaulting to the requester # themself). Label is derived from the row so admin reviewers # see a consistent name in the Feishu card and the audit log # rather than whatever free-text the requester typed. try: owner_uuid = uuid.UUID(system_owner_person_id) except (TypeError, ValueError) as e: raise AppException( code="SSO_OWNER_INVALID", message="Invalid system_owner_person_id.", status_code=400, ) from e owner_person = await self.db.get(IdentityPerson, owner_uuid) if owner_person is None: raise NotFoundError("IdentityPerson", system_owner_person_id) clean_owner = ( owner_person.display_name or owner_person.work_email or str(owner_person.id) ) # Resolve the Feishu approval mapping. Admins wire the code # under /admin/integrations//flows; we read it back on # every submit so a code rotation lands without a redeploy. flow = await get_approval_flow(_FEISHU_FLOW_KEY, session=self.db) approval_code = flow.external_code client = await self._client() definition = await client.get_approval_definition(approval_code) widgets = _parse_template_widgets(definition.get("form", "[]")) field_values: dict[str, Any] = { "sso_provider": sso_provider, "system_name": clean_system, "system_owner": clean_owner, "reason": clean_reason, } # Only forward SAML metadata when present — Feishu widgets # for these fields are non-required, so empty values would # render as blank lines on the approval card. Skipping them # keeps the card tidy for Lark requests (which never have # these) and Google requests that supplied only the XML # (the discrete fields stay empty until parse-on-execute). if clean_acs: field_values["acs_url"] = clean_acs if clean_entity: field_values["entity_id"] = clean_entity if clean_cert: field_values["x509_cert"] = ( f"[已附 X.509 certificate · {len(clean_cert)} 字符]" ) if clean_xml: # Don't dump the entire metadata XML into a Feishu text # widget — admins reading the approval card don't need # 8 KB of base64 cert. Send a placeholder describing what # was uploaded; the full content is in our DB and parsed # at execute time. field_values["saml_metadata_xml"] = ( f"[已附 SAML metadata XML · {len(clean_xml)} 字符]" ) form_payload = _build_sso_form_payload(widgets, field_values) # The Feishu instance is owned by the requester (the one # getting access). Owner attribution is in the form body, not # the instance owner. feishu_user_id = await client.resolve_user_id_by_email( requester.work_email, ) grant_id = str(uuid.uuid4()) approval_request_id = str(uuid.uuid4()) recovery_key = submission_fingerprint({ "sso_provider": sso_provider, "system_name": clean_system, }) semantic_payload = { "sso_provider": sso_provider, "system_name": clean_system, "system_owner_person_id": str(owner_person.id), "reason": clean_reason, "acs_url": clean_acs, "entity_id": clean_entity, # Keep large recovery material out of the fingerprint copy. The # full values live exactly once in ``finalization`` until the # domain row is durable, then are compacted out of the intent. "x509_cert_sha256": self._text_sha256(clean_cert), "saml_metadata_xml_sha256": self._text_sha256(clean_xml), "x509_cert_length": len(clean_cert or ""), "saml_metadata_xml_length": len(clean_xml or ""), } intent = await submit_with_durable_intent( self.db, client=client, scope=_FEISHU_FLOW_KEY, applicant_id=requester.id, recovery_key=recovery_key, semantic_payload=semantic_payload, provider_payload={ "approval_code": approval_code, "identity_type": "user_id", "identity_value": feishu_user_id, "form": form_payload, }, finalization_payload={ "grant_id": grant_id, "approval_request_id": approval_request_id, "sso_provider": sso_provider, "system_name": clean_system, "system_owner_label": clean_owner, "system_owner_person_id": str(owner_person.id), "reason": clean_reason, "acs_url": clean_acs, "entity_id": clean_entity, "x509_cert": clean_cert, "saml_metadata_xml": clean_xml, "approval_code": approval_code, }, trace_id=trace_id, ) grant = await self._finalize_submission_intent( intent.id, recovered=( intent.attempt_count > 1 or str((intent.payload or {}).get("trace_id") or "") != trace_id ), ) if grant.status == SSO_GRANT_STATUS_CANCELLED: raise AppException( code="SSO_GRANT_REQUESTER_NOT_ACTIVE", message=( "Your account is in offboarding; new SSO grant " "requests are not accepted." ), status_code=403, ) return grant @staticmethod def _text_sha256(value: str | None) -> str | None: if value is None: return None return hashlib.sha256(value.encode("utf-8")).hexdigest() @staticmethod def _compact_intent_payload(intent: ApprovalSubmissionIntent) -> None: """Drop large SAML material once the durable domain row owns it.""" payload = dict(intent.payload or {}) finalization = dict(payload.get("finalization") or {}) finalization.pop("x509_cert", None) finalization.pop("saml_metadata_xml", None) payload["finalization"] = finalization intent.payload = payload async def _finalize_submission_intent( self, intent_id: uuid.UUID, *, recovered: bool, ) -> SsoGrant: """Materialise SSO correlation or a leaver-safe cancellation.""" 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 != _FEISHU_FLOW_KEY: raise RuntimeError("SSO submission intent disappeared") if intent.state in {"completed", "compensated"}: grant_id = terminal_intent_resource_id( intent, resource_type="sso_grant", ) existing = await self.db.get(SsoGrant, grant_id) if existing is None: raise RuntimeError("terminal SSO intent has no domain row") return existing finalization = dict((intent.payload or {}).get("finalization") or {}) grant_id = str(finalization["grant_id"]) existing = await self.db.get(SsoGrant, grant_id) if intent.state == "compensation_pending": if existing is None: raise RuntimeError("SSO compensation intent has no domain row") 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 SSO intent in {intent.state!r}") # This is a locking current read after the provider-side create. It # deliberately refreshes the request identity-map entry so a concurrent # HR/offboarding transaction cannot be hidden by MySQL/TiDB RR state. requester = await self.db.scalar( select(IdentityPerson) .where(IdentityPerson.id == intent.applicant_id) .with_for_update() .execution_options(populate_existing=True) ) compensate = requester is None or is_blocked_from_new_grant(requester) local_status = ( SSO_GRANT_STATUS_CANCELLED if compensate else SSO_GRANT_STATUS_PENDING ) now = datetime.now(UTC) if existing is not None: raise RuntimeError("provider-created SSO intent already has a row") grant = SsoGrant( id=grant_id, person_id=str(intent.applicant_id), sso_provider=str(finalization["sso_provider"]), system_name=str(finalization["system_name"]), system_owner_label=str(finalization["system_owner_label"]), system_owner_person_id=str(finalization["system_owner_person_id"]), reason=str(finalization["reason"]), status=local_status, feishu_instance_code=intent.external_id, acs_url=finalization.get("acs_url"), entity_id=finalization.get("entity_id"), x509_cert=finalization.get("x509_cert"), saml_metadata_xml=finalization.get("saml_metadata_xml"), created_at=now, updated_at=now, ) self.db.add(grant) 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=_FEISHU_FLOW_KEY, applicant_id=intent.applicant_id, status=local_status, source_payload={ "sso_grant_id": grant_id, "approval_code": finalization["approval_code"], "sso_provider": finalization["sso_provider"], "system_name": finalization["system_name"], "submission_intent_id": str(intent.id), }, ) self.db.add(approval_request) await self.db.flush() grant.approval_request_id = str(approval_request.id) grant.updated_at = now 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 if compensate: self._compact_intent_payload(intent) 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=( "sso.grant_request.submit_compensating_leaver" if compensate else ( "approval.submission_intent.reconciled" if recovered else "sso.grant_request.submit" ) ), status="failure" if compensate else "success", resource_type="sso_grant", resource_id=grant_id, detail={ "intent_id": str(intent.id), "provider_request_uuid": intent.provider_request_uuid, "feishu_instance_code": intent.external_id, "sso_provider": finalization["sso_provider"], "system_name": finalization["system_name"], "recovered": recovered, "compensation_required": compensate, }, ) if not compensate: compact_terminal_intent_payload( intent, resource_type="sso_grant", resource_id=grant_id, ) await self.db.commit() await self.db.refresh(grant) if compensate: await self._compensate_provider_intent(intent_id) await self.db.refresh(grant) return grant async def _compensate_provider_intent(self, intent_id: uuid.UUID) -> bool: """Withdraw a provider-created approval for an ineligible leaver.""" intent = await self.db.get(ApprovalSubmissionIntent, intent_id) if intent is None: raise RuntimeError("SSO compensation intent disappeared") if intent.state == "compensated": return True if intent.state != "compensation_pending" or not intent.external_id: raise RuntimeError(f"cannot compensate SSO intent in {intent.state!r}") stored_provider = dict((intent.payload or {}).get("provider") or {}) grant_id = str( ((intent.payload or {}).get("finalization") or {}).get("grant_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 - cancel remains the recovery action 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 - durable retry + admin alert 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="sso_grant", resource_id=grant_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="SSO 应用开通申请", resource_type="sso_grant_submission_compensation", resource_id=grant_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("SSO compensation intent disappeared") if intent.state == "compensated": return True if intent.state != "compensation_pending": raise RuntimeError("SSO compensation state changed unexpectedly") 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="sso_grant", resource_id=grant_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="sso_grant", resource_id=grant_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]: """Recover provider create/finalize and leaver compensation gaps.""" cutoff = datetime.now(UTC) - timedelta(minutes=min_age_minutes) intent_refs = list((await self.db.execute( select( ApprovalSubmissionIntent.id, ApprovalSubmissionIntent.applicant_id, ) .where( ApprovalSubmissionIntent.scope == _FEISHU_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()) # End the listing snapshot before waiting on the same logical lock as # the HTTP submitter. The lock winner's later commit must be visible # to our lock-internal FOR UPDATE current reads under MySQL/TiDB RR. await self.db.rollback() summary = { "checked": len(intent_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 intent_refs: try: async with distributed_lock( f"sso_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 == SSO_GRANT_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 # ── Cancel (requester withdraws while pending) ────────────── async def cancel( self, *, grant_id: str, actor: IdentityPerson, trace_id: str, ) -> SsoGrant: grant = await self._lock_grant(grant_id) # Round-20 audit (A2F8): allow IT admin to cancel-on-behalf when # the requester has left the company without explicitly # cancelling. Without the admin branch, a leaver's pending SSO # grants linger forever — orphan Feishu instances, stuck leaver # list. Self-cancel is still the common case; admin-on-behalf is # distinguished in audit (``cancelled_by``) so the trail remains # unambiguous. # # Authority gate is ``global_admin`` only (mirrors the SSO # router's ``require_admin`` = GLOBAL_ADMIN_ROLE). It was briefly # ``can_manage_assets``, which leaked SSO-cancel authority to the # region-scoped ``asset_admin`` / ``asset_global_admin`` roles — # an office/laptop admin has no business cancelling another # employee's SSO grant. Fixed 2026-06-19 (self-service authz # review batch). from app.core.rbac import GLOBAL_ADMIN_ROLE, has_role is_admin = has_role(actor.role_names, GLOBAL_ADMIN_ROLE) is_owner = grant.person_id == str(actor.id) if not (is_owner or is_admin): raise AppException( code="SSO_GRANT_NOT_OWNER", message=( "Only the original requester or an IT/asset admin " "can cancel." ), status_code=403, ) if grant.status != SSO_GRANT_STATUS_PENDING: raise AppException( code="SSO_GRANT_INVALID_STATE", message=( f"Cannot cancel a grant in {grant.status!r} state." ), status_code=409, ) # Cancel the parallel Feishu instance too. A failure here # doesn't block — admins fix orphans by hand rather than # letting a Feishu hiccup pin the row pending forever. When the # actor is an admin cancelling on behalf, resolve the Feishu # user_id from the GRANT's owner (not the actor) so the Feishu # cancel call carries the right principal. self.last_feishu_cancel_synced = True if grant.feishu_instance_code: try: client = await self._client() flow = await get_approval_flow( _FEISHU_FLOW_KEY, session=self.db, ) cancel_principal_email = actor.work_email if not is_owner: # Admin-on-behalf: cancel as the original requester. owner = await self.db.get( IdentityPerson, uuid.UUID(grant.person_id), ) if owner is not None and owner.work_email: cancel_principal_email = owner.work_email feishu_user_id = await client.resolve_user_id_by_email( cancel_principal_email, ) await client.cancel_instance( instance_code=grant.feishu_instance_code, user_id=feishu_user_id, approval_code=flow.external_code, ) except Exception as e: # noqa: BLE001 logger.warning( "Feishu cancel failed for sso_grant %s (instance=%s): %s", grant.id, grant.feishu_instance_code, e, ) self.last_feishu_cancel_synced = False await notify_feishu_cancel_desync( self.db, flow_label="SSO 应用开通申请", resource_type="sso_grant", resource_id=str(grant.id), instance_code=grant.feishu_instance_code, error=e, trace_id=trace_id, ) grant.status = SSO_GRANT_STATUS_CANCELLED grant.updated_at = datetime.now(UTC) await self._mirror_approval_request_status(grant, "cancelled") await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor.id, action=( "sso.grant_request.cancel" if is_owner else "sso.grant_request.admin_cancel" ), status="success", resource_type="sso_grant", resource_id=grant.id, detail={ "cancelled_by_admin": (not is_owner), "grant_owner_person_id": grant.person_id, }, ) return grant # ── Feishu callback ───────────────────────────────────────── async def handle_feishu_callback( self, *, approval_request: ApprovalRequest, instance_detail: dict[str, Any], ) -> None: source = approval_request.source_payload or {} grant_id = source.get("sso_grant_id") if not grant_id: logger.warning( "sso callback missing sso_grant_id: ar=%s", approval_request.id, ) return feishu_status = (instance_detail.get("status") or "").upper() try: grant = await self._lock_grant(grant_id) except NotFoundError: logger.warning( "sso callback for missing grant %s — ignoring", grant_id, ) return if grant.status != SSO_GRANT_STATUS_PENDING: return now = datetime.now(UTC) if feishu_status in ("APPROVED", "PASS"): # Pre-launch hardening 2026-05-06: Feishu approvals can sit # pending for days. Re-validate the requester is still an # active employee before flipping to APPROVED + setting # ``granted_at``; otherwise we soft-launch privileged # access to a former employee whose offboarding completed # while the approval was queued. Mirror of the leaver-recheck # in approval/service._activate_software_seat. # ``grant.person_id`` is a ``String(36)`` column but # ``IdentityPerson.id`` is a ``Uuid`` PK — passing the raw str # to ``db.get`` makes the Uuid bind processor call ``.hex`` on a # str and raise ``StatementError`` BEFORE the grant is flipped to # APPROVED, so no Feishu-approved SSO grant ever reached # ``approved`` in production. Coerce to UUID like every other # call site in this domain (service.py:939, app_router.py:130…). requester = await self.db.get( IdentityPerson, uuid.UUID(grant.person_id), ) if is_blocked_from_new_grant(requester): grant.status = SSO_GRANT_STATUS_REJECTED grant.updated_at = now requester_status = ( requester.employment_status if requester else None ) await self.audit.log( trace_id=f"feishu_callback:{approval_request.external_id}", actor_type="integration", actor_user_id=None, action="sso.grant_request.callback_refused_leaver", status="failure", resource_type="sso_grant", resource_id=grant.id, detail={ "instance_code": approval_request.external_id, "feishu_status": feishu_status, "requester_status": requester_status, "reason": "requester_offboarded_during_pending", }, ) return grant.status = SSO_GRANT_STATUS_APPROVED grant.granted_at = now elif feishu_status in ("REJECTED", "REFUSED"): grant.status = SSO_GRANT_STATUS_REJECTED elif feishu_status in ("CANCELED", "CANCELLED", "DELETED"): grant.status = SSO_GRANT_STATUS_CANCELLED else: return grant.updated_at = now await self.audit.log( trace_id=f"feishu_callback:{approval_request.external_id}", actor_type="integration", actor_user_id=None, action=f"sso.grant_request.{grant.status}", status="success", resource_type="sso_grant", resource_id=grant.id, detail={ "instance_code": approval_request.external_id, "feishu_status": feishu_status, }, ) # ── Leaver handover ───────────────────────────────────────── async def list_pending_handovers_for_person( self, person_id: str, ) -> list[SsoGrant]: """Approved-and-active SSO grants where ``person_id`` is the access holder. Used by the offboarding flow + admin panel to decide what needs handover before the leave date.""" rows = ( await self.db.execute( select(SsoGrant).where( SsoGrant.person_id == person_id, SsoGrant.status == SSO_GRANT_STATUS_APPROVED, ).order_by(SsoGrant.granted_at.asc()) ) ).scalars().all() return list(rows) async def handover( self, *, grant_id: str, successor_person_id: str, actor: IdentityPerson, trace_id: str, ) -> SsoGrant: grant = await self._lock_grant(grant_id) if grant.status != SSO_GRANT_STATUS_APPROVED: raise AppException( code="SSO_GRANT_NOT_HANDOVERABLE", message=( f"Cannot handover a grant in {grant.status!r} state — " "must be 'approved'." ), status_code=409, ) successor = await self.db.get( IdentityPerson, uuid.UUID(successor_person_id), ) if successor is None: raise NotFoundError("IdentityPerson", successor_person_id) if successor.id == uuid.UUID(grant.person_id): raise AppException( code="SSO_GRANT_SAME_OWNER", message="Successor must differ from the current holder.", status_code=400, ) # Pre-launch P0 fix (Round 45, CR-303): refuse handover to an # inactive successor. Without this check, an admin can hand a # live SSO grant (SAML access, internal tool access) to a # leaver / leaver_pending / soft-deleted person — the access # then persists past offboarding under an owner who can't # exercise it. Mirrors the recipient gate seat-revoke applies. # Centralized gate covers deleted_at + left/leaver_pending AND the # HRIS-purged tombstone the hand-rolled check missed — never hand a # live SSO grant to a successor who can no longer log in. successor_status = (successor.employment_status or "").lower() if is_blocked_from_new_grant(successor): raise AppException( code="SSO_GRANT_SUCCESSOR_INACTIVE", message=( f"Cannot hand over to " f"{successor.display_name or successor.work_email} — " f"they are {successor_status or 'deleted'}." ), status_code=409, ) previous_person_id = grant.person_id grant.handover_from_person_id = previous_person_id grant.person_id = str(successor.id) grant.handover_at = datetime.now(UTC) grant.handover_actor_id = str(actor.id) grant.updated_at = grant.handover_at await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor.id, action="sso.grant.handover", status="success", resource_type="sso_grant", resource_id=grant.id, detail={ "from_person_id": previous_person_id, "to_person_id": grant.person_id, }, ) return grant async def revoke( self, *, grant_id: str, actor: IdentityPerson, reason: str | None, trace_id: str, notify_admins: bool = True, task_source: str | None = None, ) -> SsoGrant: """IT admin path — revoke an approved grant outright. Used when the access holder leaves and IT chooses NOT to handover (i.e., the SSO seat is no longer needed). ``notify_admins`` (default True) controls whether IT admins (global_admin / it_admin) receive the per-grant "[Action Required] Disable SSO access for X" card. Set to False from bulk callers (notably ``ProcessAdvanceService._force_transfer_all`` which calls revoke once per SSO grant per leaver T+0) so a leaver with N grants and M admins does not spam N×M cards. The bulk caller is responsible for emitting ONE coalesced summary card per leaver. The system-owner card is still sent (it is per-system, not per-admin, so fan-out is naturally bounded). """ grant = await self._lock_grant(grant_id) if grant.status != SSO_GRANT_STATUS_APPROVED: raise AppException( code="SSO_GRANT_NOT_REVOKABLE", message=( f"Cannot revoke a grant in {grant.status!r} state — " "must be 'approved'." ), status_code=409, ) now = datetime.now(UTC) grant.status = SSO_GRANT_STATUS_REVOKED grant.revoked_at = now grant.revoke_reason = (reason or "").strip()[:255] or None grant.updated_at = now await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor.id, action="sso.grant.revoke", status="success", resource_type="sso_grant", resource_id=grant.id, detail={"reason": grant.revoke_reason}, ) # ── Removal lifecycle (migration 603) ──────────────────────── # Revoking in Helix is the DECISION. The IdP-side removal is a human # action in a vendor console, so it gets an owner, a clock and an # evidence requirement: a ``manual_notify`` provisioning task for the # system owner (IT when there is no active owner). The daily # ``sso.removal_reconcile`` re-opens a cancelled task, reminds on a # widening ladder, opens a compliance incident at day 14 and — for # Google-IdP grants — verifies against the SAML login audit. # # Deliberately NOT best-effort: a revoke that cannot open its task # would recreate the pre-603 shape (a revoked row nobody owns), so a # failure here must roll the revoke back and surface. from app.domains.provisioning.models import ( # noqa: PLC0415 SOURCE_MANUAL, ) from app.domains.sso.removal import SsoRemovalService # noqa: PLC0415 removal = SsoRemovalService(self.db) has_owner = await removal._active_owner_id(grant) is not None # Bulk callers (offboarding T+0, ``notify_admins=False``) emit ONE # coalesced summary per IT admin per leaver. When the grant has no # active owner the task's fallback recipients would be those same IT # admins — so the task is opened as ``notified`` WITHOUT a DM and the # summary card carries the queue link instead (Round-59 R59F-C: no # N×M cards). With an owner the per-system DM is the desired routing. task = await removal.open_task( grant, source=task_source or SOURCE_MANUAL, actor_user_id=actor.id, trace_id=trace_id, reason=grant.revoke_reason, notify=(notify_admins or has_owner), ) # IT admins on the single-revoke path (router / MCP) still get a # heads-up card — the task DM went to the system owner, and IT is the # backstop who reads the ledger. When there is NO owner the task DM # already reached IT, so a second card would be a duplicate. if notify_admins and has_owner: async with best_effort( self.db, label="sso.grant.revoke admin heads-up", log=logger, ) as notify_step: from app.auth.models import AuthRole, AuthUserRole # noqa: PLC0415 from app.notifications.service import ( # noqa: PLC0415 NotificationEnvelope, NotificationService, ) notify = NotificationService(self.db) person_obj = None try: person_obj = await self.db.get( IdentityPerson, uuid.UUID(str(grant.person_id)), ) except (ValueError, TypeError): person_obj = None person_label = ( (person_obj.display_name or person_obj.work_email) if person_obj else str(grant.person_id) ) admin_ids = list( ( await self.db.execute( select(AuthUserRole.user_id) .join( AuthRole, AuthRole.id == AuthUserRole.role_id, ) .where( AuthRole.name.in_(("global_admin", "it_admin")) ) ) ).scalars().all() ) for admin_id in admin_ids: if task.assignee_id is not None and admin_id == task.assignee_id: continue admin = await self.db.get(IdentityPerson, admin_id) if ( admin is None or admin.deleted_at is not None or (admin.employment_status or "").lower() != "active" or not admin.work_email ): continue # Nested SAVEPOINT + ``one_admin.reason()``. ``dispatch`` # self-isolates and the outer ``notify_step`` already # protects the caller, so this is defence in depth plus # the real fix: ``str(exc)`` rendered an ``AppException`` # as an empty sentinel. async with best_effort( self.db, label=f"sso revoke heads-up to admin {admin_id}", log=logger, log_level=logging.DEBUG, ) as one_admin: await notify.dispatch(NotificationEnvelope( recipient=admin, title=( f"SSO grant revoked — removal task opened " f"({grant.system_name})" ), body_text=( f"Helix revoked the SSO grant for {person_label} " f"on {grant.system_name}. A removal task was " "opened for the system owner; the daily " "reconcile will verify it against the IdP " "login audit and escalate if it stays open." ), channels=["feishu", "email"], template_key="sso.revoked_admin_headsup", context={ "system_name": grant.system_name, "holder": person_label, "task_id": str(task.id), "reason": grant.revoke_reason or "", }, url="/admin/sso-apps?tab=grants&removal=open", related_resource_type="sso_grant", related_resource_id=grant.id, )) if not one_admin.ok: logger.warning( "sso.grant.revoke: admin notify failed: %s", one_admin.reason(), ) if not notify_step.ok: logger.warning( "sso.grant.revoke: admin heads-up dispatch failed: %s", notify_step.reason(), ) return grant # ── Helpers ───────────────────────────────────────────────── async def _lock_grant(self, grant_id: str) -> SsoGrant: # SELECT ... FOR UPDATE so a concurrent cancel + Feishu approval # webhook can't both observe ``pending`` and race into a # split-brain (one writes ``cancelled``, the other ``approved``). # The previous body used ``db.get`` which is a plain SELECT — # the helper's name implied a lock that wasn't there. row = ( await self.db.execute( select(SsoGrant).where(SsoGrant.id == grant_id).with_for_update() ) ).scalar_one_or_none() if row is None: raise NotFoundError("SsoGrant", grant_id) return row async def _mirror_approval_request_status( self, grant: SsoGrant, new_status: str, ) -> None: if not grant.feishu_instance_code: return ar = ( await self.db.execute( select(ApprovalRequest).where( ApprovalRequest.external_id == grant.feishu_instance_code, ApprovalRequest.approval_type == _FEISHU_FLOW_KEY, ) ) ).scalar_one_or_none() if ar is None: return ar.status = new_status ar.updated_at = datetime.now(UTC)