"""Bailian auto-grant: on the approved-status callback the helper 1. mints a RAM user via the AliCloud International cloud integration (BailianClient OpenAPI calls patched in test), 2. attaches AliyunBailianFullAccess (governance/audit anchor only — the RAM user is no longer given an AccessKey), 3. resolves the default workspace, 4. calls ModelStudio ``CreateApiKey`` to mint a real DashScope sk- API key against that workspace, 5. encrypts and writes the credential bundle (``API_KEY=sk-…``) onto the assignment, 6. upserts the RAM user into ``cmdb_iam_users``, 7. mirrors the new sk- handle into ``bailian_api_keys`` so the admin Bailian page shows it immediately, 8. dispatches an ``apikey.ready`` notification to the assignee. We don't hit Aliyun — every BailianClient OpenAPI method is patched. History (2026-05-18): the pre-fix version of this flow handed the employee a RAM AccessKey/Secret pair and instructed them to use it with the DashScope SDK in "RAM-auth mode". The SDK does not support that mode — it only reads ``dashscope.api_key``. The tests below encode the new contract: API_KEY in the bundle, no ACCESS_KEY_ID, and a real ModelStudio api_key_id stamped onto ``detail.bailian``. """ from __future__ import annotations import uuid from decimal import Decimal from unittest.mock import AsyncMock import pytest from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.domains.apikey.bailian_autogrant import ( _is_bailian_provider, maybe_auto_grant_bailian, ) from app.domains.apikey.models import ApiKeyAssignment, ApiKeyProvider from app.domains.cmdb.models import CmdbIamUser from app.identity.person_models import IdentityPerson from app.integrations.models import Integration async def _make_intl_integ(db_session: AsyncSession) -> Integration: """Build a minimal AliCloud International integration whose slug matches the Bailian autogrant heuristic ``"bailian" in slug``. ``resolve_credentials`` is patched in each test, so the ``secrets_encrypted`` / ``secrets_nonce`` placeholders never have to decrypt to anything real — but they must be non-empty so the helper doesn't short-circuit on the "no secrets" branch. """ integ = Integration( id=uuid.uuid4(), kind="cloud", provider="alicloud_intl", name="AliCloud (Bailian)", slug="aliyun-bailian", enabled=True, is_primary=False, capabilities=["servers", "iam", "bailian"], config_public={"default_region": "ap-southeast-1"}, secrets_encrypted="ciphertext-placeholder", secrets_nonce="nonce-placeholder", ) db_session.add(integ) await db_session.commit() return integ async def _make_person(db_session: AsyncSession) -> IdentityPerson: person = IdentityPerson( id=uuid.uuid4(), source_system="successfactors", hris_person_id=f"sf-{uuid.uuid4().hex[:8]}", display_name="Shuai Feng", work_email="shuai.feng@pingcap.com", employment_status="active", ) db_session.add(person) await db_session.commit() return person async def _make_bailian_assignment( db_session: AsyncSession, person_id: uuid.UUID, ) -> ApiKeyAssignment: provider = ApiKeyProvider( id=uuid.uuid4(), name="Alibaba Cloud Model Studio", description="Bailian / DashScope access", is_active=True, sort_order=0, ) db_session.add(provider) await db_session.commit() assignment = ApiKeyAssignment( id=uuid.uuid4(), provider_id=provider.id, assigned_to=person_id, status="approved", request_number=f"AK-20260506-{uuid.uuid4().hex[:4].upper()}", project_name="Bailian eval", reason="Bailian model benchmarking", monthly_budget_usd=Decimal("80.00"), ) db_session.add(assignment) await db_session.commit() return assignment def _patch_bailian_client(monkeypatch) -> None: """Replace each Aliyun OpenAPI call with an in-memory canned answer. Note: post-2026-05-18, ``create_access_key`` is NOT called in the happy path — autogrant relies on ``create_api_key`` (ModelStudio CreateApiKey) for the user-facing credential. We still patch ``create_access_key`` defensively in case a future test regresses the contract. """ from app.domains.bailian.client import BailianClient async def fake_ensure(self, *, user_name, display_name, email, comments): return { "user": { "UserId": "u-fake-1", "UserName": user_name, "DisplayName": display_name, "Email": email or "", "CreateDate": "2026-05-06T03:00:00Z", }, "created": True, } async def fake_attach(self, *, user_name, policy_name, policy_type): return {"already_attached": False} async def fake_mint(self, *, user_name): raise AssertionError( "create_access_key must NOT be called on the new Bailian " "autogrant path (sk- API key is the user-facing credential)." ) async def fake_create_api_key( self, *, workspace_id, description=None, region=None, ): return { "api_key_id": "ak-fake-9999", "api_key_value": "sk-fake1234567890abcdef", "workspace_id": workspace_id, "created_by": "helix-master", "disabled": False, "created_at_ms": 1747500000000, "description": description, "raw": {}, } monkeypatch.setattr(BailianClient, "ensure_ram_user", fake_ensure) monkeypatch.setattr(BailianClient, "attach_policy", fake_attach) monkeypatch.setattr(BailianClient, "create_access_key", fake_mint) monkeypatch.setattr(BailianClient, "create_api_key", fake_create_api_key) monkeypatch.setattr( "app.domains.apikey.bailian_autogrant.resolve_credentials", lambda integration: { "access_key_id": "x", "access_key_secret": "y", "bailian_region": "ap-southeast-1", "_provider": "alicloud_intl", }, ) async def _make_default_workspace( db_session: AsyncSession, integration_id: uuid.UUID, ): """Seed a default BailianWorkspace for the integration so the autogrant's workspace-resolution step succeeds. CreateApiKey is workspace-scoped on intl ModelStudio — the new autogrant fails loudly when no workspace is resolvable, so every happy-path test must seed one. """ from app.domains.bailian.models import BailianWorkspace ws = BailianWorkspace( id=uuid.uuid4(), integration_id=str(integration_id), workspace_id="ws-fake-default", name="Default Workspace", is_default=True, ) db_session.add(ws) await db_session.commit() return ws @pytest.mark.asyncio async def test_autogrant_succeeds_for_bailian( db_session: AsyncSession, monkeypatch, ): integ = await _make_intl_integ(db_session) await _make_default_workspace(db_session, integ.id) person = await _make_person(db_session) assignment = await _make_bailian_assignment(db_session, person.id) _patch_bailian_client(monkeypatch) notify_spy = AsyncMock() from app.notifications.service import NotificationService monkeypatch.setattr(NotificationService, "dispatch", notify_spy) result = await maybe_auto_grant_bailian(db_session, assignment) assert result["status"] == "granted" assert result["api_key_id"] == "ak-fake-9999" assert result["workspace_id"] == "ws-fake-default" assert assignment.status == "active" assert assignment.iam_integration_id == str(integ.id) assert assignment.iam_user_arn is not None assert assignment.iam_user_arn.endswith(result["ram_user_name"]) # ── Encrypted at rest, decrypts to the structured KEY=VALUE bundle. from app.domains.apikey.service import _decrypt_key_value assert assignment.key_value is not None # Round-20 audit (A2F6): Bailian auto-grant binds AES-GCM AAD # to the assignment's id; the decrypt MUST pass the same # assignment_id. decrypted = _decrypt_key_value( assignment.key_value, assignment_id=assignment.id, ) # ── New (2026-05-18) contract: the bundle's primary auth is the # sk- DashScope API key, NOT the legacy AK/SK pair. assert "API_KEY=sk-fake1234567890abcdef" in decrypted assert "API_KEY_ID=ak-fake-9999" in decrypted assert "WORKSPACE_ID=ws-fake-default" in decrypted assert "DASHSCOPE_ENDPOINT=" in decrypted # The old AK/SK keys MUST NOT appear — the DashScope SDK rejects # them and the help-doc snippet now uses dashscope.api_key. assert "ACCESS_KEY_ID=" not in decrypted assert "ACCESS_KEY_SECRET=" not in decrypted # NOTE wording: directs the user to dashscope.api_key, not the # bogus "RAM-auth mode" hint the pre-fix bundle used. assert "dashscope.api_key" in decrypted assert "/help#bailian-employee" in decrypted # ── detail.bailian carries the lifecycle handles needed for revoke. detail = assignment.detail or {} bailian_info = detail.get("bailian") or {} assert bailian_info.get("granted") is True assert bailian_info.get("api_key_id") == "ak-fake-9999" # iam_user_name MUST be set (mirror of ram_user_name) so the revoke # flow can dispatch the RAM-user teardown. Pre-fix, only # ram_user_name was written and the revoke path silently no-op'd. assert bailian_info.get("iam_user_name") == result["ram_user_name"] assert bailian_info.get("workspace_id") == "ws-fake-default" # ── IAM upsert: the RAM user must be visible in cmdb_iam_users # immediately, without waiting for the next IAM sync. iam_row = (await db_session.execute( select(CmdbIamUser).where( CmdbIamUser.integration_id == str(integ.id), CmdbIamUser.user_name == result["ram_user_name"], ) )).scalar_one() assert iam_row.arn == assignment.iam_user_arn assert iam_row.user_id_external == "u-fake-1" assert iam_row.tags["__source"] == "helix_autogrant.bailian" assert iam_row.owner_person_id == str(person.id) assert assignment.iam_user_id == iam_row.id # ── bailian_api_keys mirror: the new sk- handle (NOT the legacy # RAM AccessKeyId) must land in the table so the admin Bailian # page renders the key immediately. from app.domains.bailian.models import BailianApiKey bak_row = (await db_session.execute( select(BailianApiKey).where( BailianApiKey.assignment_id == assignment.id, ) )).scalar_one() assert bak_row.api_key_id == "ak-fake-9999" assert bak_row.ram_user_name == result["ram_user_name"] # ── Notification dispatch happened exactly once with the right # template_key + recipient. assert notify_spy.await_count == 1 env = notify_spy.await_args.args[0] assert env.recipient.id == person.id assert env.template_key == "apikey.ready" assert env.related_resource_type == "api_key_assignment" assert env.related_resource_id == assignment.id @pytest.mark.asyncio async def test_autogrant_fails_loud_without_default_workspace( db_session: AsyncSession, monkeypatch, ): """The sk- flow requires a workspace to mint against. With no default workspace seeded, autogrant must refuse cleanly and stamp the error onto detail.bailian so the admin can see it on the row. Pre-fix this path silently issued a bundle with empty WORKSPACE_ID.""" await _make_intl_integ(db_session) person = await _make_person(db_session) assignment = await _make_bailian_assignment(db_session, person.id) _patch_bailian_client(monkeypatch) result = await maybe_auto_grant_bailian(db_session, assignment) assert result["status"] == "no-default-workspace" assert assignment.key_value is None detail = assignment.detail or {} bailian_info = detail.get("bailian") or {} assert bailian_info.get("granted") is False assert "default workspace" in (bailian_info.get("error") or "").lower() @pytest.mark.asyncio async def test_autogrant_skips_when_provider_not_bailian( db_session: AsyncSession, ): """Safe no-op for unrelated providers.""" person = await _make_person(db_session) provider = ApiKeyProvider( id=uuid.uuid4(), name="Some other LLM", is_active=True, sort_order=0, ) db_session.add(provider) await db_session.commit() assignment = ApiKeyAssignment( id=uuid.uuid4(), provider_id=provider.id, assigned_to=person.id, status="approved", request_number=f"AK-20260506-{uuid.uuid4().hex[:4].upper()}", project_name="x", reason="x", monthly_budget_usd=Decimal("10.00"), ) db_session.add(assignment) await db_session.commit() result = await maybe_auto_grant_bailian(db_session, assignment) assert result == {"status": "not-bailian"} # The unrelated provider must NOT have been transitioned to active # — the helper's first check is supposed to short-circuit before # anything else happens. assert assignment.status == "approved" @pytest.mark.asyncio async def test_autogrant_idempotent_when_already_granted( db_session: AsyncSession, monkeypatch, ): """A second call after a successful grant must not mint a second sk- API key — the ``detail.bailian.granted`` sentinel short-circuits the second invocation. CreateApiKey is NOT idempotent on the Aliyun side (each call burns a workspace API-key slot), so the in-process guard is the only thing standing between us and duplicate keys.""" integ = await _make_intl_integ(db_session) await _make_default_workspace(db_session, integ.id) person = await _make_person(db_session) assignment = await _make_bailian_assignment(db_session, person.id) _patch_bailian_client(monkeypatch) notify_spy = AsyncMock() from app.notifications.service import NotificationService monkeypatch.setattr(NotificationService, "dispatch", notify_spy) first = await maybe_auto_grant_bailian(db_session, assignment) assert first["status"] == "granted" second = await maybe_auto_grant_bailian(db_session, assignment) assert second == {"status": "already-granted"} # Notification fires exactly once across the two calls. assert notify_spy.await_count == 1 @pytest.mark.asyncio async def test_reissue_replaces_legacy_aksk_with_sk_key( db_session: AsyncSession, monkeypatch, ): """Migration helper: existing assignments granted under the pre-2026-05-18 AK/SK bundle should reissue cleanly to a sk- bundle. Seeds an assignment whose detail.bailian was the legacy shape (no ``api_key_id``, ``ram_user_name`` only, key_value carrying ``ACCESS_KEY_ID=...``), then calls ``reissue_bailian_api_key`` and asserts the new bundle, the new api_key_id, and that the legacy handle was preserved for audit. """ from app.domains.apikey.bailian_autogrant import ( reissue_bailian_api_key, ) from app.domains.apikey.service import ( _decrypt_key_value, _encrypt_key_value, ) integ = await _make_intl_integ(db_session) await _make_default_workspace(db_session, integ.id) person = await _make_person(db_session) assignment = await _make_bailian_assignment(db_session, person.id) # Legacy bundle shape — what's actually on liwen.xu's row. legacy_plaintext = ( "ACCESS_KEY_ID=LTAIlegacy\n" "ACCESS_KEY_SECRET=secretlegacy\n" "REGION=ap-southeast-1\n" "WORKSPACE_ID=ws-fake-default\n" "DASHSCOPE_ENDPOINT=https://dashscope-intl.aliyuncs.com/api/v1\n" "RAM_USER_NAME=bailian-legacy-1\n" ) assignment.key_value = _encrypt_key_value( legacy_plaintext, assignment_id=assignment.id, ) assignment.status = "active" assignment.iam_integration_id = str(integ.id) assignment.detail = { "bailian": { "granted": True, "integration_id": str(integ.id), "ram_user_name": "bailian-legacy-1", "api_key_id": "LTAIlegacy", # legacy: RAM AK ID, not sk- "workspace_id": "ws-fake-default", "region": "ap-southeast-1", }, } await db_session.commit() _patch_bailian_client(monkeypatch) result = await reissue_bailian_api_key(db_session, assignment) assert result["status"] == "reissued" assert result["api_key_id"] == "ak-fake-9999" # ── Bundle replaced with sk- shape; no AK/SK leakage. decrypted = _decrypt_key_value( assignment.key_value, assignment_id=assignment.id, ) assert "API_KEY=sk-fake1234567890abcdef" in decrypted assert "ACCESS_KEY_ID=" not in decrypted assert "ACCESS_KEY_SECRET=" not in decrypted # ── Legacy api_key_id preserved for audit; current handle is the # new ModelStudio one. bailian_info = (assignment.detail or {}).get("bailian") or {} assert bailian_info.get("api_key_id") == "ak-fake-9999" assert bailian_info.get("legacy_api_key_id") == "LTAIlegacy" assert bailian_info.get("reissued_at") @pytest.mark.asyncio async def test_reissue_soft_deletes_prior_bailian_api_keys_row( db_session: AsyncSession, monkeypatch, ): """Pre-2026-05-18 autogrant inserted a `bailian_api_keys` row whose `api_key_id` was the RAM AccessKey ID (LTAI…) — not a real ModelStudio sk- handle. On reissue we must soft-delete that legacy row so the admin Bailian page does not show both the phantom LTAI row AND the freshly-minted sk- row simultaneously (reported by the user 2026-05-18 on liwen.xu's record). """ from sqlalchemy import select as _select from app.domains.apikey.bailian_autogrant import ( reissue_bailian_api_key, ) from app.domains.apikey.service import _encrypt_key_value from app.domains.bailian.models import BailianApiKey integ = await _make_intl_integ(db_session) workspace = await _make_default_workspace(db_session, integ.id) person = await _make_person(db_session) assignment = await _make_bailian_assignment(db_session, person.id) # Legacy bundle + legacy bailian_api_keys row (api_key_id = RAM AK). assignment.key_value = _encrypt_key_value( "ACCESS_KEY_ID=LTAIlegacy\nACCESS_KEY_SECRET=legacy\n", assignment_id=assignment.id, ) assignment.status = "active" assignment.iam_integration_id = str(integ.id) assignment.detail = { "bailian": { "granted": True, "integration_id": str(integ.id), "ram_user_name": "bailian-legacy-1", "api_key_id": "LTAIlegacy", "workspace_id": workspace.workspace_id, }, } db_session.add(BailianApiKey( id=uuid.uuid4(), integration_id=str(integ.id), workspace_pk=workspace.id, api_key_id="LTAIlegacy", # the bug: defunct RAM AK ID display_name="Helix · legacy", owner_person_id=str(person.id), ram_user_name="bailian-legacy-1", assignment_id=assignment.id, status="active", )) await db_session.commit() _patch_bailian_client(monkeypatch) result = await reissue_bailian_api_key(db_session, assignment) assert result["status"] == "reissued" rows = (await db_session.execute( _select(BailianApiKey) .where(BailianApiKey.assignment_id == assignment.id) .order_by(BailianApiKey.created_at), )).scalars().all() by_handle = {r.api_key_id: r for r in rows} assert "LTAIlegacy" in by_handle, "legacy row must still exist (for audit)" assert "ak-fake-9999" in by_handle, "new sk- row must have been inserted" # Legacy row soft-deleted, new row active. assert by_handle["LTAIlegacy"].status == "deleted" assert by_handle["LTAIlegacy"].deleted_at is not None assert by_handle["ak-fake-9999"].status == "active" assert by_handle["ak-fake-9999"].deleted_at is None # The legacy row's raw_data records what superseded it. assert (by_handle["LTAIlegacy"].raw_data or {}).get( "superseded_by_api_key_id", ) == "ak-fake-9999" # ─── Pure-function tests ──────────────────────────────────────────── def test_is_bailian_provider_normalises_whitespace() -> None: """Same matcher class as the Azure OpenAI prod incident (AK-20260509-EF8A): admin-typed double space / NBSP / full-width / tab in the provider name must not bypass the autogrant.""" class _P: def __init__(self, name: str) -> None: self.name = name assert _is_bailian_provider(_P("Aliyun Bailian (DashScope)")) assert _is_bailian_provider(_P("Alibaba Cloud Model Studio")) assert _is_bailian_provider(_P("Alibaba Cloud Model Studio")) # 2 sp assert _is_bailian_provider(_P("Alibaba Cloud Model Studio")) # NBSP assert _is_bailian_provider(_P("Alibaba Cloud　Model Studio")) # full assert _is_bailian_provider(_P("Alibaba Cloud\tModel Studio")) # tab assert _is_bailian_provider(_P(" bailian ")) # padding assert _is_bailian_provider(_P("阿里云 百炼")) assert _is_bailian_provider(_P("DashScope")) assert not _is_bailian_provider(_P("AWS Bedrock")) assert not _is_bailian_provider(_P("Azure OpenAI")) assert not _is_bailian_provider(None)