"""Aliyun Bailian auto-grant — called from the API key Feishu callback. When an :class:`ApiKeyAssignment` whose provider name contains "bailian" / "百炼" / "dashscope" / "model studio" transitions to ``approved`` via the Feishu callback, we: 1. Resolve the AliCloud International cloud integration that owns the workspace (slug ``aliyun-bailian`` or the ``cloud.bailian.provisioning_integration_slug`` platform setting). Falls back to the first enabled Aliyun cloud integration whose name / slug contains "bailian" so the most common mono-tenant setup works without configuration. 2. Mint a dedicated RAM user named ``bailian-{local}-{suffix}`` and attach :data:`AliyunBailianFullAccess` (system policy). The RAM user is *never* given an AccessKey — its job is purely to anchor audit / ownership in ``cmdb_iam_users`` and the workspace member list. Employee-facing inference auth goes through the sk- API key minted below, not through AK/SK. 3. Resolve the default Bailian workspace (configured via ``cloud.bailian.default_workspace_id`` or the ``is_default`` flag on a synced workspace). 4. Call ``ModelStudio CreateApiKey`` against that workspace, using the master integration's AK, to mint a *real DashScope API key* (sk-...). The plaintext value is only ever exposed once at creation — we capture it immediately and write the encrypted bundle onto ``api_key_assignments.key_value``. 5. Mirror the new sk-... handle into ``bailian_api_keys`` so the admin Bailian page shows the key without waiting for the nightly ``bailian.workspace_sync`` cron. History (2026-05-18): an earlier revision handed the employee the RAM user's AccessKey/Secret pair and a help-doc snippet claiming the DashScope SDK accepts ``dashscope.access_key_id`` / ``dashscope.access_key_secret``. The SDK does *not* — its ``_get_api_key()`` only consults ``dashscope.api_key`` / ``DASHSCOPE_API_KEY`` and throws ``No api key provided`` for everything else. The fix here switches the user-facing credential to a real sk- API key (minted via ModelStudio's intl ``CreateApiKey`` action, which *is* programmatically available — the prior "console-only" claim only ever applied to the 国内 product). RAM AK/SK is dropped entirely. Errors leave the assignment in ``approved`` status with a note in ``detail.bailian`` so an admin can still fill the key manually via the existing fill-key flow. """ from __future__ import annotations import uuid from dataclasses import dataclass from datetime import UTC, datetime from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import AppException from app.core.safe_logging import get_safe_logger from app.domains.apikey.autogrant_alerts import record_autogrant_failure from app.domains.apikey.models import ( ApiKeyAssignment, ApiKeyProvider, normalize_provider_match_name, ) from app.domains.apikey.service import _encrypt_key_value from app.domains.bailian.client import ( _BAILIAN_FULL_ACCESS_POLICY, _SUPPORTED_PROVIDERS, BailianClient, dashscope_endpoint_for_provider, default_region_for_provider, resolve_credentials, ) from app.domains.bailian.models import BailianApiKey, BailianWorkspace from app.domains.cmdb.iam_upsert import upsert_local_iam_user from app.identity.person_models import IdentityPerson from app.integrations.models import Integration, PlatformSetting from app.notifications.service import NotificationEnvelope, NotificationService logger = get_safe_logger(__name__) def _is_bailian_provider(provider: ApiKeyProvider | None) -> bool: """Return True for any provider name pointing at Aliyun Bailian. The user-facing name has gone through a few iterations — the original seed used ``Aliyun Bailian (DashScope)`` but admins often rename to ``Alibaba Cloud Model Studio`` to match Aliyun's English product name. We accept any of the canonical tokens so a one-off rename in the admin console doesn't silently break the auto-grant routing. ``normalize_provider_match_name`` collapses whitespace runs (incl. NBSP / full-width / tabs) so admin typos like ``"Alibaba Cloud Model Studio"`` survive — same bug class hit Azure OpenAI in prod (AK-20260509-EF8A). """ if provider is None: return False name = normalize_provider_match_name(provider.name) return any( token in name for token in ( "bailian", "百炼", "dashscope", "model studio", ) ) async def _resolve_integration(db: AsyncSession) -> Integration | None: """Find the AliCloud International cloud integration to provision the RAM user in. Bailian only runs on the 海外 tenant — the 国内 ``aliyun`` provider is intentionally excluded. """ override = await db.get( PlatformSetting, "cloud.bailian.provisioning_integration_slug", ) if override is not None and override.value: slug = str(override.value).strip() row = (await db.execute( select(Integration).where( Integration.slug == slug, Integration.enabled.is_(True), Integration.kind == "cloud", Integration.provider.in_(_SUPPORTED_PROVIDERS), ) )).scalar_one_or_none() if row is not None: return row # Heuristic: integrations whose slug or name contains "bailian". rows = (await db.execute( select(Integration).where( Integration.kind == "cloud", Integration.provider.in_(_SUPPORTED_PROVIDERS), Integration.enabled.is_(True), ) )).scalars().all() for r in rows: if "bailian" in (r.slug or "").lower(): return r for r in rows: if "bailian" in (r.name or "").lower(): return r # Final fallback: a single enabled Aliyun integration. Multi-tenant # setups must configure the platform_setting override above to # disambiguate; we don't silently pick "the first one" because that # would route Bailian grants to the wrong tenant in a multi-account # org. if len(rows) == 1: return rows[0] return None def _sanitize_user_name(email: str, request_number: str) -> str: """RAM user name: [A-Za-z0-9_.-], max 64 chars.""" local = (email or "").split("@")[0] or "bailian-user" local = "".join(c for c in local if c.isalnum() or c in "._-") if not local: local = "bailian-user" parts = request_number.split("-", 1) suffix = parts[1] if len(parts) > 1 else request_number name = f"bailian-{local}-{suffix}" return name[:64] @dataclass(frozen=True) class WorkspaceResolution: """Outcome of default-workspace resolution. ``workspace is None`` means nothing could be picked *deterministically* — never "there was nothing to pick". ``reason`` + ``candidates`` distinguish the two for the operator-facing error, which is the whole point: the 2026-08-19 outage (AK-20260819-346E) surfaced as a generic "no default workspace" while three perfectly healthy workspaces sat in the table, and the admin had no way to tell which knob to turn. """ workspace: BailianWorkspace | None reason: str candidates: tuple[BailianWorkspace, ...] = () preferred_region: str | None = None def describe_candidates(self) -> str: """``ws-a(cn-beijing), ws-b(us-east-1)`` — for the error text.""" if not self.candidates: return "none synced" return ", ".join( f"{ws.workspace_id}({ws.region or 'region?'})" for ws in self.candidates ) async def resolve_default_workspace( db: AsyncSession, integration_id: str, *, preferred_region: str | None = None, ) -> WorkspaceResolution: """Pick the workspace to mint the sk- API key in. Resolution order: 1. ``cloud.bailian.default_workspace_id`` platform setting matching a non-deleted ``BailianWorkspace.workspace_id``. A configured id that matches nothing stops here (``platform_setting_unmatched``) rather than falling through — an admin who pinned a workspace that has since been deleted must hear about it, not silently get a key minted in some other project. 2. The workspace flagged ``is_default=True`` for this integration. Two or more flagged ⇒ ambiguous, stop. 3. **The workspace in the integration's Bailian region.** Added 2026-08-19. ModelStudio is per-region and each region carries its own 「默认业务空间」, so when the per-region sync (PR #2772) landed on 2026-08-15 this table went from one row to three and rule 4 below stopped firing — every Bailian auto-grant died from that day until the first casualty was noticed four days later. The integration's own Bailian region is the value every write call (``create_api_key``, rotation, revoke) already routes on, so converging on it reproduces exactly the workspace the pre-#2772 grants used: deterministic, and needs no configuration. 4. The single workspace, if there is only one. 5. Nothing — the caller fails loud. """ candidates = tuple((await db.execute( select(BailianWorkspace).where( BailianWorkspace.integration_id == integration_id, BailianWorkspace.deleted_at.is_(None), ).order_by(BailianWorkspace.workspace_id) )).scalars().all()) def _out(ws: BailianWorkspace | None, reason: str) -> WorkspaceResolution: return WorkspaceResolution( workspace=ws, reason=reason, candidates=candidates, preferred_region=preferred_region, ) setting = await db.get( PlatformSetting, "cloud.bailian.default_workspace_id", ) if setting is not None and setting.value: wanted = str(setting.value).strip() for ws in candidates: if ws.workspace_id == wanted: return _out(ws, "platform_setting") return _out(None, "platform_setting_unmatched") flagged = [ws for ws in candidates if ws.is_default] if len(flagged) == 1: return _out(flagged[0], "is_default") if len(flagged) > 1: return _out(None, "ambiguous_is_default") if preferred_region: in_region = [ ws for ws in candidates if (ws.region or "").strip() == preferred_region ] if len(in_region) == 1: return _out(in_region[0], "preferred_region") if len(in_region) > 1: return _out(None, "ambiguous_region") if len(candidates) == 1: return _out(candidates[0], "single") if not candidates: return _out(None, "none_synced") return _out(None, "ambiguous") async def _fail( db: AsyncSession, assignment: ApiKeyAssignment, detail: dict[str, Any], bailian_info: dict[str, Any], *, stage: str, error: str, provider_row: ApiKeyProvider | None, ) -> dict[str, Any]: """Land a failed auto-grant: stamp the row, commit, alert IT. Every failure branch goes through here so none of them can drift back to being silent. The commit is what makes ``detail.bailian.error`` survive — several branches used to leave the stamp in a session the caller might never flush — and it happens *before* the alert so an admin clicking through from the card sees the same error it quotes. """ bailian_info["granted"] = False bailian_info["in_flight"] = False bailian_info["error"] = error[:500] bailian_info["failed_stage"] = stage bailian_info["failed_at"] = datetime.now(UTC).isoformat() detail["bailian"] = bailian_info assignment.detail = detail await db.commit() await db.refresh(assignment) await record_autogrant_failure( db, assignment, kind="bailian", stage=stage, error=error, provider_label=getattr(provider_row, "name", None), ) return {"status": stage} async def maybe_auto_grant_bailian( db: AsyncSession, assignment: ApiKeyAssignment, ) -> dict[str, Any]: """Invoked when an assignment becomes ``approved``. Safe no-op when the provider isn't Bailian or when auto-grant has already run (idempotent on ``detail.bailian.granted`` / ``detail.bailian.in_flight``). Mirrors the Bedrock contract: the in-flight tombstone is committed before any Aliyun call so a mid- callback crash refuses to mint a second RAM user. """ provider_row = await db.get(ApiKeyProvider, assignment.provider_id) if not _is_bailian_provider(provider_row): return {"status": "not-bailian"} detail = dict(assignment.detail or {}) bailian_info = dict(detail.get("bailian") or {}) if bailian_info.get("granted"): return {"status": "already-granted"} if bailian_info.get("in_flight"): return {"status": "in-flight-needs-manual-recovery"} integration = await _resolve_integration(db) if integration is None: return await _fail( db, assignment, detail, bailian_info, stage="no-integration", error=( "No enabled Aliyun cloud integration found. Set " "platform_setting cloud.bailian.provisioning_integration_slug " "or create an integration whose slug contains 'bailian'." ), provider_row=provider_row, ) person = await db.get(IdentityPerson, assignment.assigned_to) email = (person.work_email or "") if person else "" user_name = _sanitize_user_name(email, assignment.request_number) try: creds = resolve_credentials(integration) client = BailianClient(creds) except AppException as exc: return await _fail( db, assignment, detail, bailian_info, stage="client-init-failed", error=f"Bailian client init failed: {exc.message}", provider_row=provider_row, ) # Pre-flight tombstone (durable across mid-callback crash). bailian_info["in_flight"] = True bailian_info["user_name_attempted"] = user_name bailian_info["integration_id"] = str(integration.id) bailian_info["integration_slug"] = integration.slug detail["bailian"] = bailian_info assignment.detail = detail await db.commit() await db.refresh(assignment) detail = dict(assignment.detail or {}) bailian_info = dict(detail.get("bailian") or {}) # 1. Create or fetch RAM user. try: user_resp = await client.ensure_ram_user( user_name=user_name, display_name=( person.display_name if person else user_name ) or user_name, email=email or None, comments=( f"Bailian auto-grant for assignment " f"{assignment.request_number} ({email or 'unknown'})" ), ) except AppException as exc: return await _fail( db, assignment, detail, bailian_info, stage="ram-user-create-failed", error=f"RAM CreateUser failed: {exc.message}", provider_row=provider_row, ) ram_user = user_resp.get("user") or {} ram_user_id = str(ram_user.get("UserId") or "") ram_user_arn = f"acs:ram::*:user/{user_name}" # CreateDate from Aliyun RAM is an ISO-8601 string; tolerate both # ISO and pre-parsed datetime so the helper works whether we got a # freshly-created or pre-existing user row. raw_create_date = ram_user.get("CreateDate") parsed_create_date: datetime | None = None if isinstance(raw_create_date, datetime): parsed_create_date = raw_create_date elif isinstance(raw_create_date, str) and raw_create_date: try: parsed_create_date = datetime.fromisoformat( raw_create_date.replace("Z", "+00:00"), ) except ValueError: # Don't swallow silently — the next IAM sync will refill # ``create_date`` from RAM's ListUsers, but a logged warning # here surfaces a future Aliyun shape change (RFC1123, # epoch, renamed field) before the dashboard mysteriously # shows half the autogrant rows with empty create_date. logger.warning( "Bailian RAM CreateDate unparseable for %s: %r — " "row upserts with create_date=None; next IAM sync will " "refill.", user_name, raw_create_date, ) parsed_create_date = None # 2. Attach AliyunBailianFullAccess (governance / audit only — the # RAM user is never given an AccessKey, so this policy isn't on # a hot auth path; it just makes the cmdb_iam_users row meaningful # and leaves a future-extensible identity that an admin can mint # a console password against if needed). try: await client.attach_policy( user_name=user_name, policy_name=_BAILIAN_FULL_ACCESS_POLICY, policy_type="System", ) except AppException as exc: return await _fail( db, assignment, detail, bailian_info, stage="attach-policy-failed", error=f"AttachPolicyToUser failed: {exc.message}", provider_row=provider_row, ) # 3. Resolve workspace. CreateApiKey is workspace-scoped on intl # ModelStudio — we MUST have one. Fail loud (admin sees the # explicit error on the assignment row) rather than silently # issuing a key against the wrong project. Previous behaviour # here was to keep going with ``workspace_member_status= # covered_by_policy`` because the old AK/SK bundle didn't need # a workspace; the sk- flow does. # Region convergence (2026-08-19): ModelStudio is per-region and # each region has its own 「默认业务空间」, so the integration's own # Bailian region is what disambiguates the now-plural workspace # list — the same value ``create_api_key`` below routes on. preferred_region = ( creds.get("bailian_region") or default_region_for_provider(integration.provider) ) resolution = await resolve_default_workspace( db, str(integration.id), preferred_region=preferred_region, ) workspace = resolution.workspace if workspace is None: return await _fail( db, assignment, detail, bailian_info, stage="no-default-workspace", error=( "Bailian autogrant cannot mint an sk- API key: no default " "workspace resolved for integration " f"{integration.slug or integration.id} " f"(reason={resolution.reason}, " f"preferred_region={preferred_region}, " f"candidates=[{resolution.describe_candidates()}]). " "Pick one on the 百炼 admin page (工作空间 → 设为默认), or " "set platform_setting cloud.bailian.default_workspace_id, " "then re-run the autogrant via the admin fill-key flow." ), provider_row=provider_row, ) bailian_info["workspace_resolved_by"] = resolution.reason # 4. Mint the real DashScope API key (sk-...). The plaintext is only # exposed in this CreateApiKey response — we capture it inline # and never round-trip it through ListApiKeys (which masks). # Description encodes the Helix linkage so the Bailian console # surfaces "who owns this key" at a glance. description = ( f"Helix · {assignment.request_number} · {user_name}" ) try: api_key_resp = await client.create_api_key( workspace_id=workspace.workspace_id, # ModelStudio 分区域:打错区域会在错误的区域建 Key。 region=workspace.region, description=description, ) except AppException as exc: return await _fail( db, assignment, detail, bailian_info, stage="create-apikey-failed", error=f"CreateApiKey failed: {exc.message}", provider_row=provider_row, ) # 5. Persist credentials onto the assignment. # Same trap as BailianClient: integration.config_public.region is the # SSO/CMDB region (often cn-shanghai on 海外 tenants), NOT a valid # Bailian region. ``preferred_region`` above already resolved the # explicit ``bailian_region`` override / provider default — reuse it # so the bundle handed to the employee can never name a different # region than the one the workspace was resolved against. region = preferred_region dashscope_endpoint = dashscope_endpoint_for_provider(integration.provider) # New bundle shape (2026-05-18 fix): API_KEY is the only auth # credential the employee needs. The DashScope SDK reads # ``dashscope.api_key`` / ``DASHSCOPE_API_KEY`` and ignores any # ``access_key_id`` / ``access_key_secret`` attrs — the prior AK/SK # bundle threw ``No api key provided`` at runtime. WORKSPACE_ID, # ENDPOINT and REGION stay informational; API_KEY_ID + RAM_USER_NAME # are surfaced for transparency (and so admins can correlate the # row in the Bailian admin page during incident review). plaintext_bundle = ( f"API_KEY={api_key_resp['api_key_value']}\n" f"WORKSPACE_ID={workspace.workspace_id}\n" f"DASHSCOPE_ENDPOINT={dashscope_endpoint}\n" f"REGION={region}\n" f"API_KEY_ID={api_key_resp['api_key_id']}\n" f"RAM_USER_NAME={user_name}\n" f"NOTE=Use as `dashscope.api_key = \"\"` with the " f"DashScope Python SDK, or export DASHSCOPE_API_KEY= " f"for CI / shell. The RAM_USER_NAME is the audit-side identity " f"only — it has no AccessKey/Secret and no console password. " f"See Helix /help#bailian-employee for the full setup." ) # Encrypt at rest for parity with bedrock_autogrant. The reveal # endpoint decrypts on the way out; legacy plaintext rows still # decode via ``_decrypt_key_value``'s passthrough path. # Round-20 audit (A2F6): bind AES-GCM AAD to the owning row's id # so an attacker with DB-write can't ciphertext-swap one row onto # another. Mirror Azure OpenAI / Bedrock pattern. assignment.key_value = _encrypt_key_value( plaintext_bundle, assignment_id=assignment.id, ) assignment.status = "active" assignment.iam_user_arn = ram_user_arn assignment.iam_integration_id = str(integration.id) # 4b. Upsert the freshly-created RAM user into ``cmdb_iam_users`` so # the IAM 用户 page reflects it immediately rather than waiting # for the next scheduled IAM sync. Idempotent against the sync # job's UQ key — a subsequent sync collapses cleanly onto this # row. We tag ``__source=helix_autogrant.bailian`` so admins # auditing the IAM page can tell at a glance which rows came # from Helix-driven autogrant vs. cloud-side admin actions. iam_row = await upsert_local_iam_user( db, integration_id=str(integration.id), user_name=user_name, arn=ram_user_arn, user_id_external=ram_user_id or None, create_date=parsed_create_date, attached_policies=[ {"PolicyName": _BAILIAN_FULL_ACCESS_POLICY, "PolicyType": "System"}, ], tags={ "__source": "helix_autogrant.bailian", "helix-assignment-id": str(assignment.id), "helix-request-number": assignment.request_number, "helix-owner-email": email or "", "helix-project-name": (assignment.project_name or "")[:256], }, raw_data={ "DisplayName": (person.display_name if person else user_name) or user_name, "UserName": user_name, "UserId": ram_user_id or "", "Email": email or "", }, owner_person_id=( str(assignment.assigned_to) if assignment.assigned_to else None ), ) assignment.iam_user_id = iam_row.id bailian_info.update({ "granted": True, "in_flight": False, "integration_id": str(integration.id), "integration_slug": integration.slug, "integration_provider": integration.provider, "ram_user_name": user_name, # Mirror the RAM user name onto the canonical key the apikey # service revoke path reads. Prior to the 2026-05-18 fix the # bailian branch only stamped ``ram_user_name`` while # service._revoke_cloud_resources looked up ``iam_user_name`` # — meaning Bailian assignments NEVER had their RAM user # cleaned up on revoke. Bedrock + Azure OpenAI already use # ``iam_user_name``; align here. "iam_user_name": user_name, "ram_user_id": ram_user_id, "ram_user_created": user_resp.get("created"), "policy_name": _BAILIAN_FULL_ACCESS_POLICY, "workspace_id": workspace.workspace_id, # The sk-... API key's numeric Aliyun handle. Required by the # revoke path (``BailianClient.delete_api_key``) to invalidate # the key before tearing down the RAM user. "api_key_id": api_key_resp["api_key_id"], "api_key_created_by": api_key_resp.get("created_by"), "api_key_description": description, "region": region, "dashscope_endpoint": dashscope_endpoint, "granted_at": datetime.now(UTC).isoformat(), }) detail["bailian"] = bailian_info assignment.detail = detail # 6. Mirror into bailian_api_keys for sync visibility. Unlike the # legacy AK/SK era — where the only handle was the RAM AccessKey # ID — we now have a real ModelStudio ``apiKeyId``, which # matches what the nightly ``bailian.workspace_sync`` cron # upserts. The UQ key on (integration_id, api_key_id) means the # sync collapses cleanly onto this row instead of inserting a # duplicate. bailian_key_row = BailianApiKey( id=uuid.uuid4(), integration_id=str(integration.id), workspace_pk=workspace.id, api_key_id=api_key_resp["api_key_id"], display_name=description, owner_person_id=( str(assignment.assigned_to) if assignment.assigned_to else None ), ram_user_name=user_name, assignment_id=assignment.id, status="active", raw_data={ "source": "auto-grant", "request_number": assignment.request_number, "created_by": api_key_resp.get("created_by"), }, ) db.add(bailian_key_row) try: await db.commit() await db.refresh(assignment) except Exception: # noqa: BLE001 — log + re-raise logger.exception( "Bailian auto-grant: ModelStudio API key created " "(apiKeyId=%s, ram_user=%s) but DB commit failed. Manual " "reconciliation required — see cmdb-ops.md in-flight " "recovery section. The sk- key remains active in Bailian " "until an admin runs DeleteApiKey.", api_key_resp.get("api_key_id"), user_name, ) raise # Notify the assignee that the credential is ready. Done after the # main commit so a notification-channel failure (Feishu rate-limited # / SMTP down) cannot roll back the durable AWS+DB state. Dispatch # itself is durable via the notification_outbox (retry worker picks # up partial / failed sends), so we don't block the autogrant return # on the channel result. Skipped only when the assignee row has been # deleted between approval and callback — extremely rare; the audit # log already records the autogrant succeeded. if person is not None: provider_label = ( provider_row.name if provider_row is not None else "Bailian" ) try: await NotificationService(db).dispatch( NotificationEnvelope( recipient=person, title="[Helix] Bailian API Key 已就绪 / API key ready", body_text=( f"Hi {person.display_name or person.work_email or ''},\n\n" f"你申请的 {provider_label} API Key (申请号 " f"{assignment.request_number},项目 " f"{assignment.project_name or '-'}) 已自动开通。\n\n" f"请到 Helix 的「我的 API Keys」页面查看凭证和使用方法:\n" f" /me/api-keys\n\n" f"这是一把 DashScope API Key(sk- 开头),直接配进 " f"`dashscope.api_key` 或 `DASHSCOPE_API_KEY` " f"环境变量即可调模型。详细使用手册见: " f"/help#bailian-employee\n\n" f"— Helix IT 平台\n" f"---\n" f"Hi {person.display_name or person.work_email or ''},\n\n" f"Your {provider_label} API key (request " f"{assignment.request_number}, project " f"{assignment.project_name or '-'}) is now active.\n\n" f"View the credential and usage notes on your " f"Helix \"My API Keys\" page:\n" f" /me/api-keys\n\n" f"This is a DashScope API key (sk-...). Plug it " f"into ``dashscope.api_key`` or the " f"``DASHSCOPE_API_KEY`` env var and you're " f"ready to call the models. Full SDK setup at " f"/help#bailian-employee.\n\n" f"— Helix IT Platform" ), template_key="apikey.ready", context={ "person_name": ( person.display_name or person.work_email or "" ), "person_email": person.work_email or "", "provider_name": provider_label, "request_number": assignment.request_number, "project_name": assignment.project_name or "", "ram_user_name": user_name, "portal_url": "/me/api-keys", "help_anchor": "/help#bailian-employee", }, url="/me/api-keys", related_resource_type="api_key_assignment", related_resource_id=assignment.id, ), ) except Exception: # noqa: BLE001 — never fail the grant on notify error logger.exception( "Bailian auto-grant: notification dispatch failed for " "assignment %s; credential remains active and is visible " "on /me/api-keys.", assignment.request_number, ) logger.info( "Bailian auto-grant succeeded: assignment=%s ram_user=%s " "api_key_id=%s workspace=%s", assignment.id, user_name, api_key_resp["api_key_id"], workspace.workspace_id, ) return { "status": "granted", "ram_user_name": user_name, "api_key_id": api_key_resp["api_key_id"], "workspace_id": workspace.workspace_id, } async def reissue_bailian_api_key( db: AsyncSession, assignment: ApiKeyAssignment, ) -> dict[str, Any]: """Re-mint the DashScope sk- key on an already-granted Bailian assignment. Use this for assignments that were granted under the legacy AK/SK bundle (pre-2026-05-18) and need to be migrated to a real sk- key, or to rotate a leaked sk- without going through the full revoke + re-apply path. Contract: * ``assignment.detail.bailian.granted`` must already be ``True`` (otherwise call :func:`maybe_auto_grant_bailian` instead). * Re-uses the existing ``integration_id`` recorded on the prior autogrant so the new key lives in the same Aliyun tenant as the legacy bundle. Falls back to the default-resolution heuristic only if no integration was recorded (very old rows). * Re-uses the existing ``workspace_id`` if present; otherwise resolves the default workspace exactly like :func:`maybe_auto_grant_bailian`. * Mints a brand-new sk- via ``CreateApiKey``, writes a fresh bundle (``API_KEY=...``) to ``assignment.key_value``, and replaces ``detail.bailian.api_key_id`` with the new handle. The previous ``api_key_id`` (which for legacy rows was the defunct RAM AccessKeyId, NOT an sk- handle) is preserved under ``detail.bailian.legacy_api_key_id`` for audit. * Inserts a new ``bailian_api_keys`` row with the real ModelStudio handle. Legacy rows whose ``api_key_id`` was a defunct RAM AccessKeyId stay untouched — the nightly Bailian sync will mark them ``deleted_at`` once they fall out of ``ListApiKeys`` (which they never matched in the first place). Does NOT touch the companion RAM sub-account or its (now unused) legacy AccessKeys. The next revoke pass deletes them transitively via :func:`ApiKeyService._dispatch_cloud_revoke`. Idempotency: this method is NOT idempotent on the Aliyun side — each call mints a new sk-. Callers must guard with an admin-only HTTP endpoint + a confirm dialog. Re-running burns an extra slot in the Bailian workspace's API-key budget but is otherwise safe. """ provider_row = await db.get(ApiKeyProvider, assignment.provider_id) if not _is_bailian_provider(provider_row): raise AppException( code="BAILIAN_REISSUE_WRONG_PROVIDER", message=( "reissue_bailian_api_key called on a non-Bailian " f"assignment (provider={provider_row.name if provider_row else None!r})." ), status_code=400, ) detail = dict(assignment.detail or {}) bailian_info = dict(detail.get("bailian") or {}) if not bailian_info.get("granted"): raise AppException( code="BAILIAN_REISSUE_NOT_GRANTED", message=( "reissue requires the assignment to already be granted " "via the legacy or new autogrant flow. Run " "maybe_auto_grant_bailian first." ), status_code=400, ) integ_id = ( bailian_info.get("integration_id") or (str(assignment.iam_integration_id) if assignment.iam_integration_id else None) ) integration: Integration | None = None if integ_id: integration = await db.get(Integration, uuid.UUID(integ_id)) if integration is None: integration = await _resolve_integration(db) if integration is None: raise AppException( code="BAILIAN_REISSUE_NO_INTEGRATION", message=( "Cannot resolve a Bailian-capable Aliyun integration to " "reissue the key against. Configure one under " "/admin/integrations or set platform_setting " "cloud.bailian.provisioning_integration_slug." ), status_code=503, ) workspace: BailianWorkspace | None = None existing_ws_id = bailian_info.get("workspace_id") if existing_ws_id: workspace = (await db.execute( select(BailianWorkspace).where( BailianWorkspace.integration_id == str(integration.id), BailianWorkspace.workspace_id == str(existing_ws_id), BailianWorkspace.deleted_at.is_(None), ) )).scalar_one_or_none() creds = resolve_credentials(integration) region = ( creds.get("bailian_region") or default_region_for_provider(integration.provider) ) if workspace is None: resolution = await resolve_default_workspace( db, str(integration.id), preferred_region=region, ) workspace = resolution.workspace if workspace is None: raise AppException( code="BAILIAN_REISSUE_NO_WORKSPACE", message=( "Cannot resolve a workspace to reissue the key against " f"for integration {integration.slug or integration.id} " f"(reason={resolution.reason}, " f"preferred_region={region}, " f"candidates=[{resolution.describe_candidates()}]). " "Pick one on the 百炼 admin page (工作空间 → 设为默认) " "or set platform_setting " "cloud.bailian.default_workspace_id." ), status_code=503, ) # ``user_name_attempted`` is the third fallback on purpose: a grant # that died *after* CreateUser but before the success stamp only ever # recorded the attempted name — which is exactly the 2026-08-19 # no-default-workspace shape. Falling straight through to the # "bailian-reissue" literal would stamp ``iam_user_arn`` with a RAM # user that does not exist, and the revoke path would later tear down # that phantom while the real user lived on. ram_user_name = ( bailian_info.get("ram_user_name") or bailian_info.get("iam_user_name") or bailian_info.get("user_name_attempted") or "bailian-reissue" ) description = ( f"Helix · {assignment.request_number} · {ram_user_name} (reissue)" ) client = BailianClient(creds) api_key_resp = await client.create_api_key( workspace_id=workspace.workspace_id, region=workspace.region, description=description, ) dashscope_endpoint = dashscope_endpoint_for_provider(integration.provider) plaintext_bundle = ( f"API_KEY={api_key_resp['api_key_value']}\n" f"WORKSPACE_ID={workspace.workspace_id}\n" f"DASHSCOPE_ENDPOINT={dashscope_endpoint}\n" f"REGION={region}\n" f"API_KEY_ID={api_key_resp['api_key_id']}\n" f"RAM_USER_NAME={ram_user_name}\n" f"NOTE=Reissued by Helix on " f"{datetime.now(UTC).isoformat()}. Use as " f"`dashscope.api_key = \"\"` with the DashScope Python " f"SDK, or export DASHSCOPE_API_KEY=. See " f"/help#bailian-employee." ) assignment.key_value = _encrypt_key_value( plaintext_bundle, assignment_id=assignment.id, ) assignment.status = "active" if not assignment.iam_user_arn: assignment.iam_user_arn = f"acs:ram::*:user/{ram_user_name}" if not assignment.iam_integration_id: assignment.iam_integration_id = str(integration.id) # Preserve the legacy handle for audit; replace with new sk- handle. if bailian_info.get("api_key_id"): bailian_info["legacy_api_key_id"] = bailian_info["api_key_id"] bailian_info.update({ "iam_user_name": ram_user_name, "api_key_id": api_key_resp["api_key_id"], "api_key_created_by": api_key_resp.get("created_by"), "api_key_description": description, "workspace_id": workspace.workspace_id, "region": region, "dashscope_endpoint": dashscope_endpoint, "reissued_at": datetime.now(UTC).isoformat(), }) detail["bailian"] = bailian_info assignment.detail = detail # Soft-delete every prior ``bailian_api_keys`` row that belongs to # this assignment so the admin Bailian page doesn't show the new # sk- row alongside the now-defunct legacy row(s). Pre-2026-05-18 # autogrant misused the ``api_key_id`` column to store the RAM # AccessKey ID (LTAI…) — that handle was never a real ModelStudio # api key, and the nightly ``bailian.api_key_sync`` cron would # eventually mark it deleted on its own, but we don't make the # admin wait until that runs. Idempotent: rows already soft-deleted # stay where they are. superseded_rows = (await db.execute( select(BailianApiKey).where( BailianApiKey.assignment_id == assignment.id, BailianApiKey.deleted_at.is_(None), ) )).scalars().all() for stale in superseded_rows: stale.status = "deleted" stale.deleted_at = datetime.now(UTC) stale_raw = dict(stale.raw_data or {}) stale_raw["superseded_by_reissue_at"] = ( datetime.now(UTC).isoformat() ) stale_raw["superseded_by_api_key_id"] = api_key_resp["api_key_id"] stale.raw_data = stale_raw bailian_key_row = BailianApiKey( id=uuid.uuid4(), integration_id=str(integration.id), workspace_pk=workspace.id, api_key_id=api_key_resp["api_key_id"], display_name=description, owner_person_id=( str(assignment.assigned_to) if assignment.assigned_to else None ), ram_user_name=ram_user_name, assignment_id=assignment.id, status="active", raw_data={ "source": "auto-grant.reissue", "request_number": assignment.request_number, "created_by": api_key_resp.get("created_by"), "superseded": [str(s.id) for s in superseded_rows], }, ) db.add(bailian_key_row) await db.commit() await db.refresh(assignment) logger.info( "Bailian reissue succeeded: assignment=%s ram_user=%s " "new_api_key_id=%s workspace=%s", assignment.id, ram_user_name, api_key_resp["api_key_id"], workspace.workspace_id, ) return { "status": "reissued", "ram_user_name": ram_user_name, "api_key_id": api_key_resp["api_key_id"], "workspace_id": workspace.workspace_id, } __all__ = [ "maybe_auto_grant_bailian", "reissue_bailian_api_key", ]