"""Cloud scope request flow — employee asks for a new sub-account / subscription / project, IT admin approves, the driver actually creates it, and a :class:`CmdbCloudScope` row lands with the requester as owner + platform-default budget + alert rules + an owner-equivalent grant. Approval today is **Helix-internal** — IT Admin clicks Approve in the admin UI and the worker calls ``driver.create_scope`` synchronously. The ``feishu_instance_code`` column on the request row is wired so a later PR can route through Feishu approval (the schema_code lives in ``platform_settings.cloud.scope_request_feishu_schema_code``); until then the column stays NULL and the internal-approval path is the only one. We deliberately don't fake a Feishu instance code — empty value in the DB ≠ "Feishu approved", and we don't want a future migration to misinterpret stub data. State machine (see ``models.CmdbCloudScopeRequest`` docstring): pending → feishu_approved → provisioning → provisioned (success) pending → rejected / withdrawn (terminal) provisioning → failed (retryable) """ from __future__ import annotations import hashlib import json import logging import re import uuid from datetime import UTC, datetime, timedelta from decimal import Decimal from typing import Any from sqlalchemy import func, select from sqlalchemy.exc import IntegrityError, InvalidRequestError, OperationalError 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, submit_with_durable_intent, terminal_intent_resource_id, ) from app.audit.service import AuditService from app.core.database import async_session_factory from app.core.exceptions import ( AppException, NotFoundError, format_exception_for_log, ) from app.core.txn import best_effort from app.domains.cmdb.lifecycle_service import ( _build_default_alert_rules, ) from app.domains.cmdb.models import ( SCOPE_REQUEST_STATUS_FAILED, SCOPE_REQUEST_STATUS_FEISHU_APPROVED, SCOPE_REQUEST_STATUS_PENDING, SCOPE_REQUEST_STATUS_PROVISIONED, SCOPE_REQUEST_STATUS_PROVISIONING, SCOPE_REQUEST_STATUS_REJECTED, SCOPE_REQUEST_STATUS_WITHDRAWN, TENANT_LEVEL_SCOPE_TYPES, CmdbCloudScope, CmdbCloudScopeRequest, ) from app.identity.person_models import IdentityPerson from app.integrations.compat import get_feishu_config from app.integrations.driver_base import CloudDriver from app.integrations.models import Integration from app.integrations.service import get_approval_flow from app.notifications.service import NotificationEnvelope, NotificationService logger = logging.getLogger(__name__) # ── Feishu form payload helpers ──────────────────────────────────── # # The cloud_scope_request approval template (configured per-integration # in ``integration_approval_flows``) is owned by the IT-Admin who runs # Feishu — they design widgets, ordering, names. We can't depend on # specific widget IDs, so we substring-match each widget's display name # against keyword fragments to figure out which structured field it # wants. Mirrors the pattern in dns_request_service / ApiKeyService. # # Iteration order matters — the first key whose keyword set hits wins. # Specific keys (cloud_account_id, request_id, type) come BEFORE less # specific ones (name, owner, description) so a widget called # "Cloud account (project) ID" doesn't get grabbed by ``name``. # NOTE on full-width vs half-width punctuation (Chinese 全角 vs 半角): # the production Feishu approval template uses full-width parens — e.g. # the "name" widget is literally `云账号（项目）名称`, not `云账号(项目)名`. # Half-width substrings like "(project) name" / "账号(项目)名" do NOT # match a full-width-parens widget. We list BOTH variants so a template # admin who flips paren width doesn't silently break the form fill — # the bug-class to avoid is "submission goes through, widget falls back # to description, approver sees the description text in the ID column". _CLOUD_SCOPE_FIELD_KEYWORDS: dict[str, tuple[str, ...]] = { "cloud_account_id": ( "(project) id", "（项目）id", "cloud account (project) id", "云账号（项目）id", "云账号(项目)id", "账号 id", "账号id", "项目 id", "项目id", ), "request_id": ( "unique identification", "唯一标识", "唯一编号", "management id", # production template uses this label ), "cloud_account_type": ("类型", "type"), "provider": ( "cloud provider", "provider", # Production templates use both 厂商 and 供应商 — they're # synonyms but share zero characters, so each needs its own # keyword. PR-pre-fix shipped only 云厂商 and silently dropped # the value on every 云供应商 template (verified prod instance # 6721A06B…BEA, 2026-06-02). "云厂商", "厂商", "云供应商", "供应商", ), "owner": ("owner",), "department": ("归属部门", "department", "部门"), "name": ( "(project) name", "（项目）名", "项目名", "账号名", "云账号（项目）名", "云账号(项目)名", ), "description": ( "description", "use of", "用途描述", # NOTE: "说明" / bare "描述" / bare "用途" intentionally NOT here — # they are too generic and would substring-match future widgets # ("说明 N" headers, "用途分类", "备注描述", etc.) into the # description fallback. Production templates use "说明 1" / # "说明 2" as section-header text widgets (already skip-listed # by type), so explicit listing here would only create false- # positive bait. Add narrow compound keywords (e.g. "项目描述") # if a future template needs them. ), "total_budget": ("total budget", "总预算"), "monthly_budget": ( "estimated budget", "monthly budget", "expenses", "月度预算", "预算预估", "预算估算", # NOTE: bare "预算" intentionally NOT here — too generic, would # catch widgets like "预算审批人" or "预算说明". The three # compound Chinese variants cover every production-template # column header we've seen. ), "month": ("月份", " month"), } # Widget types submit-side should skip outright. ``fieldList`` is # handled specially below (one row, recursive child match) — it's # kept off this list because the budget block in the template is a # fieldList we DO want to fill. # # 2026-06-02: ``radioV2`` REMOVED from the skip set. The original # 2026-05-12 rationale was that the "类型" widget is radioV2 and # Helix's ``cloud_account_type`` is free-text (cannot map to an # option key). But a blanket skip also dropped ``provider`` — a # clean enum (aws / gcp / azure / aliyun) that maps 1:1 onto the # template's "云供应商" radio options — leaving the required field # empty on every submission (verified prod instance 6721A06B…BEA). # New behaviour: serialise radioV2 widgets too, but route the value # through ``_resolve_single_choice_value``. Clean enums (provider) # resolve to an option value and submit. Free-text mismatches # (cloud_account_type) fail to resolve → widget is dropped → same # behaviour as the old blanket skip. No regression on the original # bug class. _SCOPE_SKIP_WIDGET_TYPES = frozenset( {"text", "section", "image", "attachmentV2"} ) def _resolve_single_choice_value( widget: dict[str, Any], display_text: str, ) -> str | None: """For ``radioV2`` widgets, Feishu submit expects the option's ``value`` field (its internal key), not the human-readable ``text``. Look up the option whose text matches ``display_text`` case-insensitively and return its value, or ``None`` if nothing matches — caller drops the widget rather than submit garbage. Mirrors the helper in ``sso/service.py`` and ``vpn_application/ service.py``; kept inline because the three call-sites don't share enough surrounding logic to justify a shared module yet. """ 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", "")) # Fallback: caller already passed the option value (e.g. from # an upstream system that knows Feishu keys). Matching on # value too means we don't double-resolve and accidentally # blank the field. if str(opt.get("value", "")).strip().lower() == needle: return str(opt.get("value", "")) return None def _match_scope_widget(widget_name: str, field_key: str) -> bool: name_lower = (widget_name or "").lower() keywords = _CLOUD_SCOPE_FIELD_KEYWORDS.get(field_key, ()) return any(kw.lower() in name_lower for kw in keywords) def _scope_field_for_widget(widget_name: str) -> str | None: """Return the structured-field key that best matches ``widget_name``, or ``None`` if no key matches. Walks ``_CLOUD_SCOPE_FIELD_KEYWORDS`` in insertion order so more specific keys win. """ for field_key in _CLOUD_SCOPE_FIELD_KEYWORDS: if _match_scope_widget(widget_name, field_key): return field_key return None def _serialize_scope_widget( widget: dict[str, Any], values: dict[str, Any], ) -> dict[str, Any] | None: """Build the Feishu instance form entry for a single widget. Returns ``None`` to mean "skip this widget" (unknown / empty / skip-listed type). Handles ``fieldList`` widgets by walking their ``children`` and emitting a single row. """ wtype = widget.get("type", "input") if wtype in _SCOPE_SKIP_WIDGET_TYPES: return None wid = widget.get("id") if not wid: return None wname = widget.get("name") or "" if wtype == "fieldList": # The approval template's monthly-budget block is a fieldList # whose children are (月份, expenses). Helix only carries a # single ``monthly_budget_usd`` value, so we emit exactly one # row — admins who want a per-month plan can still edit the # row inside Feishu before approving. children = widget.get("children") or [] row: list[dict[str, Any]] = [] for child in children: child_entry = _serialize_scope_widget(child, values) if child_entry is not None: row.append(child_entry) if not row: return None return {"id": wid, "type": wtype, "value": [row]} field_key = _scope_field_for_widget(wname) matched_value: Any = None if field_key is not None and field_key in values: matched_value = values[field_key] elif widget.get("required") and wtype in ("textarea", "input"): # Required free-text widgets we couldn't match fall back to the # description so the approver at least sees something coherent. # Intentionally NOT extended to radioV2 — dumping arbitrary # text into a single-choice widget gets silently dropped by # Feishu and adds zero value to the approver. matched_value = values.get("description", "") or values.get("name", "") if matched_value in (None, "", [], {}): return None text = ( str(matched_value) if not isinstance(matched_value, (dict, list)) else json.dumps(matched_value, ensure_ascii=False) ) if wtype == "radioV2": # Resolve the human-readable text to the option's internal # value. If nothing matches (e.g. free-text fields like # cloud_account_type submitted into a fixed-option widget), # drop the widget rather than submit garbage — same outcome # as the pre-2026-06-02 blanket skip for radioV2. resolved = _resolve_single_choice_value(widget, text) if not resolved: return None text = resolved return {"id": wid, "type": wtype, "value": text} def _build_scope_form_payload( widgets: list[dict[str, Any]], values: dict[str, Any], ) -> str: # Honour Feishu ``display_condition`` — widgets whose visibility # gate evaluates false (given the trigger values currently in the # answer set) are skipped entirely: not required-checked, not # serialised to the Feishu payload. Without this, a conditional # required widget would always block submit even on the happy # path. Same evaluator used across every Feishu-backed flow — # see ``app/integrations/feishu_form_visibility.py``. Pass values # under the widget-name → field-key map we already built so name- # based trigger references resolve too. from app.integrations.feishu_form_visibility import ( evaluate_display_conditions, ) # Build a widget-id keyed answer view so the evaluator can compare # trigger answers against ``cond.widget_id``. Each widget is # asked: "given the values that would land in your slot, are # you visible?". For cloud_scope_request the values dict is # field-key keyed (description / monthly_budget / etc.); we # cannot resolve trigger values without first running the # widget→field-key resolver, so we just pass an empty map — the # evaluator fail-opens for triggers it can't resolve, which is # the correct semantics for cloud_scope_request today (no live # template uses display_condition on a widget that's gated by # another applicant-fillable widget). When admins later add such # a condition, extend this answer-builder. visibility = evaluate_display_conditions(widgets, {}, assume_visible_when_answers_unsourced=True) serialized: list[dict[str, Any]] = [] for widget in widgets: wid = widget.get("id") if wid and not visibility.get(wid, True): continue entry = _serialize_scope_widget(widget, values) if entry is not None: serialized.append(entry) return json.dumps(serialized, ensure_ascii=False) def _parse_scope_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 [] _FEISHU_FLOW_KEY = "cloud_scope_request" # ── Form validation ───────────────────────────────────────────────── # Same character set the request modal advertises in its tooltip: # letters / digits / spaces / dash. Keeping the regex aligned with the # UI hint means a name that the form accepted will always pass server # validation — no surprise 400s. # Capped at 50, not 255: this name becomes the cloud-side DisplayName # (Aliyun Resource Directory + AWS Organizations both cap DisplayName at # 50 chars). Letters / digits / spaces / '-' only; at least one # alphanumeric so the alphanumeric-only ``AccountNamePrefix`` the Aliyun # driver derives from it (login-email seed) is non-empty. Keep this in # lockstep with the frontend rule in ``SubmitScopeRequestModal`` so a # name the form accepts never surprises the user with a downstream 400 / # cloud-provider rejection. _NAME_PATTERN = re.compile(r"^[A-Za-z0-9 \-]{1,50}$") def _validate_name(name: str) -> str: name = (name or "").strip() if not name: raise AppException( code="CLOUD_SCOPE_REQUEST_NAME_REQUIRED", message="Please set a meaningful name.", status_code=400, ) if not _NAME_PATTERN.match(name) or not any(c.isalnum() for c in name): raise AppException( code="CLOUD_SCOPE_REQUEST_NAME_INVALID", message=( "Please set a meaningful name: letters, numbers, spaces, " "and '-', at least one letter or number, up to 50 " "characters, e.g., 'region-department-purpose'." ), status_code=400, ) return name def _validate_budget(amount: float | int | Decimal) -> Decimal: try: d = Decimal(str(amount)) except (TypeError, ValueError) as e: raise AppException( code="CLOUD_SCOPE_REQUEST_BUDGET_INVALID", message=f"monthly_budget_usd is not numeric: {amount!r}", status_code=400, ) from e if d <= 0: raise AppException( code="CLOUD_SCOPE_REQUEST_BUDGET_TOO_LOW", message="monthly_budget_usd must be greater than 0", status_code=400, ) if d > Decimal("10000000"): # 10M USD/mo is well above any reasonable single sub-account # spend; if a request comes in this large the form's almost # certainly mistyped (off-by-3 or yen/won pasted). raise AppException( code="CLOUD_SCOPE_REQUEST_BUDGET_TOO_HIGH", message=( "monthly_budget_usd is unreasonably large — confirm " "the value is in USD and not a non-USD currency." ), status_code=400, ) return d.quantize(Decimal("0.01")) # ── Per-provider default scope_type ──────────────────────────────── _DEFAULT_SCOPE_TYPE_BY_PROVIDER: dict[str, str] = { "aws": "aws_account", "gcp": "gcp_project", "azure": "azure_subscription", "aliyun": "aliyun_account", "alicloud_intl": "aliyun_account", "tencent": "standalone_iam", "ucloud": "standalone_iam", } def _default_scope_type(provider: str) -> str: """The canonical ``CmdbCloudScope.scope_type`` for each provider. Used by the manual-provision path when the admin doesn't specify a ``scope_type`` explicitly. Conservative default ``standalone_iam`` for unknown providers so the row at least lands cleanly.""" return _DEFAULT_SCOPE_TYPE_BY_PROVIDER.get(provider, "standalone_iam") # ── AWS owner-alias derivation ───────────────────────────────────── # # AWS Organizations CreateAccount requires a globally-unique Email per # member account (the root user's identity). Reusing the requester's # work email collides on the second sub-account. The platform's answer # is a per-scope Google Workspace user-alias on the requester's # mailbox: ``@pingcap.com`` — the slug is the sanitised request # name itself, NOT prefixed with ``aws-`` (per 2026-05-25 owner call: # names already convey provider intent — e.g. ``aws-pingcap-datalake`` # stays ``aws-pingcap-datalake``, but ``pingcap-foo`` stays # ``pingcap-foo`` and is NOT forced to ``aws-pingcap-foo``). AWS sees # a fresh address, the requester still receives the AWS root # verification mail. # # Domain is hardcoded to ``pingcap.com`` (per 2026-05-25 decision — # no need for per-org abstraction until we expand beyond a single # org). Promote to ``platform_settings`` when multi-domain support # matters. _AWS_OWNER_ALIAS_DOMAIN = "pingcap.com" # Google caps the local part at 64 chars. We cap at the same bound — # request names that slug longer than this raise so the admin renames # rather than letting a silently-truncated alias collide with another # truncated row. _AWS_OWNER_ALIAS_LOCAL_PART_MAX = 64 _AWS_OWNER_ALIAS_SLUG_RE = re.compile(r"[^a-z0-9-]+") _AWS_OWNER_ALIAS_COLLAPSE_RE = re.compile(r"-+") def _derive_aws_owner_alias_local_part(scope_name: str) -> str: """Slugify a scope name into the local-part of an AWS owner alias. Rules: * Lowercase * Map non ``[a-z0-9-]`` (incl. CJK, whitespace, punctuation) to ``-`` * Collapse consecutive dashes * Strip leading / trailing dashes Returns the slug verbatim — caller (:func:`derive_aws_owner_alias`) enforces non-empty + length cap and raises ``ValueError`` for the edge cases. We deliberately do NOT fall back to a synthetic ``unnamed`` placeholder: the alias is the AWS root user email and a fixed placeholder would collide on the second pathological-name request — fast-fail with a clear message instead so the admin picks a better name. """ s = scope_name.strip().lower() s = _AWS_OWNER_ALIAS_SLUG_RE.sub("-", s) s = _AWS_OWNER_ALIAS_COLLAPSE_RE.sub("-", s).strip("-") return s def derive_aws_owner_alias(scope_name: str) -> str: """Public-API name for the alias derivation. Returns the full address ``@pingcap.com`` — the slug is the sanitised request name itself, with NO forced ``aws-`` prefix. Examples: * ``"aws-pingcap-datalake"`` → ``aws-pingcap-datalake@pingcap.com`` * ``"pingcap-foo"`` → ``pingcap-foo@pingcap.com`` * ``"My Project NN!"`` → ``my-project-nn@pingcap.com`` Raises ``ValueError`` when the slug is empty (pathological all-CJK / all-punctuation names) or exceeds Google's 64-char local-part limit — both cases tell the admin to rename the scope to something valid before retrying. A synthetic fallback ("unnamed", truncated-with-suffix, etc.) would silently collide between two pathological-name requests, so we'd rather fail loudly than ship a wrong-on-purpose alias. Exposed at module level so unit tests + future callers (admin preview UI on the Feishu approval card) can preview the alias without invoking the side-effecting ensure path. """ slug = _derive_aws_owner_alias_local_part(scope_name) if not slug: raise ValueError( f"scope name {scope_name!r} sanitises to an empty slug " "(contains no ASCII letters, digits, or dashes) — " "please rename the scope to something containing ASCII " "characters before retrying.", ) if len(slug) > _AWS_OWNER_ALIAS_LOCAL_PART_MAX: raise ValueError( f"scope name {scope_name!r} slugs to {len(slug)} chars, " f"exceeds Google's {_AWS_OWNER_ALIAS_LOCAL_PART_MAX}-char " "local-part cap — please pick a shorter name.", ) return f"{slug}@{_AWS_OWNER_ALIAS_DOMAIN}" # ── Service ──────────────────────────────────────────────────────── class CloudScopeRequestService: # Outcome of the most recent ``withdraw()`` call's best-effort # Feishu side-cancel. Routers / MCP tools read it after # ``withdraw()`` to surface ``feishu_cancel_synced``. last_feishu_cancel_synced: bool = True def __init__(self, db: AsyncSession) -> None: self.db = db self.audit = AuditService(db) self.notify = NotificationService(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, integration_id: str, name: str, description: str | None, monthly_budget_usd: float, cloud_account_type: str, trace_id: str, ) -> CmdbCloudScopeRequest: from app.core.distributed_lock import LockBusy, distributed_lock clean_name = _validate_name(name) clean_budget = _validate_budget(monthly_budget_usd) clean_type = (cloud_account_type or "").strip() if not clean_type: raise AppException( code="CLOUD_SCOPE_REQUEST_TYPE_REQUIRED", message="Please choose / fill in the account type.", status_code=400, ) clean_description = (description or "").strip() or None integration = await self.db.get( Integration, uuid.UUID(integration_id), ) if integration is None or integration.kind != "cloud": raise NotFoundError("Integration", integration_id) # Pre-launch hardening 2026-05-16 (LS4 P1): two near-simultaneous # clicks on "提交云子账号申请" used to create TWO Feishu instances # + TWO DB rows; on auto-create providers (AWS Organizations) the # admin approving each side spawned TWO REAL sub-accounts because # each row was treated as a separate request — and AWS Organizations # CreateAccount has NO native idempotency token (the previously- # assumed ``ClientRequestToken`` is not a valid parameter on this # API; corrected 2026-05-25). The lock + idempotent re-return # guarantee a single instance per (requester, integration, name) # at a time. LockBusy converts to a 409 so the FE can show # "请稍候,正在处理…" rather than triggering the rare second-flush race. import hashlib name_hash = hashlib.sha1( clean_name.encode("utf-8"), usedforsecurity=False, ).hexdigest()[:16] lock_key = ( f"cloud_scope_request_submit:{requester.id}:" f"{integration.id}:{name_hash}" ) async def _check_existing() -> CmdbCloudScopeRequest | None: existing = (await self.db.execute( select(CmdbCloudScopeRequest).where( CmdbCloudScopeRequest.requester_person_id == str(requester.id), CmdbCloudScopeRequest.integration_id == str(integration.id), CmdbCloudScopeRequest.name == clean_name, CmdbCloudScopeRequest.status.in_( ( SCOPE_REQUEST_STATUS_PENDING, SCOPE_REQUEST_STATUS_FEISHU_APPROVED, SCOPE_REQUEST_STATUS_PROVISIONING, ), ), ).with_for_update().execution_options(populate_existing=True) )).scalars().first() return existing # Round-1 review follow-up (Review-E P1-1): pre-fix wrapped # ``distributed_lock(...)`` in try/except, but the call is # synchronous and returns an ``@asynccontextmanager`` — it # never raises here. Redis-down failures surface inside # ``async with cm:`` as a Redis / Connection error (NOT # ``LockBusy``), so the original dead-code branch never ran # and Redis outages would land as a generic 500. Now: enter # the context manager directly, catch ``LockBusy`` (legitimate # contention → 409 retry), and let any other Redis-layer # error propagate as a clean infrastructure failure so ops # sees the real cause instead of a fake "single-app fallback". try: async with distributed_lock(lock_key, ttl=120): dup = await _check_existing() if dup is not None: raise AppException( code="CLOUD_SCOPE_REQUEST_DUPLICATE", message=( "You already have a pending request for this name " "on this cloud integration; check the request " "list. Helix refuses to double-submit to avoid " "creating two cloud sub-accounts." ), status_code=409, ) # Re-load integration inside the lock so a concurrent # admin disable between ``db.get`` (above) and here is # respected (Review-E P2-4). # Round-3 review follow-up (LS4 P1): if the integration # was hard-deleted between ``db.get`` and this refresh # (sub-ms race with an admin DELETE), SQLAlchemy raises # ``InvalidRequestError`` / ``OperationalError`` — not # ``LockBusy`` — and the request 500s. Convert to the # same 404 the original ``db.get`` path produces so # the caller sees a clean signal. try: await self.db.refresh(integration) except (InvalidRequestError, OperationalError) as exc: raise NotFoundError( "Integration", str(integration.id), ) from exc req = await self._submit_inner( requester=requester, integration=integration, clean_name=clean_name, clean_budget=clean_budget, clean_type=clean_type, clean_description=clean_description, trace_id=trace_id, ) # Commit INSIDE the lock so the dedup row is durable BEFORE the # lock is released. The router used to commit only after this # returned (lock already gone), leaving a window where a racing # submit's _check_existing (a separate transaction) couldn't see # this still-uncommitted row and double-created a REAL cloud # sub-account — CreateAccount has no native idempotency token, # so the Redis lock is the sole guard and must cover the commit. await self.db.commit() return req except LockBusy: raise AppException( code="CLOUD_SCOPE_REQUEST_IN_FLIGHT", message=( "Another submission is being processed for this name. " "Please retry in a few seconds." ), status_code=409, ) async def _submit_inner( self, *, requester: IdentityPerson, integration: Integration, clean_name: str, clean_budget: Decimal, clean_type: str, clean_description: str | None, trace_id: str, ) -> CmdbCloudScopeRequest: """Inner body of :meth:`submit` — runs inside the de-dup lock. Split out so the lock-acquire path stays linear / readable and the body can be reached from both the lock-protected and the lock-unavailable branches without code duplication. """ # Direct attribute access (NOT ``getattr(..., True)``): if the # column ever gets renamed in a future migration the getattr # fallback would silently fail-OPEN and let disabled # integrations through. The bug class bit PRs 122/127/128/129; # codify the rule here. if not integration.enabled: raise AppException( code="CLOUD_INTEGRATION_DISABLED", message="Selected cloud integration is disabled", status_code=409, ) # Round-22 audit: cap simultaneous PENDING / FEISHU_APPROVED # requests per (requester, integration). Without a cap, an # employee with a stolen JWT or a buggy automation can fire # 200 submit calls with 200 distinct names — each lands a # Feishu approval card in every global_admin's inbox and an # ApprovalRequest + CmdbCloudScopeRequest row. If any one # gets approved, real cloud sub-account spawns. The name-hash # dedup lock catches duplicates of the SAME name; different # names slip through. A small cap (≤ 5 simultaneous open # requests per requester per integration) bounds the spam # vector while still allowing a developer with legitimately- # many parallel project asks to file them via successive # approvals. # # Concurrency: this is a count-then-insert pattern, not a # FOR UPDATE on a synthetic per-(requester, integration) # sentinel row. With N truly-parallel submits from the same # requester, the cap is bypassable to N — each submit reads # ``open_count < cap`` and proceeds. The outer dedup lock # ``cloud_scope_request_submit:{requester}:{integration}: # {name_hash}`` serializes SAME-NAME submits, so the realistic # bypass requires a script firing distinct names faster than # the count query commits. The cap is a SPAM bound (Feishu # inbox protection), NOT a security gate — the four-eyes # approval invariant on every request is unchanged. Accepting # the residual bypass; widening the lock to per-requester # would serialize legitimate same-day-different-project # submits and hurt UX for no security benefit. pending_count_res = await self.db.execute( select(func.count(CmdbCloudScopeRequest.id)) .where( CmdbCloudScopeRequest.requester_person_id == str(requester.id), CmdbCloudScopeRequest.integration_id == str(integration.id), CmdbCloudScopeRequest.status.in_( ( SCOPE_REQUEST_STATUS_PENDING, SCOPE_REQUEST_STATUS_FEISHU_APPROVED, SCOPE_REQUEST_STATUS_PROVISIONING, ), ), ) ) open_count = int(pending_count_res.scalar() or 0) _PER_USER_OPEN_REQUEST_CAP = 5 if open_count >= _PER_USER_OPEN_REQUEST_CAP: raise AppException( code="CLOUD_SCOPE_REQUEST_QUOTA_EXCEEDED", message=( f"You already have {open_count} open scope-creation " f"requests on this cloud integration " f"(cap = {_PER_USER_OPEN_REQUEST_CAP}). Wait for " f"existing requests to be approved/rejected or " f"withdraw stale ones before filing more." ), status_code=429, ) # All cloud integrations are eligible for a scope request. If # the provider's driver implements ``create_scope`` (only AWS # Organizations today), ``approve`` will dispatch to the cloud # API automatically; otherwise the approval path stops at # ``feishu_approved`` and IT manually creates the sub-account # in the cloud console, then calls # ``POST /scope-requests/{id}/manual-provision`` with the # resulting external_id to close the loop. We deliberately do # NOT block the submission here on driver capability — the # request flow is the same end-to-end, and an admin can always # finish the manual leg even when the driver can't. from app.integrations.registry import get_driver_class driver_cls = get_driver_class("cloud", integration.provider) if driver_cls is None: raise AppException( code="CLOUD_DRIVER_NOT_FOUND", message=f"No driver registered for {integration.provider!r}", status_code=409, ) # 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_scope_template_widgets(definition.get("form", "[]")) request_id = str(uuid.uuid4()) # Department auto-fills from HRIS (the approval template asks # for it but the requester shouldn't have to retype data we # already own). Falls back to "—" so a person with no # department on file still produces a non-empty form value. department = ( requester.department or requester.primary_department or "—" ) owner_label = ( requester.display_name or requester.work_email or str(requester.id) ) # Per the 2026-05-02 ADR with the requester ("a"): Total = monthly × 1. # The Feishu template has Total + monthly-per-row block; we fill # both with the same number so the approver sees a coherent # picture without having to mentally multiply. budget_str = f"{float(clean_budget):.2f}" current_month = datetime.now(UTC).strftime("%Y-%m") # 2026-05-12: the production approval template does NOT have a # dedicated "金额" / "expenses" widget — the budget-estimate # fieldList ships with just a "月份" child column (verified by # dumping ``latest_instance_detail.form`` for instance # 202605120025). Per the requester's call ("don't change the # template, change the code"), we surface the budget number in # two places that DO exist: appended to the description, and # inline on the 月份 cell. Approvers see the dollar figure # whether they read the narrative or scan the table. description_with_budget = ( f"{clean_description or clean_name}\n\n" f"月度预算 / Monthly budget: $ {budget_str} USD" ) month_with_budget = f"{current_month} $ {budget_str} USD" field_values: dict[str, Any] = { "provider": integration.provider, "owner": owner_label, "name": clean_name, "department": department, "description": description_with_budget, "monthly_budget": budget_str, "total_budget": budget_str, "month": month_with_budget, "request_id": request_id, "cloud_account_type": clean_type, # The production template's "云账号（项目）ID" widget is # `required=true` even though the cloud account ID won't # exist until after approval+provisioning. If we leave it # unset, the required-input fallback (line ~155) silently # writes the description into the ID column — which is # how PR-pre-fix surfaced "for NA GTM Engineering" inside # the ID slot in production. Pass an explicit placeholder # so the approver sees a clear "TBD" instead of a # misleading copy of the description. "cloud_account_id": "(待开通后填写 / TBD)", } form_payload = _build_scope_form_payload(widgets, field_values) # Map the requester's email to a Feishu user_id (same approach # every other Feishu-backed flow uses). feishu_user_id = await client.resolve_user_id_by_email( requester.work_email, ) semantic_payload = { "integration_id": str(integration.id), "provider": integration.provider, "name": clean_name, "description": clean_description, "monthly_budget_usd": str(clean_budget), "cloud_account_type": clean_type, } # The outer lock/dedup boundary is (requester, integration, name), so # give each such logical stream its own intent scope. This preserves # the product's supported parallel requests for different names. name_hash = hashlib.sha1( clean_name.encode("utf-8"), usedforsecurity=False, ).hexdigest()[:16] intent_scope = f"cmdb_scope:{integration.id}:{name_hash}" intent = await submit_with_durable_intent( self.db, client=client, scope=intent_scope, applicant_id=requester.id, semantic_payload=semantic_payload, provider_payload={ "approval_code": approval_code, "identity_type": "user_id", "identity_value": feishu_user_id, "form": form_payload, }, finalization_payload={ **semantic_payload, "request_id": request_id, "approval_request_id": str(uuid.uuid4()), "department": department, "approval_code": approval_code, }, trace_id=trace_id, ) req = await self._finalize_submission_intent( intent.id, recovered=intent.attempt_count > 1, ) # 2026-05-12: do NOT push a "[云子账号申请] 去 Helix 审批" card to # global_admins here. The Feishu approval-instance create call # above already lands the native approval card in every # approver's Feishu inbox via Feishu's own dispatch (the # template's "approver" node controls the recipient list). # Pushing a parallel Helix-branded "去 Helix 审批" card creates # two competing inboxes — the user policy is that **all # approvals happen inside Feishu**, so the Helix path was # always misleading: approving there does not write back to # the Feishu instance (the FeishuApprovalClient has no # approve-task endpoint), leaving the Feishu card forever # pending. Cleaner to surface a single inbox. # The other three lifecycle notifications (rejected, # provisioned, manual_required) stay — they're post-decision # outcomes the approver/requester needs, not duplicate # approval prompts. return req async def _finalize_submission_intent( self, intent_id: uuid.UUID, *, recovered: bool, ) -> CmdbCloudScopeRequest: """Atomically materialise the scope row, mirror, audit and intent.""" 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 not intent.scope.startswith("cmdb_scope:"): raise RuntimeError("cloud scope submission intent disappeared") if intent.state == "completed": request_id = terminal_intent_resource_id( intent, resource_type="cmdb_cloud_scope_request", ) else: payload = dict((intent.payload or {}).get("finalization") or {}) request_id = str(payload["request_id"]) existing = await self.db.scalar( select(CmdbCloudScopeRequest) .where(CmdbCloudScopeRequest.id == request_id) .with_for_update() .execution_options(populate_existing=True) ) if intent.state == "completed": if existing is None: raise RuntimeError("completed cloud-scope intent has no row") return existing payload = dict((intent.payload or {}).get("finalization") or {}) if intent.state != "provider_created" or not intent.external_id: raise RuntimeError( f"cannot finalize cloud scope intent in {intent.state!r}" ) now = datetime.now(UTC) if existing is None: req = CmdbCloudScopeRequest( id=request_id, requester_person_id=str(intent.applicant_id), integration_id=str(payload["integration_id"]), provider=str(payload["provider"]), name=str(payload["name"]), description=payload.get("description"), monthly_budget_usd=Decimal(str(payload["monthly_budget_usd"])), status=SCOPE_REQUEST_STATUS_PENDING, feishu_instance_code=intent.external_id, metadata_json={ "cloud_account_type": str(payload["cloud_account_type"]), "department": str(payload["department"]), }, created_at=now, updated_at=now, ) self.db.add(req) await self.db.flush() approval_request = ApprovalRequest( id=uuid.UUID(str(payload["approval_request_id"])), external_id=intent.external_id, approval_type=_FEISHU_FLOW_KEY, applicant_id=intent.applicant_id, status=SCOPE_REQUEST_STATUS_PENDING, source_payload={ "scope_request_id": req.id, "approval_code": str(payload["approval_code"]), "integration_id": str(payload["integration_id"]), "provider": str(payload["provider"]), "name": str(payload["name"]), "cloud_account_type": str(payload["cloud_account_type"]), }, ) self.db.add(approval_request) await self.db.flush() else: req = existing intent.approval_request_id = uuid.UUID(str(payload["approval_request_id"])) intent.state = "completed" intent.completed_at = 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 else "user", actor_user_id=None if recovered else intent.applicant_id, action=( "approval.submission_intent.reconciled" if recovered else "cloud.scope_request.submit" ), status="success", resource_type="cmdb_cloud_scope_request", resource_id=request_id, detail={ "intent_id": str(intent.id), "provider_request_uuid": intent.provider_request_uuid, "feishu_instance_code": intent.external_id, "recovered": recovered, }, ) compact_terminal_intent_payload( intent, resource_type="cmdb_cloud_scope_request", resource_id=request_id, ) await self.db.commit() await self.db.refresh(req) return req async def reconcile_submission_intents( self, *, min_age_minutes: int = 2, batch_limit: int = 100, ) -> dict[str, int]: """Recover cloud-scope creates without requiring an HTTP retry.""" cutoff = datetime.now(UTC) - timedelta(minutes=min_age_minutes) recoverable_states = ( "prepared", "submitting", "unknown", "provider_created", ) intent_ids = list((await self.db.scalars( select(ApprovalSubmissionIntent.id) .where( ApprovalSubmissionIntent.scope.like( r"cmdb\_scope:%", escape="\\", ), ApprovalSubmissionIntent.state.in_(recoverable_states), ( (ApprovalSubmissionIntent.state == "provider_created") | (ApprovalSubmissionIntent.updated_at < cutoff) ), reconciliation_due_predicate(datetime.now(UTC)), ) .order_by(*reconciliation_order()) .limit(batch_limit) )).all()) summary = { "checked": len(intent_ids), "reconciled": 0, "still_unknown": 0, "failed": 0, } if not intent_ids: return summary client = await self._client() for intent_id in intent_ids: try: intent = await self.db.get(ApprovalSubmissionIntent, intent_id) if intent is None or intent.state == "completed": continue intent_payload = dict(intent.payload or {}) intent = await submit_with_durable_intent( self.db, client=client, scope=intent.scope, applicant_id=intent.applicant_id, semantic_payload=dict(intent_payload.get("semantic") or {}), provider_payload=dict(intent_payload.get("provider") or {}), finalization_payload=dict( intent_payload.get("finalization") or {} ), trace_id=str( intent_payload.get("trace_id") or f"approval-intent:{intent.id}" ), ) await self._finalize_submission_intent(intent.id, recovered=True) summary["reconciled"] += 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 # ── Approve / reject (admin actions) ──────────────────────── async def approve( self, *, req_id: str, approver: IdentityPerson, trace_id: str, approver_is_system: bool = False, ) -> CmdbCloudScopeRequest: """Approve a scope-creation request. ``approver_is_system=True`` is used by the Feishu callback dispatcher when the upstream approval came from Feishu — Feishu's payload gives a Feishu user_id, not a Helix IdentityPerson, so we cannot attribute to a specific human approver. The audit row is marked ``actor_type='integration'`` and the original Feishu instance code is captured in ``trace_id`` for forensic linking. The self-approval guard is also bypassed in this path because the ``approver`` argument is the requester (sentinel), not the actual approver. """ req = await self._lock_request(req_id) # Recovery window (added 2026-05-12 with the mid-method commit # at line ~696): if a worker died between the FEISHU_APPROVED # commit and the PROVISIONING state-set on the auto-create # path, the row is stuck at FEISHU_APPROVED with no cloud # account yet. Allow the admin to retry ``approve()`` to # recover — the retry skips the status-flip / audit / Feishu # cancel (already done in the original attempt, idempotent # nonetheless) and goes straight to the provisioning step. # ``_provision`` itself is idempotent: the AWS driver sweeps # ``list_create_account_status`` for an IN_PROGRESS / SUCCEEDED # entry matching AccountName before calling create_account, so # a re-run resumes polling the existing CreateAccountStatus # row rather than minting a second one (corrected 2026-05-25 — # AWS Organizations has no native ClientRequestToken). # # The callback path (``approver_is_system=True``) does NOT # benefit from the retry-on-FEISHU_APPROVED widening — Feishu's # callback contract is "this is the FIRST time we've seen this # approval", and treating a stale re-delivery as a fresh # transition could re-enter provisioning while a previous # attempt is still mid-flight in another worker. is_recovery_retry = ( req.status == SCOPE_REQUEST_STATUS_FEISHU_APPROVED and not approver_is_system ) valid_states = ( SCOPE_REQUEST_STATUS_PENDING, SCOPE_REQUEST_STATUS_FAILED, ) if req.status not in valid_states and not is_recovery_retry: raise AppException( code="CLOUD_SCOPE_REQUEST_INVALID_STATE", message=( f"Cannot approve a request in {req.status!r} state" ), status_code=409, ) # Block self-approval (human path only). Even a global_admin # should not approve their own scope-creation request — # creating a real cloud sub-account with up to 9.9M USD/mo # budget without four-eyes is exactly the wrong default. # Recovery-retry path bypasses this for the same reason # ``approver_is_system`` does: the FOUR-eyes invariant was # already enforced on the original attempt that left the row # at FEISHU_APPROVED. if ( not approver_is_system and not is_recovery_retry and str(req.requester_person_id) == str(approver.id) ): raise AppException( code="CLOUD_SCOPE_REQUEST_SELF_APPROVAL_FORBIDDEN", message=( "Cannot approve your own scope-creation request — " "ask another global_admin to review." ), status_code=403, ) # Round-22 audit suggested adding a self-approval block on # the recovery-retry path too. Round-2 review caught that # this would break the documented contract — the recovery # branch was specifically added to let "the original approver # retries" (which can legitimately be the requester themselves # when the original approval came via the Feishu callback, # ``approver_is_system=True``). The four-eyes invariant on # the FIRST attempt is what gatekeeps the FEISHU_APPROVED # transition; recovery-retry is a resume, not a fresh approve. # Tenant-side Feishu template misconfiguration that resolves # the approver to the requester is an operator-config issue, # not something Helix can or should re-validate here. # Leaver invariant (approval-time gate, 2026-07-07): a scope # request can sit pending in Feishu for days; if the requester # offboards in that window, approving it must NOT mint a real, # billable AWS/cloud sub-account owned by a departed employee. # Every sibling Feishu-approval execute path re-checks this at # execute time (ai_tool / announcement / apikey / sso / vpn / # soc2_email_group / data_access) — scope-request now matches. # ``_provision`` carries a last-ditch RuntimeError gate (#1706), # but refusing HERE — before the FEISHU_APPROVED / PROVISIONING # transitions and before the cloud driver call — lands the row # at status=rejected with a clear reason (not a retryable # status=failed) and emits the sibling # ``*.callback_refused_leaver`` audit action. Applies to the # admin HTTP path, the Feishu callback path # (``approver_is_system=True``), and the recovery-retry resume: # a departure between the crash and the resume must also refuse. requester_row = await self.db.get( IdentityPerson, uuid.UUID(req.requester_person_id), ) if requester_row is None or requester_row.deleted_at is not None or ( requester_row.employment_status in ("left", "leaver_pending") ): requester_status = ( requester_row.employment_status if requester_row else None ) req.status = SCOPE_REQUEST_STATUS_REJECTED # Same Feishu-attribution rule as ``reject()``: never stamp # the requester as their own rejector on the system path. req.approver_person_id = ( None if approver_is_system else str(approver.id) ) req.rejected_at = datetime.now(UTC) req.rejection_reason = ( "Requester offboarded while the request was pending — " "refusing to provision a cloud sub-account for a " "departing/departed employee." ) await self.audit.log( trace_id=trace_id, actor_type="integration" if approver_is_system else "user", actor_user_id=None if approver_is_system else approver.id, action="cloud.scope_request.callback_refused_leaver", status="failure", resource_type="cmdb_cloud_scope_request", resource_id=req.id, detail={ "requester_status": requester_status, "reason": "requester_offboarded_during_pending", }, ) # Admin override path: cancel the parallel Feishu card so the # approver's inbox doesn't sit on a stale "pending" forever # (mirrors ``reject()``). Commit the durable refusal first so # a Feishu API hiccup can't undo it. The callback path # (``approver_is_system=True``) skips this — Feishu's own # state already moved and the outer dispatcher commits. if not approver_is_system and req.feishu_instance_code: await self.db.commit() await self._cancel_feishu_instance_on_admin_decision( req, approver, approver_is_system, trace_id, kind="reject", ) return req if not is_recovery_retry: # First-pass approval: flip state + audit + (admin path) # commit-and-cancel-Feishu. On the recovery-retry path we # skip ALL of this — it was already done in the original # attempt that crashed before reaching the provisioning # commit. Re-doing it would write duplicate audit rows and # re-fire a no-op Feishu cancel on an already-cancelled # instance (which Feishu's API rejects, generating warning # noise on every recovery). # Mark approved in one transaction so a worker crash mid-create # still leaves a well-defined "we said yes" state. req.status = SCOPE_REQUEST_STATUS_FEISHU_APPROVED # When the approval is system-attributed, do NOT set # approver_person_id to the requester (which would falsely # show "X approved their own request" in the audit log). # Leave it null and rely on trace_id (Feishu instance code) # for the forensic link to the actual Feishu approver. req.approver_person_id = None if approver_is_system else str(approver.id) req.approved_at = datetime.now(UTC) # Re-approve from a previously ``failed`` state must clear the # stale failure reason — otherwise the row reads ``status= # provisioning`` with last attempt's error like # "AWS Organizations CLOSING_ACCOUNT" still showing in the UI, # confusing reviewers into thinking the new approval also # failed. (Round-17 audit.) req.failure_reason = None await self.audit.log( trace_id=trace_id, actor_type="integration" if approver_is_system else "user", actor_user_id=None if approver_is_system else approver.id, action="cloud.scope_request.approve", status="success", resource_type="cmdb_cloud_scope_request", resource_id=req.id, ) await self.db.flush() # 2026-05-12: cancel the parallel Feishu approval instance # when this approve came from the Helix admin UI override # path (not the Feishu callback). Without this, the Feishu # approver's inbox carries the card as "pending" forever # — Helix has no ``approve_task`` endpoint to mark it # approved on the Feishu side, so cancelling is the only # way to get the inbox in agreement with reality. The # Feishu callback path (``approver_is_system=True``) skips # this — Feishu's own state has already moved. # # Commit local state BEFORE the Feishu call so a Feishu # API hiccup doesn't roll back our durable decision; # mirrors ``withdraw()`` and the round-2-audit pattern. if not approver_is_system and req.feishu_instance_code: await self.db.commit() await self._cancel_feishu_instance_on_admin_decision( req, approver, approver_is_system, trace_id, kind="approve", ) if is_recovery_retry: # Concurrent-admin race window (post-commit at line ~696 # releases the X-lock from ``_lock_request``): if Admin A # is still mid-flight in the auto-provisioning path and # Admin B clicks Retry, both workers can reach # ``_provision`` concurrently. The AWS driver sweeps # ``list_create_account_status`` for an IN_PROGRESS / # SUCCEEDED entry matching AccountName before calling # ``create_account``; both workers see the same row and # converge to the SAME external_id (corrected 2026-05-25 — # AWS Organizations has no native ClientRequestToken). The # downstream # ``CmdbCloudScope`` insert is gated by UNIQUE # ``(integration_id, external_id)``; the losing worker # catches IntegrityError inside ``_provision`` (see the # try/except around the flush) and recovers the existing # row, converging to the same final state. The only # user-observable artifact is a duplicate # ``_notify_requester_provisioned`` push (same scope_id, # tolerable). Trace a recovery retry distinctly so ops # can spot a crash-window recovery in the audit log # without it being confused with a fresh approval. The # original approval audit row remains; this adds a paired # entry. await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=approver.id, action="cloud.scope_request.approve_resume", status="success", resource_type="cmdb_cloud_scope_request", resource_id=req.id, detail={ "prior_status": SCOPE_REQUEST_STATUS_FEISHU_APPROVED, "reason": ( "admin retried approve to resume provisioning " "after a crash between the FEISHU_APPROVED " "commit and the PROVISIONING transition" ), }, ) # Driver capability gate. If the provider's driver implements # ``create_scope`` (today: AWS Organizations) we go straight # into auto-provisioning. Otherwise the request stays in # ``feishu_approved`` and an admin finishes via # ``manual_provision`` after creating the sub-account in the # cloud console — both paths are first-class. from app.integrations.registry import get_driver_class driver_cls = get_driver_class("cloud", req.provider) auto_create = ( driver_cls is not None and getattr(driver_cls, "create_scope", None) is not CloudDriver.create_scope ) if not auto_create: await self._notify_admins_manual_required(req, approver, trace_id) return req # Move into provisioning + actually call the cloud API. We # COMMIT the approve→provisioning state transition before # the driver call so a worker crash leaves the request in a # well-defined retryable state. # # Pre-launch hardening 2026-05-06 (round 2): the original # plan was to write the breadcrumb via a fresh out-of-tx # session. That deadlocked against the parent's # ``with_for_update`` X-lock from ``_lock_request`` (the # breadcrumb session waits for the parent's lock to release, # which can't happen inside the same handler). The robust fix # is to commit the parent session itself — that releases the # X-lock AND persists the PROVISIONING transition. SQLAlchemy # auto-starts a new transaction on the next write, so the # post-driver ``req.status = PROVISIONED`` lands cleanly. # # If the driver call crashes between this commit and the # PROVISIONED write, the row stays in PROVISIONING. The # state guard at the top of ``approve()`` refuses retries # on PROVISIONING (deliberate safety net — a driver call # may still be running in another worker; double-creating # would be wasteful). Recovery from this state is a DB # poke (UPDATE status='failed' then retry approve) or # ``manual_provision`` if the admin already knows the # cloud-side external_id. The FEISHU_APPROVED recovery # widening (2026-05-12) addresses the smaller, earlier # window before this commit — the PROVISIONING window # remains intentionally strict. # Round-22 audit follow-up: re-fetch the Integration row with # FOR UPDATE before committing the PROVISIONING transition so # an admin who toggled ``enabled=False`` between submit and # approve can't have us mint a real cloud sub-account against # a disabled integration. The submit-side gate happens at # request time; this is the approval-time gate. Without it, # an Integration disabled mid-flight still gets a cloud-side # account + a CmdbCloudScope bound to it. # # Note (Round-2 review): the FOR UPDATE lock is released at # ``await self.db.commit()`` BEFORE the multi-second cloud # ``create_scope`` call. An admin toggling ``enabled=False`` # AFTER the commit and BEFORE the driver returns can still # land an account. This gate is defense-in-depth that # closes the larger pre-commit window; the residual narrow # window past commit is accepted (CONFIG drift detection + # mid-flight admin actions are rare and operators are # expected to wait until provisioning settles before # toggling). from sqlalchemy import select as _select_local from app.integrations.models import Integration integ_check_res = await self.db.execute( _select_local(Integration) .where(Integration.id == uuid.UUID(req.integration_id)) .with_for_update(), ) integ_row_check = integ_check_res.scalar_one_or_none() if integ_row_check is None or not integ_row_check.enabled: raise AppException( code="CLOUD_INTEGRATION_DISABLED", message=( "Cannot provision a scope on a disabled integration. " "Re-enable the integration in admin console then retry." ), status_code=409, ) req.status = SCOPE_REQUEST_STATUS_PROVISIONING await self.db.commit() try: scope = await self._provision(req, trace_id=trace_id) except Exception as exc: # noqa: BLE001 logger.exception( "scope_request %s provisioning failed", req.id, ) req.status = SCOPE_REQUEST_STATUS_FAILED req.failure_reason = str(exc)[:1000] await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=approver.id, action="cloud.scope_request.provision", status="failure", resource_type="cmdb_cloud_scope_request", resource_id=req.id, detail={"error": format_exception_for_log(exc, max_len=400)}, ) # Round-22 audit: commit the FAILED transition + audit row # NOW so the durable state doesn't depend on the outer # caller's commit. Previously a crash between this return # and the router's trailing ``db.commit()`` (e.g. the post- # provision BackgroundTask wiring throwing) left the row # at PROVISIONING — the very "stuck retryable state" the # inline comment promises against. async with best_effort( self.db, label="scope_request_service best-effort step", log=logger, ) as be_1471_step: await self.db.commit() if not be_1471_step.ok: logger.error( "scope_request %s: failed-state commit failed; " "row may persist as PROVISIONING until manual fix", req.id, ) return req req.status = SCOPE_REQUEST_STATUS_PROVISIONED req.created_scope_id = str(scope.id) req.provisioned_at = datetime.now(UTC) await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=approver.id, action="cloud.scope_request.provision", status="success", resource_type="cmdb_cloud_scope_request", resource_id=req.id, detail={"scope_id": str(scope.id)}, ) # Commit the PROVISIONED transition + audit row NOW so the # durable state survives any error in the notify step below # (or anywhere between here and the router's trailing commit). # Symmetric with the failure-path commit at line ~1088 — the # same rationale ("durable state doesn't depend on the outer # caller's commit") applies to the success branch. Pre-fix, # a Feishu/email outage causing ``_notify_requester_provisioned`` # to raise left the row at PROVISIONING with a freshly-minted # cloud scope already created provider-side — the exact # "stuck retryable state" the failure-path commit guards # against. async with best_effort( self.db, label="scope_request_service best-effort step", log=logger, ) as be_1505_step: await self.db.commit() if not be_1505_step.ok: logger.error( "scope_request %s: provisioned-state commit failed; " "row may persist as PROVISIONING until manual fix", req.id, ) await self._notify_requester_provisioned(req, scope, trace_id) return req async def _cancel_feishu_instance_on_admin_decision( self, req: CmdbCloudScopeRequest, actor: IdentityPerson, approver_is_system: bool, trace_id: str, *, kind: str, ) -> None: """Best-effort: when a global_admin overrides the request via the Helix admin UI (not via the Feishu callback), cancel the parallel Feishu approval instance so the approver's inbox doesn't sit on a stale "pending" card forever. Skipped on the Feishu callback path (``approver_is_system=True``) — the Feishu instance has already moved to its terminal state on that side. Skipped when there's no instance code (legacy rows from before Feishu approval was mandatory, or rare submit-race rollback survivors). Mirrors the cancel pattern in :meth:`withdraw` — failure to cancel does NOT block the local decision (admins can clean orphans through the Feishu admin UI). Same shape as ``_revoke_pending_seat_on_rejection`` in ``app.approval.service`` etc. """ if approver_is_system or not req.feishu_instance_code: return try: client = await self._client() flow = await get_approval_flow( _FEISHU_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, ) logger.info( "scope_request %s admin %s cancelled feishu instance %s", req.id, kind, req.feishu_instance_code, ) except Exception as e: # noqa: BLE001 logger.warning( "scope_request %s admin %s: feishu cancel failed for " "instance %s — orphan card lingers in approver inbox: %s", req.id, kind, req.feishu_instance_code, e, ) await notify_feishu_cancel_desync( self.db, flow_label="云子账号申请(管理员决策)", resource_type="cmdb_cloud_scope_request", resource_id=str(req.id), instance_code=req.feishu_instance_code, error=e, trace_id=trace_id, ) async def _backfill_account_id_to_feishu( self, *, req: CmdbCloudScopeRequest, requester: IdentityPerson, trace_id: str, ) -> None: """Surface the freshly-provisioned cloud account/project ID back onto the Feishu approval instance. Why a comment and not the form field: the template's "Cloud account (project) ID" widget is editable at the IT 开通 node, but by the time provisioning runs the instance is already in its terminal ``APPROVED`` state — the approval IS what triggers creation, so the ID literally does not exist until after the node is signed off. Feishu forbids editing the form of a submitted approval (see ``FeishuApprovalClient.add_instance_comment`` — the ``approve_task`` form-write path the asset supplement flow uses only works on a *pending* task, of which there are none here). The sanctioned in-flight channel for post-approval material is the instance comment stream, so we post the ID there; approvers and CC recipients see it in the approval-card timeline instead of the "(待开通后填写 / TBD)" placeholder. Best-effort: the durable PROVISIONED state is already committed by the caller, so a Feishu hiccup here must never bubble. """ if not req.feishu_instance_code or not req.created_scope_id: return scope = await self.db.get( CmdbCloudScope, uuid.UUID(req.created_scope_id), ) if scope is None or not scope.external_id: return try: client = await self._client() feishu_user_id = await client.resolve_user_id_by_email( requester.work_email, ) lines = [ "云账号（项目）已开通 / Cloud account (project) provisioned", f"账号（项目）ID / Account (project) ID: {scope.external_id}", ] if scope.display_name: lines.append(f"名称 / Name: {scope.display_name}") await client.add_instance_comment( instance_code=req.feishu_instance_code, user_id=feishu_user_id, content_text="\n".join(lines), ) logger.info( "scope_request %s: backfilled account id %s onto feishu " "instance %s (trace=%s)", req.id, scope.external_id, req.feishu_instance_code, trace_id, ) except Exception as e: # noqa: BLE001 — best-effort, never block logger.warning( "scope_request %s: feishu account-id backfill comment " "failed for instance %s: %s", req.id, req.feishu_instance_code, e, ) async def reject( self, *, req_id: str, approver: IdentityPerson, reason: str | None, trace_id: str, approver_is_system: bool = False, ) -> CmdbCloudScopeRequest: """Reject a pending scope-creation request. Mid-method commit-before-notify ordering (2026-05-12): on the admin override path (``approver_is_system=False``) with a live Feishu instance, we commit the local reject + audit row first, then best-effort cancel the parallel Feishu card. Net effect: a Feishu API hiccup never undoes the reject, and the requester's rejection notification still fires after. Callers MUST NOT rely on the outer transaction extending beyond ``reject()`` completion to atomically batch follow-up writes; do those in a new transaction. The Feishu callback path (system attribution) keeps the original single-commit flow — the rejection IS the commit point. """ req = await self._lock_request(req_id) if req.status != SCOPE_REQUEST_STATUS_PENDING: raise AppException( code="CLOUD_SCOPE_REQUEST_INVALID_STATE", message=( f"Cannot reject a request in {req.status!r} state" ), status_code=409, ) req.status = SCOPE_REQUEST_STATUS_REJECTED # Same Feishu-attribution rule as ``approve()``: when the # rejection came from Feishu we cannot map to a Helix person, # so do not falsely stamp the requester as their own rejector. req.approver_person_id = None if approver_is_system else str(approver.id) req.rejected_at = datetime.now(UTC) req.rejection_reason = (reason or "").strip()[:500] or None await self.audit.log( trace_id=trace_id, actor_type="integration" if approver_is_system else "user", actor_user_id=None if approver_is_system else approver.id, action="cloud.scope_request.reject", status="success", resource_type="cmdb_cloud_scope_request", resource_id=req.id, detail={"reason": req.rejection_reason}, ) # Same Feishu-inbox-orphan defence as ``approve()``: when the # admin rejects via the Helix UI override path, cancel the # parallel Feishu card so the approver doesn't see a stale # "pending" forever. if not approver_is_system and req.feishu_instance_code: await self.db.commit() await self._cancel_feishu_instance_on_admin_decision( req, approver, approver_is_system, trace_id, kind="reject", ) await self._notify_requester_rejected(req, trace_id) return req async def withdraw( self, *, req_id: str, actor: IdentityPerson, trace_id: str, ) -> CmdbCloudScopeRequest: """Requester pulls back a still-pending request. Also cancels the parallel Feishu approval instance so the approver doesn't see a stale "pending" card after the requester has changed their mind. A Feishu cancel failure does NOT block the local withdraw — admins fix orphans through the Feishu admin UI rather than letting a Feishu hiccup pin the request in pending forever. """ req = await self._lock_request(req_id) if req.requester_person_id != str(actor.id): raise AppException( code="CLOUD_SCOPE_REQUEST_NOT_OWNER", message="Only the original requester can withdraw", status_code=403, ) if req.status != SCOPE_REQUEST_STATUS_PENDING: raise AppException( code="CLOUD_SCOPE_REQUEST_INVALID_STATE", message=( f"Cannot withdraw a request in {req.status!r} state" ), status_code=409, ) # Flip local state FIRST, commit, then call Feishu best- # effort. The previous order (Feishu first, then local flip) # had a partial-failure window: Feishu succeeds, local commit # crashes (DB drop, deadlock) → Feishu shows cancelled while # Helix shows pending. Reversing the order makes Helix the # source of truth — if Feishu cancel later fails, the orphan # is reconciled on the next Feishu callback (which sees # status=WITHDRAWN locally and no-ops). req.status = SCOPE_REQUEST_STATUS_WITHDRAWN # Set ``updated_at`` explicitly so the post-commit hydration in # the router doesn't lazy-load — same MissingGreenlet trap # ``submit()`` documents at the top of this file. req.updated_at = datetime.now(UTC) # Mirror status onto the parallel ApprovalRequest so the admin # centre + Feishu callback dispatcher stop trying to act on it. await self._mirror_approval_request_status( req, "cancelled", ) await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor.id, action="cloud.scope_request.withdraw", status="success", resource_type="cmdb_cloud_scope_request", resource_id=req.id, ) # Commit locally so the row + audit are durable BEFORE we # touch Feishu. The router's outer commit will be a no-op if # nothing else changed. await self.db.commit() # Best-effort Feishu cancel. self.last_feishu_cancel_synced = True if req.feishu_instance_code: try: client = await self._client() flow = await get_approval_flow( _FEISHU_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 e: # noqa: BLE001 logger.warning( "Feishu cancel failed for cloud_scope %s (instance=%s): %s", req.id, req.feishu_instance_code, e, ) self.last_feishu_cancel_synced = False await notify_feishu_cancel_desync( self.db, flow_label="云子账号申请", resource_type="cmdb_cloud_scope_request", resource_id=str(req.id), instance_code=req.feishu_instance_code, error=e, trace_id=trace_id, ) return req # ── Feishu callback ───────────────────────────────────────── async def handle_feishu_callback( self, *, approval_request: ApprovalRequest, instance_detail: dict[str, Any], ) -> None: """Apply a Feishu callback to the linked cloud scope request. The dispatcher (``approval/service.py``) has already updated the parent ``ApprovalRequest`` row's ``status``; we drive the ``cmdb_cloud_scope_requests`` row's state machine to match. Wraps the existing ``approve()`` / ``reject()`` logic so the same notify + audit + provisioning path runs whether the decision came from Feishu or from a manual admin click. """ source = approval_request.source_payload or {} scope_request_id = source.get("scope_request_id") if not scope_request_id: logger.warning( "cloud_scope callback missing scope_request_id: ar=%s", approval_request.id, ) return feishu_status = (instance_detail.get("status") or "").upper() logger.info( "cloud_scope feishu callback: scope_request_id=%s feishu_status=%s", scope_request_id, feishu_status, ) try: req = await self._lock_request(scope_request_id) except NotFoundError: logger.warning( "cloud_scope callback for missing request %s — ignoring", scope_request_id, ) return # Only act on terminal Feishu states. PENDING / RUNNING etc. # land here on intermediate notifications and we just no-op. if feishu_status in ("APPROVED", "PASS"): if req.status != SCOPE_REQUEST_STATUS_PENDING: # Round-22 audit: write a forensic audit row so a # re-delivered or late callback against an already- # advanced state leaves a trail. Previously this was a # silent no-op and forensics couldn't distinguish # "Feishu never told us anything" from "we received # and ignored a redelivered callback". await self.audit.log( trace_id=f"feishu_callback:{approval_request.external_id}", actor_type="integration", actor_user_id=None, action="cloud.scope_request.callback_ignored_state_advanced", status="skipped", resource_type="cmdb_cloud_scope_request", resource_id=req.id, detail={ "feishu_status": feishu_status, "current_status": req.status, }, ) return # Feishu's payload does not give us a Helix IdentityPerson # for the actual approver — it carries a Feishu user_id # and our person-mapping lives only on the requester side. # Pass the requester as a sentinel and signal # ``approver_is_system=True`` so: # • audit_log records actor_type='integration' (NOT # "user X approved themselves" — that was the bug), # • the approver_person_id column is left null, # • the self-approval guard is bypassed. # Forensic trail: the Feishu instance_code is in trace_id, # which links to the actual approver on the Feishu side. requester = await self.db.get( IdentityPerson, uuid.UUID(req.requester_person_id), ) if requester is None: logger.error( "cloud_scope callback: requester %s not found", req.requester_person_id, ) return approved_req = await self.approve( req_id=req.id, approver=requester, trace_id=f"feishu_callback:{approval_request.external_id}", approver_is_system=True, ) # Post-provision finalize parity with the HTTP approve path # (lifecycle_router.approve_scope_request). PR #671 (2026-06-02) # moved the bg-task into an inline ``await`` on the router, # but the Feishu-callback path enters via the service's # ``approve()`` directly — so without this call, every # Feishu-approved AWS scope STILL silently skipped # ``_post_provision_finalize`` and the audit table stayed # empty for ``cloud.scope_request.auto_owner_grant`` on the # production happy-path (all approvals come from Feishu in # prod). The ``cmdb.auto_owner_grant_backfill`` cron is the # backstop but introduces up to 10 min of owner-IAM lag for # fresh scopes. ``_post_provision_finalize`` has its own # outer ``except Exception`` so any crash is logged + audited # and the backfill cron retries. if ( approved_req.status == SCOPE_REQUEST_STATUS_PROVISIONED and approved_req.created_scope_id ): from app.domains.cmdb.lifecycle_router import ( _post_provision_finalize, ) # Reuse the Feishu-callback trace_id so the finalize # audit rows correlate with the callback approve audit row. await _post_provision_finalize( approved_req.created_scope_id, trace_id=f"feishu_callback:{approval_request.external_id}", ) # Backfill the freshly-minted account/project ID onto the # still-live Feishu instance (the callback path never # cancels it). The form field can't be edited post- # approval — see ``_backfill_account_id_to_feishu`` — so # this posts the ID as an instance comment. await self._backfill_account_id_to_feishu( req=approved_req, requester=requester, trace_id=f"feishu_callback:{approval_request.external_id}", ) elif feishu_status in ("REJECTED", "REFUSED"): if req.status != SCOPE_REQUEST_STATUS_PENDING: await self.audit.log( trace_id=f"feishu_callback:{approval_request.external_id}", actor_type="integration", actor_user_id=None, action="cloud.scope_request.callback_ignored_state_advanced", status="skipped", resource_type="cmdb_cloud_scope_request", resource_id=req.id, detail={ "feishu_status": feishu_status, "current_status": req.status, }, ) return requester = await self.db.get( IdentityPerson, uuid.UUID(req.requester_person_id), ) if requester is None: return await self.reject( req_id=req.id, approver=requester, reason="Rejected via Feishu approval", trace_id=f"feishu_callback:{approval_request.external_id}", approver_is_system=True, ) elif feishu_status in ("CANCELED", "CANCELLED", "DELETED"): if req.status != SCOPE_REQUEST_STATUS_PENDING: await self.audit.log( trace_id=f"feishu_callback:{approval_request.external_id}", actor_type="integration", actor_user_id=None, action="cloud.scope_request.callback_ignored_state_advanced", status="skipped", resource_type="cmdb_cloud_scope_request", resource_id=req.id, detail={ "feishu_status": feishu_status, "current_status": req.status, }, ) return # Mirror the ``withdraw()`` pattern — flipping status alone # silently drops the audit trail + leaves the parallel # ApprovalRequest stuck in pending, which the admin centre # idempotency dispatcher then keeps trying to act on. req.status = SCOPE_REQUEST_STATUS_WITHDRAWN req.updated_at = datetime.now(UTC) await self._mirror_approval_request_status(req, "cancelled") await self.audit.log( trace_id=f"feishu_callback:{approval_request.external_id}", actor_type="integration", actor_user_id=None, action="cloud.scope_request.cancelled_via_feishu", status="success", resource_type="cmdb_cloud_scope_request", resource_id=req.id, ) requester = await self.db.get( IdentityPerson, uuid.UUID(req.requester_person_id), ) if requester is not None and requester.work_email: ctx = self._scope_request_context(req, requester=requester) baseline_title = ( f"[已撤回] 云子账号申请 {ctx['scope_name']}" ) try: await self.notify.dispatch( NotificationEnvelope( recipient=requester, title=baseline_title, body_text=baseline_title, template_key="cloud.scope_request.cancelled", context=ctx, url=f"/my/cloud/scope-requests?id={req.id}", channels=["feishu", "email"], related_resource_type="cmdb_cloud_scope_request", related_resource_id=req.id, ), trace_id=f"feishu_callback:{approval_request.external_id}", ) except Exception: # noqa: BLE001 logger.exception( "scope-request notify cancelled failed", ) async def _mirror_approval_request_status( self, req: CmdbCloudScopeRequest, new_status: str, ) -> None: """Update the parallel ApprovalRequest row to keep the admin centre + dispatcher idempotency guard in sync. No-op if no ApprovalRequest row exists (legacy rows from before Feishu wiring).""" if not req.feishu_instance_code: return ar = ( await self.db.execute( select(ApprovalRequest).where( ApprovalRequest.external_id == req.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) async def manual_provision( self, *, req_id: str, external_id: str, scope_type: str | None, actor: IdentityPerson, trace_id: str, ) -> CmdbCloudScopeRequest: """Admin closes the loop after creating the sub-account in the cloud console manually. Inserts the ``cmdb_cloud_scopes`` row with the requester as owner + default budget / alert rules, same shape as the auto path. Pre-conditions: * Request must be in ``feishu_approved`` (auto path that stalled at the driver-capability gate) or ``failed`` (auto path that errored — admin can rescue by completing the create manually). * ``external_id`` must not already correspond to a :class:`CmdbCloudScope` for this integration — would mean the next ``cmdb.scopes_sync`` sweep already imported it and admin should bind there instead of duplicating. * ``scope_type`` is admin-supplied (the cloud's natural scope kind: ``aws_account``, ``azure_subscription``, ``gcp_project``, ``aliyun_account``, ``standalone_iam``). Defaults to a per-provider canonical when omitted. """ req = await self._lock_request(req_id) if req.status not in ( SCOPE_REQUEST_STATUS_FEISHU_APPROVED, SCOPE_REQUEST_STATUS_FAILED, ): raise AppException( code="CLOUD_SCOPE_REQUEST_INVALID_STATE", message=( f"Cannot manually provision a request in {req.status!r} " "state — must be 'feishu_approved' or 'failed'." ), status_code=409, ) # 2026-05-12: if the request's provider has a working # ``driver.create_scope`` (auto-create capable, today only # AWS Organizations), the admin recovery path is "retry # Approve" — NOT "type in an external_id manually". Manually # provisioning an auto-capable provider is dangerous because: # the original ``approve()`` attempt may have already invoked # the cloud-side ``CreateAccount`` and the account exists on # the cloud side; an admin who types a fabricated external_id # here would bind Helix to a wrong / phantom row. Force the # admin to use the retry-Approve path, which will discover and # bind the already-created account via ``_provision``'s # idempotency contract (the AWS driver sweeps # ``list_create_account_status`` by AccountName to resume the # in-flight RequestId). if req.status == SCOPE_REQUEST_STATUS_FEISHU_APPROVED: from app.integrations.registry import get_driver_class driver_cls = get_driver_class("cloud", req.provider) auto_create_capable = ( driver_cls is not None and getattr(driver_cls, "create_scope", None) is not CloudDriver.create_scope ) if auto_create_capable: raise AppException( code="CLOUD_SCOPE_REQUEST_USE_RETRY_APPROVE", message=( f"Provider {req.provider!r} supports auto-create — " "click 'Retry Approve' instead of typing an " "external_id. The retry path is idempotent on the " "cloud side and will bind to the existing account " "if the previous attempt already created it." ), status_code=409, ) # Four-eyes guard: the requester (even when global_admin) cannot # be the one who manually provisions the cloud-side resource for # their own request. The auto-approve path already enforces # ``CLOUD_SCOPE_REQUEST_SELF_APPROVAL_FORBIDDEN``; this is the # equivalent check on the manual-rescue path so an admin who # asked for a scope cannot also bind their own ownership when # the auto path falls through to ``failed``. (A copy-paste of # this same check lived a few lines below until round-18 — it # was unreachable dead code; collapsed to a single guard.) if str(req.requester_person_id) == str(actor.id): raise AppException( code="CLOUD_SCOPE_REQUEST_SELF_PROVISION_FORBIDDEN", message=( "You cannot manually provision a cloud scope you " "requested. Ask another global admin to complete the " "rescue so the four-eyes review on cloud-account " "creation holds." ), status_code=403, ) external_id_clean = (external_id or "").strip() if not external_id_clean: raise AppException( code="CLOUD_SCOPE_REQUEST_EXTERNAL_ID_REQUIRED", message=( "external_id is required — paste the cloud-side " "account / subscription / project id IT just created." ), status_code=400, ) if len(external_id_clean) > 512: raise AppException( code="CLOUD_SCOPE_REQUEST_EXTERNAL_ID_TOO_LONG", message="external_id must be <= 512 chars", status_code=400, ) existing = await self.db.scalar( select(CmdbCloudScope).where( CmdbCloudScope.integration_id == req.integration_id, CmdbCloudScope.external_id == external_id_clean, CmdbCloudScope.deleted_at.is_(None), ) ) if existing is not None: raise AppException( code="CLOUD_SCOPE_ALREADY_EXISTS", message=( f"A scope with external_id {external_id_clean!r} already " "exists on this integration — bind there instead." ), status_code=409, ) requester = await self.db.get( IdentityPerson, uuid.UUID(req.requester_person_id), ) if requester is None: raise NotFoundError( "IdentityPerson", req.requester_person_id, ) # Refuse to bind a new scope to a leaver. If the requester # departed between submit and the IT admin's manual-provision # step, the sub-account has nobody actively responsible. # Force the admin to either reassign ownership (out-of-band) # or cancel + reopen the request under a current employee. if ( requester.deleted_at is not None or requester.employment_status in ("left", "leaver_pending") ): # Round-22 audit: widen the gate to include # ``leaver_pending`` so HR-flagged-as-leaving employees # can't be bound as scope owners. The platform-wide # ``leaver_pending`` semantics already lock them out of # asset / grant request flows (see # ``lifecycle_service.assign_scope_owner`` at # SCOPE_OWNER_REBIND_CLOSED_FORBIDDEN); manual_provision # must match for parity. raise AppException( code="CLOUD_SCOPE_REQUEST_REQUESTER_INACTIVE", message=( f"Cannot create scope owned by {requester.work_email or requester.id} — " "they are no longer active in HR. Cancel this request and " "have a current employee re-submit, or assign a different " "owner via the admin console once the scope exists." ), status_code=409, ) scope_type_clean = (scope_type or "").strip() or _default_scope_type( req.provider, ) # ``scope_type`` is free-form admin input (the body field is just # ``str | None``, max_length 32), and it is load-bearing far # beyond display: every per-scope driver fan-out — grants / # roles / billing_daily / cost_facts — filters its scope query # to ``TENANT_LEVEL_SCOPE_TYPES``, so a typo (``azure_subscriptions``) # or an invented value lands a row that is SILENTLY skipped by # cloud-access sync forever: no discovered grants, no role # catalog, no cost. A manually-provisioned sub-account is by # definition a tenant-level unit of access (the five values the # docstring above enumerates), so reject anything else at the # door rather than materialising an inert scope. 2026-07-30, # alongside the ``InvalidSubscriptionId`` grants_sync fix — that # incident is exactly what a scope_type the fan-out doesn't # understand costs, just from the other direction. if scope_type_clean not in TENANT_LEVEL_SCOPE_TYPES: raise AppException( code="CLOUD_SCOPE_TYPE_INVALID", message=( f"scope_type {scope_type_clean!r} is not a unit of " "cloud access; expected one of: " + ", ".join(sorted(TENANT_LEVEL_SCOPE_TYPES)) ), status_code=422, ) default_rules = await _build_default_alert_rules( self.db, float(req.monthly_budget_usd), ) scope = CmdbCloudScope( integration_id=req.integration_id, scope_type=scope_type_clean, external_id=external_id_clean, display_name=req.name, owner_person_id=str(requester.id), budget_usd_monthly=req.monthly_budget_usd, alert_rules=default_rules, status="active", raw_data={ "manual_provision": True, "scope_request_id": req.id, "provisioned_by": str(actor.id), }, notes=req.description, ) self.db.add(scope) await self.db.flush() req.status = SCOPE_REQUEST_STATUS_PROVISIONED req.created_scope_id = str(scope.id) req.provisioned_at = datetime.now(UTC) req.failure_reason = None await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor.id, action="cloud.scope_request.manual_provision", status="success", resource_type="cmdb_cloud_scope_request", resource_id=req.id, detail={ "scope_id": str(scope.id), "external_id": external_id_clean, "scope_type": scope_type_clean, }, ) # Same backfill as the auto path: a manual-provisioned request # reached here from the Feishu-callback approve (which leaves the # instance live, never cancelled), so surface the admin-supplied # account ID onto the approval timeline. await self._backfill_account_id_to_feishu( req=req, requester=requester, trace_id=trace_id, ) await self._notify_requester_provisioned(req, scope, trace_id) return req # ── AWS owner-alias ensure ─────────────────────────────────── async def _ensure_aws_owner_alias( self, *, req: CmdbCloudScopeRequest, requester: IdentityPerson, trace_id: str, ) -> str: """Derive + create (idempotently) the Google Workspace user-alias used as the AWS root-user email for this scope. Reuses an already-persisted alias from ``req.metadata_json`` if a previous attempt got past alias creation but failed downstream — so retries don't accumulate stray Google aliases. Failures (no GWS integration / GWS driver returns !ok and not the 409 ALREADY_EXISTS case) raise ``RuntimeError`` with a message suitable for ``failure_reason``: the caller's ``approve()`` puts the request in ``failed`` and an admin can either fix the GWS integration and Retry-Approve, or manual-provision past the alias step. We deliberately do NOT fall back to ``requester.work_email`` — that would silently hit AWS EMAIL_ALREADY_EXISTS on the second sub-account and produce a confusing "Organizations.CreateAccount rejected" message far from the real root cause. """ # Local imports to avoid a domain-level circular: this method # is only ever called on the AWS auto-provision hot path, so # the cost of in-function import is negligible. from app.integrations.drivers.google_workspace import ( GoogleWorkspaceDriver, ) from app.integrations.service import _resolve as _resolve_integration # Reuse if a prior attempt already persisted one. This bounds # retries: alias create is best-effort idempotent on the # Google side too, but reusing avoids racing parallel admin # retries against AWS's "AccountName-already-in-flight" path. meta = req.metadata_json or {} existing = (meta.get("owner_alias_email") or "").strip() derived = derive_aws_owner_alias(req.name) alias = existing or derived # Locate the primary enabled Google Workspace integration — # same lookup as ``google_workspace.account_provisioner``. gws_integration = (await self.db.execute( select(Integration).where( Integration.kind == "software_provisioning", Integration.provider == "google_workspace", Integration.enabled.is_(True), ).order_by( Integration.is_primary.desc(), Integration.created_at, ) )).scalars().first() if gws_integration is None: raise RuntimeError( "AWS sub-account requires a Google Workspace alias on " "the requester's mailbox, but no active Google Workspace " "integration is configured. Set one up under " "Integrations → Google Workspace and Retry-Approve.", ) target_email = (requester.work_email or "").strip() if not target_email: raise RuntimeError( f"requester {requester.id} has no work_email — cannot " "attach AWS owner alias", ) resolved = await _resolve_integration(self.db, gws_integration) driver = GoogleWorkspaceDriver( integration_id=str(gws_integration.id), config=resolved.config, ) try: result = await driver.add_alias( target_email=target_email, alias_email=alias, target_kind="user", ) finally: try: await driver.close() except Exception: # noqa: BLE001 pass if not result.ok: raise RuntimeError( f"Google Workspace add_alias failed for " f"{target_email!r} alias={alias!r}: {result.error_message}", ) # 409 ALREADY_EXISTS lands as ok=True + already_exists=True; # treat as success. # Persist on the request row so retries see it + audit can # tie scope → alias → requester later. Reassign the JSON # column rather than mutating in place — SQLAlchemy's JSON # change-tracking is reliable on assignment, fragile on # in-place mutate. new_meta = dict(meta) new_meta["owner_alias_email"] = alias req.metadata_json = new_meta await self.audit.log( trace_id=trace_id, actor_type="system", action="cloud.scope_request.owner_alias_ensured", status="success", resource_type="cmdb_cloud_scope_request", resource_id=req.id, detail={ "alias_email": alias, "target_email": target_email, "already_existed": bool(result.already_exists), "reused_persisted": bool(existing), }, ) return alias # ── Provisioning (internal) ───────────────────────────────── async def _provision( self, req: CmdbCloudScopeRequest, *, trace_id: str, ) -> CmdbCloudScope: """Call ``driver.create_scope`` and persist a CmdbCloudScope. Failures bubble up — caller (``approve``) translates them into ``status=failed + failure_reason`` so the request stays retryable. """ integration = await self.db.get( Integration, uuid.UUID(req.integration_id), ) if integration is None or integration.kind != "cloud": raise RuntimeError( f"integration {req.integration_id} vanished mid-flight", ) requester = await self.db.get( IdentityPerson, uuid.UUID(req.requester_person_id), ) if requester is None: raise RuntimeError( f"requester {req.requester_person_id} vanished mid-flight", ) # Leaver invariant (execute-time gate): a cloud-scope request can sit # pending in Feishu for days; if the requester offboards in that # window, approving it must NOT mint a real, billable AWS/cloud # sub-account owned by a departed employee. Every sibling # Feishu-approval execute path re-checks this at execute time # (ai_tool / announcement / apikey / sso / vpn / soc2 / data_access); # scope-request was the one flow missing it. Raising here lands the # request at status=failed with a clear reason instead of # provisioning infra post-departure. if requester.employment_status in ("left", "leaver_pending"): raise RuntimeError( f"requester {req.requester_person_id} is " f"'{requester.employment_status}' — refusing to provision a " "cloud scope for a departing/departed employee.", ) # Build a CloudScopeRef-like context to load the driver. Reuse # the lifecycle helper so credential loading + driver caching # follows the same code path as scope/grants sync. from app.domains.cmdb.lifecycle_service import _load_cloud_driver driver = await _load_cloud_driver(self.db, integration) # AWS sub-accounts need a globally-unique Email per AWS # Organizations rule (each member account's root user is # identified by email). We can't reuse ``requester.work_email`` # directly because the requester is already the root user on # the payer (or some prior sub-account) — AWS would reject # with EMAIL_ALREADY_EXISTS. Instead derive a per-scope # Google Workspace user-alias on the requester's mailbox # (``aws-@pingcap.com``) so AWS sees a fresh address # while mail still reaches the requester. Skip for non-AWS # providers — GCP / Azure / Aliyun all model "owner email" # differently (delegate the requester directly). provisioning_owner_email = requester.work_email or "" owner_alias_email: str | None = None if req.provider == "aws": owner_alias_email = await self._ensure_aws_owner_alias( req=req, requester=requester, trace_id=trace_id, ) provisioning_owner_email = owner_alias_email try: # ``idempotency_key`` is the request id (stable across retries # because the request row exists before approve fires). It # is accepted by the driver signature but AWS Organizations # CreateAccount has no native idempotency token — the AWS # driver achieves redrive safety via # ``list_create_account_status`` sweep by AccountName. The # kwarg is forwarded for drivers that DO support native # tokens (future GCP / Azure paths). info = await driver.create_scope( name=req.name, description=req.description, owner_email=provisioning_owner_email, budget_usd_monthly=float(req.monthly_budget_usd), metadata={ "scope_request_id": req.id, **( {"owner_alias_email": owner_alias_email} if owner_alias_email else {} ), }, idempotency_key=str(req.id)[:64], ) finally: try: await driver.close() except Exception: # noqa: BLE001 pass # Cloud-side create succeeded — audit-log the external_id # immediately, in a FRESH session that commits independently of # the caller's transaction. If the DB persist below fails (e.g. # UNIQUE-constraint race or a flush-time crash) the caller's # transaction will be rolled back by the raised exception, which # would also roll back an in-tx audit row written via # ``self.audit.log``. That'd leave an invisible orphan: a real # cloud account with no Helix-side breadcrumb tying it to this # request_id. Mirrors ``_audit_decommission_failure`` in # ``lifecycle_service.py`` (decommission failure path). try: async with async_session_factory() as audit_session: audit = AuditService(audit_session) await audit.log( trace_id=trace_id, actor_type="user", actor_user_id=uuid.UUID(req.requester_person_id), action="cloud.scope_request.cloud_created", status="success", resource_type="cmdb_cloud_scope_request", resource_id=req.id, detail={ "external_id": info.external_id, "scope_type": info.scope_type, "provider": req.provider, }, ) await audit_session.commit() except Exception as audit_exc: # noqa: BLE001 — best-effort logger.warning( "Out-of-tx audit write for scope_request %s cloud-created " "failed: %s — cloud account %s exists but breadcrumb missing", req.id, audit_exc, info.external_id, ) # Persist a CmdbCloudScope row with the requester as owner + # default alert rules (ditto sync_scopes_for_integration). The # next ``cmdb.scopes_sync`` cron run will pick it up and refresh # ``last_synced_at`` etc., but the row exists *now* so the # requester's grant-add doesn't 404 on missing scope. # # 2026-05-12 race-handling: ``approve()`` widened its state # guard to accept FEISHU_APPROVED on the admin recovery # path. If two admins click Retry in the crash window, both # workers reach this insert with the SAME ``(integration_id, # external_id)`` (because ``driver.create_scope`` is # idempotent on ``idempotency_key=req.id``). The UNIQUE index # on that pair raises IntegrityError on the loser. SELECT the # existing row instead so the loser converges to the same # final state as the winner — net outcome of the race is # one CmdbCloudScope row + a possibly-duplicated # ``cloud.scope_request.provision`` audit row + duplicated # ``_notify_requester_provisioned`` (acceptable: same # external_id, same scope_id, idempotent semantically). default_rules = await _build_default_alert_rules( self.db, float(req.monthly_budget_usd), ) scope = CmdbCloudScope( integration_id=str(integration.id), scope_type=info.scope_type, external_id=info.external_id, display_name=info.display_name or req.name, owner_person_id=str(requester.id), budget_usd_monthly=req.monthly_budget_usd, alert_rules=default_rules, status="active", raw_data=info.raw or {}, notes=req.description, ) self.db.add(scope) try: await self.db.flush() except IntegrityError as e: # Concurrent recovery race — the winning worker already # persisted the scope row. Roll back THIS worker's pending # insert and re-fetch the existing one. Do NOT bubble: # from the caller's perspective, _provision returns # success and approve() lands the row at PROVISIONED. await self.db.rollback() logger.info( "scope_request %s _provision UNIQUE-collision on " "(integration_id=%s, external_id=%s) — recovering " "the existing CmdbCloudScope row (concurrent admin retry)", req.id, integration.id, info.external_id, ) scope = await self.db.scalar( select(CmdbCloudScope).where( CmdbCloudScope.integration_id == str(integration.id), CmdbCloudScope.external_id == info.external_id, ) ) if scope is None: # Shouldn't happen — the IntegrityError fired because # the row existed. If it's gone now (e.g. concurrent # decommission), let the original exception surface # so the admin sees a clear failure rather than a # confusing None. raise e # The post-provision FastAPI BackgroundTask # (``lifecycle_router._post_provision_finalize``) handles # both the role-catalog pre-warm AND the auto-Owner grant — # see ``_DEFAULT_OWNER_ROLE_BY_PROVIDER`` for the per-provider # mapping that drives the grant. We deliberately don't # in-line that work here: # * ``driver.grant_access`` is a cloud-side call; running # it inside ``_provision`` would extend the approve # latency by another round-trip + a setIamPolicy that # itself races IAM propagation. # * The bg task acquires the per-integration # ``cloud_sync`` Redis lock so the role pre-warm + the # grant write don't race the cron-driven # ``run_scopes_sync`` / ``run_roles_sync`` / # ``run_grants_sync`` mark-and-sweep. # * Failure is best-effort: a driver crash on grant_access # leaves a PENDING ``CmdbCloudAccessGrant`` row that the # requester can re-attempt via the standard owner-add # flow, instead of marking the scope_request itself # failed and rolling back a real cloud-side project. # # The next ``cmdb.grants_sync`` will discover any # provider-side roles the create flow happened to bind and # back-fill them with ``source='discovered'`` — also accurate # for the natural-success path. return scope # ── Helpers ───────────────────────────────────────────────── async def _lock_request(self, req_id: str) -> CmdbCloudScopeRequest: req = await self.db.scalar( select(CmdbCloudScopeRequest) .where(CmdbCloudScopeRequest.id == req_id) .with_for_update() ) if req is None: raise NotFoundError("CmdbCloudScopeRequest", req_id) return req def _scope_request_context( self, req: CmdbCloudScopeRequest, *, requester: IdentityPerson | None = None, approver: IdentityPerson | None = None, scope: CmdbCloudScope | None = None, ) -> dict[str, object]: """Build the variable dict shared across every scope-request notification template. Centralised so all four notifications bind the same names — flips to en_US automatically because the NotificationService picks the en variant per recipient. """ return { "requester_name": ( (requester.display_name or requester.work_email) if requester else "" ), "requester_email": requester.work_email if requester else "", "approver_name": ( (approver.display_name or approver.work_email) if approver else "" ), "provider": req.provider, "provider_upper": req.provider.upper(), "scope_name": req.name, "description": req.description or "(无 / n/a)", "budget": float(req.monthly_budget_usd), "external_id": scope.external_id if scope else "", "reason": req.rejection_reason or "(未填写 / not provided)", } # NOTE 2026-05-12: ``_notify_admins_pending`` was removed. Feishu's # native approval-card push covers the approver inbox; the Helix # duplicate push pointed at /admin/cloud/scopes which the user # policy ("审批一律在飞书做") explicitly disallows as an approval # surface. The ``cloud.scope_request.pending`` notification # template rows in DB (migrations 071, 104) are now unused but # left in place — removing them would require a fresh migration # for no behavior change. Re-introducing the push would mean # restoring this method and calling it from ``submit`` after the # Feishu instance is created. async def _notify_admins_manual_required( self, req: CmdbCloudScopeRequest, approver: IdentityPerson, trace_id: str, ) -> None: """Approver-triggered notice that ``approve`` stalled at the driver-capability gate — IT must create the sub-account in the cloud console manually and call back via ``POST /scope-requests/{id}/manual-provision``. Recipients are the ``global_admin`` cohort (same as ``_notify_admins_pending`` — same group that approves).""" from app.auth.models import AuthRole, AuthUserRole 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()) ctx = self._scope_request_context(req, approver=approver) baseline_title = ( f"[需手动开通] {ctx['provider_upper']} 子账号 {ctx['scope_name']}" ) for aid in admin_ids: person = await self.db.get(IdentityPerson, aid) if person is None or not person.work_email: continue try: await self.notify.dispatch( NotificationEnvelope( recipient=person, title=baseline_title, body_text=baseline_title, template_key="cloud.scope_request.manual_required", context=ctx, url=f"/admin/cloud/scopes?tab=requests&id={req.id}", channels=["feishu", "email"], related_resource_type="cmdb_cloud_scope_request", related_resource_id=req.id, ), trace_id=trace_id, ) except Exception: # noqa: BLE001 logger.exception( "scope-request notify-manual failed: aid=%s req=%s", aid, req.id, ) async def _notify_requester_rejected( self, req: CmdbCloudScopeRequest, trace_id: str, ) -> None: person = await self.db.get( IdentityPerson, uuid.UUID(req.requester_person_id), ) if person is None or not person.work_email: return ctx = self._scope_request_context(req) baseline_title = f"[已驳回] 云子账号申请 {ctx['scope_name']}" try: await self.notify.dispatch( NotificationEnvelope( recipient=person, title=baseline_title, body_text=baseline_title, template_key="cloud.scope_request.rejected", context=ctx, url=f"/my/cloud/scope-requests?id={req.id}", channels=["feishu", "email"], related_resource_type="cmdb_cloud_scope_request", related_resource_id=req.id, ), trace_id=trace_id, ) except Exception: # noqa: BLE001 logger.exception("scope-request notify rejection failed") async def _notify_requester_provisioned( self, req: CmdbCloudScopeRequest, scope: CmdbCloudScope, trace_id: str, ) -> None: person = await self.db.get( IdentityPerson, uuid.UUID(req.requester_person_id), ) if person is None or not person.work_email: return ctx = self._scope_request_context(req, scope=scope) baseline_title = f"[已开通] 云子账号 {ctx['scope_name']}" try: await self.notify.dispatch( NotificationEnvelope( recipient=person, title=baseline_title, body_text=baseline_title, template_key="cloud.scope_request.provisioned", context=ctx, url=f"/cloud/scopes/{scope.id}", channels=["feishu", "email"], related_resource_type="cmdb_cloud_scope", related_resource_id=scope.id, ), trace_id=trace_id, ) except Exception: # noqa: BLE001 logger.exception("scope-request notify provisioned failed") __all__ = ["CloudScopeRequestService"]