"""AliCloud International (海外阿里云) driver. Inherits resource/billing capabilities from :class:`AliyunCloudDriver` but overrides: * ``list_scope_grants`` — uses Cloud SSO ``ListAccessAssignments`` (the source of truth for PingCAP's RD/SSO setup) instead of the parent's RAM-user-policy enumeration. RAM users on the master are service accounts / kept for inventory (``cmdb_iam_users``); the actual people-with-access view comes from SSO assignments mapped through the SSO directory. * ``get_monthly_billing`` aggregator (in ``AlicloudIntlProvider``) — uses ``IsGroupByProduct=false`` so the cross-RD-account total surfaces (master-only collapse otherwise hides 99% of spend). Why a class at all (vs. just config differences): * ``provider = "alicloud_intl"`` lets ``integrations.provider`` and downstream UI (tenant filter, analytics, palette) distinguish 国内 vs 海外 even though both use AK/SK. * ``provider_cls = AlicloudIntlProvider`` swaps in the subclass whose default region list targets ap-southeast / us-east / eu-central instead of the CN region set. """ from __future__ import annotations import logging from datetime import date, timedelta from typing import Any from app.domains.cmdb.providers.alicloud_intl import AlicloudIntlProvider from app.integrations.driver_base import ( COST_BUCKET_USAGE, CloudAccessContext, CloudDailyCost, CloudGrantResult, CloudRoleInfo, CloudScopeRef, DiscoveredGrant, ) from app.integrations.drivers.aliyun_cloud import AliyunCloudDriver from app.integrations.registry import register_driver logger = logging.getLogger(__name__) # Static fallback FX table for converting non-USD line items on the # AliCloud International BSS endpoint to USD. Intl tenants bill in USD # by default but a small fraction of legacy CNY-priced products can # still appear in the line-item feed (verified 2026-04-26 against # PingCAP's Resource Directory). USD is identity; CNY uses the # platform's CNY→USD rate (passed in via ``cny_per_usd``); other # currencies fall through the static table — keep tight, drop unknown # rows rather than silently treating "1 EUR" as "1 USD" (which is the # bug class :class:`AlicloudIntlProvider.get_monthly_billing` already # guards against for the monthly-aggregator path). _INTL_STATIC_FX_TO_USD: dict[str, float] = { "USD": 1.0, # CNY handled dynamically via cny_per_usd kwarg — listed here so # the unknown-currency branch in ``_convert_pretax_amount`` doesn't # mistake it for "drop the row when fx_rate is missing". "EUR": 1.08, # mirrors AlicloudIntlProvider._FX_TO_USD } # Cloud SSO API host. The directory itself is region-pinned (verified # 2026-04-26 against prod: ``Region=cn-shanghai`` even on an intl # tenant) — Aliyun's Cloud SSO is a single global service exposed via # the cn-shanghai endpoint regardless of where the master account is. _CLOUDSSO_HOST = "cloudsso.cn-shanghai.aliyuncs.com" _CLOUDSSO_VERSION = "2021-05-15" @register_driver class AlicloudIntlCloudDriver(AliyunCloudDriver): provider = "alicloud_intl" display_name = "AliCloud (海外)" description = ( "海外阿里云租户(ap-southeast / us-east / eu-central 等 Region)" "— 协议与国内阿里云一致,只是默认 Region 列表不同。凭证格式与" "国内阿里云相同。" ) provider_cls = AlicloudIntlProvider # International BSS lives at ``business.ap-southeast-1.aliyuncs.com``; # calling the CN ``business.aliyuncs.com`` from an intl AK returns # ``NotApplicable: caller site doesn't match API region`` (verified # 2026-04-25 against prod 海外 tenant). Override the parent's CN # default so daily billing actually returns rows. _BSS_HOST = "business.ap-southeast-1.aliyuncs.com" # ActionTrail is region-pinned (see parent docstring). The intl # tenant runs in ap-southeast-1 / us-east-1 / eu-central-1 (matches # the ``regions`` config_public default). Walk all three so resources # in any region get their create event attributed. Verified 2026-05-25 # against prod: ``getent hosts actiontrail.ap-southeast-1.aliyuncs.com`` # resolves; the CN ``actiontrail.cn-shanghai.aliyuncs.com`` host # ALSO resolves from intl callers but the intl AK on a CN endpoint # returns 0 events (calling-region mismatch), so we don't include CN # in the walk for the intl subclass. _ACTIONTRAIL_REGIONS: tuple[str, ...] = ( "ap-southeast-1", "us-east-1", "eu-central-1", ) # Intl tenant bills primarily in USD (海外 BSS endpoint default). # ``requires_cny_fx`` on the parent gates the entire ingestion on # a successful CNY→USD platform-FX resolution, but here CNY appears # only on the rare legacy product line — and ``_convert_pretax_amount`` # already drops those rows with a WARNING when the FX rate is # missing. Inheriting ``requires_cny_fx=True`` would kill an entire # intl ingestion on an unrelated FX-platform-setting outage. Disable. requires_cny_fx = False # SSO assignments are PER-target (RD account), so each scope sees # its own assignment set — unlike RAM-user-policy pairs which the # parent treats as tenant-wide. With grants_scope_wide=True the # service iterates every active scope and calls list_scope_grants # once per scope; that's the right shape for SSO. grants_scope_wide = True # Cloud SSO access configuration name treated as full ownership. # PingCAP uses literal ``admin`` (verified prod 2026-04-26). Custom # admin configs (DBaaS-Prod-Admin-Role, etc.) are ALSO admin-class # but live on different scopes and have their own owner candidate # selection — keep this conservative. owner_role_keys = frozenset({"admin", "AdministratorAccess"}) # Role governance (see app/integrations/cloud_role_governance.py). # CloudSSO AccessConfigurations are tenant-defined; "admin" is # owner-equivalent (above). DBaaS-Prod-Admin-Role is a bounded, # intentionally self-serviceable admin (classifier allowlists it from # the admin keyword) — still elevated, so it routes to admin approval. elevated_role_keys = frozenset({"DBaaS-Prod-Admin-Role"}) # ── Per-row currency conversion (USD-native override) ─────── def _convert_pretax_amount( self, item: dict[str, Any], native: float, fx_rate: float, ) -> tuple[float, str] | None: """Convert a BSS line-item ``PretaxAmount`` to USD honouring the per-row ``Currency`` field. Override of :meth:`AliyunCloudDriver._convert_pretax_amount`. 国内阿里云 BSS only ever returns CNY, so the parent hardcodes the conversion. The 海外 (international / SG) BSS endpoint at ``business.ap-southeast-1.aliyuncs.com`` returns USD by default but can also surface CNY (legacy products) and EUR (some intl marketplace SKUs). Reading ``Currency`` per row keeps the USD column accurate; the previous inherited behaviour silently divided every row by ``cny_per_usd``, undercounting USD bills by ~7× and breaking budget alerts on the intl tenant. Decision table: * ``Currency=USD`` (or empty / missing — the BSS default for intl) → ``amount_usd = native``, label ``USD``. * ``Currency=CNY`` → ``amount_usd = native / fx_rate``, label ``CNY``. Same math the parent applies tenant-wide. * ``Currency`` in :data:`_INTL_STATIC_FX_TO_USD` (e.g. EUR) → apply the static rate. Mirrors :class:`AlicloudIntlProvider`'s monthly aggregator so the line-item facts and the headline monthly card agree. * Unknown currency → drop the row with a WARNING. Silently treating an unknown unit as 1:1 USD is the bug class the monthly aggregator already explicitly guards against (:mod:`alicloud_intl` _FX_TO_USD comment, 2026-05-01). """ currency = str(item.get("Currency") or "USD").upper() if currency == "USD": return native, "USD" if currency == "CNY": if fx_rate <= 0: # No FX rate → we can't compute USD, but the CostFact base # contract says an unconvertible non-USD row must be # PERSISTED with ``amount_usd=0.0`` + the native amount + # native currency, NOT discarded — dropping it loses the # native amount and the audit trail entirely (the intl # tenant forces cny_per_usd=0, so this dropped EVERY CNY # line). The caller sets ``amount_native=native``, so # returning 0.0 here preserves the CNY amount for later # reconciliation / backfill once a rate is available. logger.warning( "AliCloud Intl integration %s: CNY line item with no " "cny_per_usd — persisting at amount_usd=0.0 (native CNY " "preserved) instead of dropping. item_keys=%s", self.integration_id, sorted(item.keys()), ) return 0.0, "CNY" return native / fx_rate, "CNY" rate = _INTL_STATIC_FX_TO_USD.get(currency) if rate is not None: return native * rate, currency logger.warning( "AliCloud Intl integration %s: dropping line item with " "unknown currency=%r (no static FX entry). PretaxAmount=%s " "ProductCode=%s", self.integration_id, currency, native, item.get("ProductCode"), ) return None # ── Per-day billing override (USD-native, not CNY) ────────── async def list_billing_daily( self, start_date: date, end_date: date, *, scope: CloudScopeRef | None = None, include_owner_breakdown: bool = False, bq_billing: dict[str, Any] | None = None, ) -> list[CloudDailyCost]: """USD-native override using BSS ``DescribeInstanceBill`` (Round-34). **Why this overrides AliyunCloudDriver.list_billing_daily AND swaps the BSS endpoint.** The parent calls ``QueryAccountBill`` which on an RD **Reseller** master (PingCAP intl prod has ``rd_account_type=ResellAccount``, the Aliyun default) returns only the calling account's own bill. In Reseller mode the master pays for every sub-account's resources, so a sub-account's own ``QueryAccountBill`` row shows ONLY cross-account residual fees (Data Transfer / K8s mgmt / Cloud Monitor / ApsaraMQ) — every ECS / EBS / OSS / SLB line item is INVISIBLE because master paid it. Prod evidence (2026-05-18, alicloud-intl-primary 18 sub-account scopes): post-Round-33 fix with ``BillOwnerId=`` still returned the same ~$5-50/day-per-sub residual; cost_facts' ``DescribeInstanceBill`` showed real spend of $20-$4500/day-per-sub. Round-33 added BillOwnerId on the correct hypothesis ("API needs to know the sub") but with the wrong API choice ("QueryAccountBill"). Round-34 fixes that by switching the API. ``DescribeInstanceBill`` is the **line-item** endpoint that ``list_cost_facts`` already uses. It returns the FULL RD- consolidated bill by default, with each row carrying both ``BillAccountID`` (payer = master) and ``OwnerID`` (resource owner = sub-account). We aggregate per-day per-(filtered scope) to produce ``CloudDailyCost`` rows. Result: billing_daily values now match cost_facts SUM(amount_usd) per (date, scope) **by construction** — same upstream feed, same per-row currency conversion via :meth:`_convert_pretax_amount`, just a different aggregation grain. ``currency="USD"`` stamped because the intl override's ``_convert_pretax_amount`` converts every row to USD (USD identity / EUR via static FX / CNY via ``cny_per_usd`` if available / unknown dropped with WARNING). Mixed-currency days collapse to USD (the dashboard expects a single per-day scalar). ``cny_per_usd`` is read from the integration config_public if present (operator override) or platform setting ``ai_spending.cny_to_usd_rate`` (refreshed weekly by FX cron); falls through to 0.0 (drops CNY rows with WARNING) if both fail. The signature is kept identical to the base contract so the lifecycle service doesn't need to learn intl-specific kwargs. """ del bq_billing # AWS-driver-specific kwarg; accept and ignore if scope is None: return [] # FX rate for the rare CNY line item, resolved ONCE per call # (the lifecycle service iterates per-scope; resolving FX here # means one platform-setting read per scope per cron, which is # the right grain — the FX value is identical for every day in # the requested window). Resolution order matches the cost- # facts surface: # # 1. ``self.config['cny_per_usd']`` (per-integration override — # legacy / ops-pinned path). # 2. Platform setting ``ai_spending.cny_to_usd_rate`` # (refreshed weekly by the FX cron — single source of # truth aligned with the cost-facts path). # # The cost-ingestion service resolves this for cost-facts and # passes ``cny_per_usd`` as a kwarg; the base # ``list_billing_daily`` contract doesn't carry that kwarg # today (would require a cross-driver refactor). To stay # aligned with the cost-facts numbers without forcing the # contract change, the override reaches into the platform # setting itself via a short-lived session — same FX value the # cost-facts path uses. Failure to resolve emits a WARNING # (symmetric with the cost-facts surface; pre-fix this was # INFO-level which hid FX outages from the failure-notifier) # and falls through to 0.0 which drops CNY rows in # ``_convert_pretax_amount``. from app.core.database import async_session_factory from app.core.fx_rate import resolve_cny_per_usd try: fx_rate = float(self.config.get("cny_per_usd") or 0.0) except (TypeError, ValueError): fx_rate = 0.0 if fx_rate <= 0: try: async with async_session_factory() as fx_session: fx_rate = await resolve_cny_per_usd( fx_session, integration_config=self.config, ) except Exception as exc: # noqa: BLE001 logger.warning( "AliCloud Intl integration %s: resolve_cny_per_usd " "failed (%s) — CNY line items will drop with WARNING. " "Pin a per-integration ``cny_per_usd`` in config_public " "or fix the platform FX setting to suppress this.", self.integration_id, exc, ) fx_rate = 0.0 # Empty-OwnerID (master-allocated) rows attribute to the master # scope only — mirrors list_cost_facts so cost_facts and # billing_daily stay reconciled and neither over-counts by # fanning master lines across every sub-account scope. master_id = await self._master_account_id() out: list[CloudDailyCost] = [] cur = start_date rows_missing_gross = 0 while cur <= end_date: by_service_usd: dict[str, float] = {} total_usd = 0.0 # 双口径累加器(见下方白名单分支):list_usd = 牌价合计, # list_seen 区分「当天真的没有白名单行」(→ None)与 # 「合计恰为 0」。 list_usd = 0.0 list_seen = False # 缺牌价行**两个**计数器,不要合成一个:``day_missing_gross`` # 每天重置,喂本天那条点名日期的 WARN(可操作); # ``rows_missing_gross`` 跨整个区间累加,喂循环之后那条汇总 # WARN。合成一个必然二选一出错——按天重置会让汇总只看见最后 # 一天(前面几天静默漏报),不重置又会让按天 WARN 每天复读前 # 面所有天的累计数。 day_missing_gross = 0 next_token: str | None = None # Page ceiling + same-token guard. A stable non-empty NextToken # (degenerate BSS response) would otherwise loop forever, hanging # the worker. 500 pages × 300 rows = 150k line items per day, # comfortably above any realistic daily bill. _MAX_BILL_PAGES = 500 pages_walked = 0 while True: pages_walked += 1 if pages_walked > _MAX_BILL_PAGES: raise RuntimeError( f"AliCloud Intl DescribeInstanceBill {cur} hit " f"{_MAX_BILL_PAGES}-page ceiling — refusing to loop; " "surface as transient failure rather than hang.", ) params: dict[str, Any] = { "BillingCycle": cur.strftime("%Y-%m"), "BillingDate": cur.isoformat(), "Granularity": "DAILY", # IsHideZeroCharge=false: keep paired charge-and-credit # rows so finance can reconcile RI cancellations / CSR # true-ups. Mirrors ``list_cost_facts``. The per-row # ``if native == 0: continue`` below still drops noise. "IsHideZeroCharge": "false", "MaxResults": 300, } if next_token: params["NextToken"] = next_token try: data = await self._acs3_call( host=self._BSS_HOST, version="2017-12-14", action="DescribeInstanceBill", params=params, ) except RuntimeError as e: raise RuntimeError( f"AliCloud Intl DescribeInstanceBill {cur} failed: {e}", ) from e raw_data = data.get("Data") or {} # ``DescribeInstanceBill`` returns ``Items`` as a plain # list (unlike ``QueryAccountBill`` which wraps it in # ``{"Item": [...]}``). Mirrors ``list_cost_facts``. items = raw_data.get("Items") or [] for item in items: # Client-side filter by OwnerID. DescribeInstanceBill # has no server-side per-sub filter that works on a # Reseller master; client filter is the only option. # Empty-OwnerID (master-allocated) rows attribute to # the master scope only — same posture as # ``list_cost_facts`` — so they're counted once, not # once per sub-account scope. if not self._billing_row_belongs_to_scope( str(item.get("OwnerID") or ""), scope.external_id, master_id, ): continue native = float(item.get("PretaxAmount") or 0) converted = self._convert_pretax_amount( item, native, fx_rate, ) if converted is None: continue usd_amount, _native_currency = converted svc = str( item.get("ProductName") or item.get("ProductCode") or "Other", ) # #2223: the invoice-zero skip used to run HERE, before # the list accumulation below, and its comment asserted # it "only drops single-row zeros". A single row can # perfectly well carry ``PretaxAmount == 0`` and # ``PretaxGrossAmount > 0`` — a usage row fully covered # by a voucher / RI / discount — so the skip was # silently dropping real LIST money, the one measure a # discount must never move (#2175 / #2180, same shape). # The invoice side is a no-op for a zero row anyway, so # it now simply adds 0 and falls through to the list # accumulation. if usd_amount: by_service_usd[svc] = ( by_service_usd.get(svc, 0.0) + usd_amount ) total_usd += usd_amount # 双口径（研发口径 = 牌价 PretaxGrossAmount）——与 # ``list_cost_facts`` 完全同源:同一白名单 # (``_classify_bss_bucket`` == usage)、同一牌价读取 # (``_bss_gross_amount``,缺失绝不用实付兜底)、同一 # 按行币种换算(``_convert_pretax_amount``,intl 的 # USD/CNY/EUR 决策表)。父类 CN 版在 PR3 就写了这两列, # 而生产用的是本 intl 覆盖 —— 漏改导致阿里云的 # billing_daily list 全程 NULL(2026-07-29 回填验收 # 发现),Owner 预算/波动提醒与账号账单页的日汇总兜底 # 因此拿不到研发口径。 if self._classify_bss_bucket(item, native) == COST_BUCKET_USAGE: gross_native = self._bss_gross_amount(item) if gross_native is None: # A zero-invoice row with no gross column is # the pure metering noise the old skip meant # to drop — not an undercount worth alarming. if native != 0: day_missing_gross += 1 rows_missing_gross += 1 else: gross_converted = self._convert_pretax_amount( item, gross_native, fx_rate, ) # FX 停摆闸(2026-07-29):``_convert_pretax_amount`` # 对「CNY 行 + 无可用汇率」刻意返回 (0.0, "CNY") # —— 实付列必须把行留下(amount_usd=0 + 原币金额 # 另存),那是它的契约。牌价列在 ``CloudDailyCost`` # 上没有原币伴生列,吞下这个 0.0 就等于凭空造出 # 「当日牌价 $0」:``daily_engineering_amount()`` # = COALESCE(list, invoice) 于是报 $0,而当天明明 # 还有 USD 行的真实支出。本子类 ``requires_cny_fx # = False`` 正是为了让 FX 停摆不杀掉整次摄取,所以 # 这个状态按设计可达。按「换不出来」处理。 if ( gross_converted is not None and gross_converted[1] == "CNY" and fx_rate <= 0 ): gross_converted = None if gross_converted is None: day_missing_gross += 1 rows_missing_gross += 1 else: list_usd += gross_converted[0] list_seen = True new_token = raw_data.get("NextToken") or None if not new_token: break if new_token == next_token: # NextToken didn't advance — a degenerate response that # would loop forever. Raise rather than spin the worker. raise RuntimeError( f"AliCloud Intl DescribeInstanceBill {cur}: NextToken " "did not advance between pages — refusing to loop; " "surface as transient failure.", ) next_token = new_token if day_missing_gross: # 云管理团队规则:缺牌价行单独告警核查,绝不用实付兜底。 # 用按天计数器,好让这条 WARN 里的数字与它点名的 ``cur`` # 对得上(区间累计数由循环之后那条汇总 WARN 报)。 logger.warning( "AliCloud Intl billing_daily %s scope=%s: %d usage " "row(s) missing PretaxGrossAmount — excluded from the " "engineering (list) measure; flag to cloud-mgmt", cur, scope.external_id, day_missing_gross, ) out.append(CloudDailyCost( date=cur, scope=scope, total_cost_usd=total_usd, # 双口径:USD(与 total 同一换算路径);阿里牌价即 # PretaxGrossAmount → public == list(与 CN 父类一致)。 # 整天无可计白名单行 → None(口径未接入),绝不伪造 0。 total_cost_usd_list=(list_usd if list_seen else None), total_cost_usd_public=(list_usd if list_seen else None), currency="USD", cost_by_service=by_service_usd, )) cur += timedelta(days=1) if rows_missing_gross: logger.warning( "AliCloud Intl billing_daily: %d rows missing/unconvertible " "PretaxGrossAmount (scope=%s %s..%s) — list-cost measure " "undercounts; per the whitelist rule these rows are NEVER " "backfilled from the post-discount amount. Investigate the " "BSS response shape.", rows_missing_gross, scope.external_id, start_date, end_date, ) return out # ── Cloud SSO grants discovery ────────────────────────────── async def _sso_directory_id(self) -> str | None: """Discover the org's Cloud SSO directory id. ``ListDirectories`` is global per-tenant — there's at most one directory per AliCloud account. Returns ``None`` ONLY when the tenant genuinely has no directory provisioned (legit non-SSO tenant). Raises ``RuntimeError`` on transport / API failure so callers can distinguish "no SSO here, fall back to RAM" from "we can't reach Aliyun, this is transient — surface a retry message and don't silently mark a revoke as already_absent". Round-2 code review caught the prior swallow-and-return-None path conflating these two states, masking a real revoke bug (an Aliyun blip during an offboarding revoke would land ``ok=True already_absent=True``, i.e. silent success even though the assignment is still live cloud-side). Result cached on the driver instance for both success and the "no directory" case to avoid re-querying on every method call. Transport failures are NOT cached — next call retries. """ if hasattr(self, "_cached_sso_dir"): return self._cached_sso_dir data = await self._acs3_call( host=_CLOUDSSO_HOST, version=_CLOUDSSO_VERSION, action="ListDirectories", method="POST", ) dirs = data.get("Directories") or [] if not dirs: self._cached_sso_dir = None return None self._cached_sso_dir = str(dirs[0].get("DirectoryId") or "") or None return self._cached_sso_dir async def _sso_list_assignments_for_target( self, directory_id: str, target_id: str, ) -> list[dict[str, Any]]: """Page ``ListAccessAssignments`` for one RD account. AliCloud Cloud SSO rejects ``MaxResults`` as ``InvalidParameter`` on this endpoint — must omit it and let the API default page size win. Pagination loop stops on ``IsTruncated=false`` OR a missing ``NextToken`` (defensive against an Aliyun final-page edge case where ``NextToken`` is sent even when there are no more rows; harmonised in Round-5 with the other 3 SSO helpers). """ out: list[dict[str, Any]] = [] next_token: str | None = None while True: params: dict[str, Any] = { "DirectoryId": directory_id, "TargetType": "RD-Account", "TargetId": target_id, } if next_token: params["NextToken"] = next_token try: data = await self._acs3_call( host=_CLOUDSSO_HOST, version=_CLOUDSSO_VERSION, action="ListAccessAssignments", method="GET", params=params, ) except RuntimeError as exc: # Raise so the caller records this scope as errored and # skips drift-sweep — a partial page followed by a break # would make the incomplete list look authoritative and # trigger silent revocation of legitimate access. raise RuntimeError( f"AliCloud Intl ListAccessAssignments target={target_id!r} " f"failed during pagination: {exc}" ) from exc out.extend(data.get("AccessAssignments") or []) # Stop on either IsTruncated=false OR missing NextToken # — same defensive pattern as # _sso_list_access_configurations / _sso_resolve_user_id # (Round-5 review caught the asymmetric paginators). Some # Aliyun endpoints return a NextToken on the final page # (when MaxResults exactly equals the last-page row # count), so a pure-NextToken loop could in theory re- # poll the same page indefinitely. if not data.get("IsTruncated"): break next_token = data.get("NextToken") or None if not next_token: break return out async def _sso_list_group_members( self, directory_id: str, group_id: str, ) -> list[dict[str, Any]]: """Expand a Group-type assignment into its individual users. Cloud SSO assignments can target Users OR Groups. For Groups (e.g. ``ADSyncAdmins``, ``dev-control-plane-services``) we list group members so the Helix grant rolls down to each person. ``ListGroupMembers`` returns ``GroupMembers`` array with embedded user records. """ out: list[dict[str, Any]] = [] next_token: str | None = None while True: params: dict[str, Any] = { "DirectoryId": directory_id, "GroupId": group_id, } if next_token: params["NextToken"] = next_token try: data = await self._acs3_call( host=_CLOUDSSO_HOST, version=_CLOUDSSO_VERSION, action="ListGroupMembers", method="GET", params=params, ) except RuntimeError as exc: # Same rationale as _sso_list_assignments_for_target: # a partial member list looks authoritative and silently # revokes access for group members on page 2+. raise RuntimeError( f"AliCloud Intl ListGroupMembers group={group_id!r} " f"failed during pagination: {exc}" ) from exc out.extend(data.get("GroupMembers") or []) # Same defensive pagination pattern as _sso_list_assignments # for_target / _sso_list_access_configurations — stop on # IsTruncated=false OR missing NextToken (Round-5 harmonisation). if not data.get("IsTruncated"): break next_token = data.get("NextToken") or None if not next_token: break return out async def list_scope_grants( self, scope: CloudScopeRef, ) -> list[DiscoveredGrant]: """Discover grants from Cloud SSO assignments on this RD account. Replaces the parent's RAM-user-policy enumeration. The parent was reading every RAM user × policy pair on the master and emitting them as grants — but SSO-managed tenants don't put humans in RAM (RAM users are service accounts / legacy). The truthful "who has access to this sub-account" view lives in Cloud SSO ``AccessAssignments``. Maps: * ``PrincipalType=User`` → one grant email = ``PrincipalName`` (SSO username = work email for PingCAP, validated) role_key = ``AccessConfigurationName`` (admin / poweruser / DBaaS-Prod-Admin-Role / …) * ``PrincipalType=Group`` → expand to members, emit one grant per (member, access config) tuple. Group name kept on the grant ``raw`` for debugging. Outcomes: * ``scope_type != "aliyun_account"`` — fall through to parent RAM (standalone-iam scopes). * Tenant has no CloudSSO directory (genuine non-SSO tenant) — fall through to parent RAM so legacy setups still surface their RAM-based grants. * ``ListDirectories`` transport-fail — propagate the ``RuntimeError`` so the orchestrator's per-scope try/except records this scope in ``errors`` and SKIPS drift-sweep for it. Round-4 logic review: falling back to parent RAM here on a transient blip would return near-zero grants on a CloudSSO tenant (RAM users are mostly service accounts), and the reconciler would mass-flip live CloudSSO grants to ``drift_removed`` for scopes whose active-grant count falls below the 50% drop-tolerance floor. * Otherwise — normal CloudSSO ``ListAccessAssignments`` discovery with Group expansion via ``ListGroupMembers``. """ if scope.scope_type != "aliyun_account": # SSO assignments only exist for RD accounts. Fall back # to parent (RAM) for standalone-iam scopes. return await super().list_scope_grants(scope) # _sso_directory_id raises on transport. We deliberately do # NOT catch it here: a transient ListDirectories failure on # a CloudSSO-managed tenant must propagate so the # ``sync_grants_for_integration`` orchestrator skips drift- # sweep for this scope. Falling back to parent RAM grants # discovery (which returns near-zero rows on a CloudSSO # tenant) WOULD pass the 50% drop-tolerance guard if the # scope had fewer than ~10 active grants, and the reconciler # would silently flip live CloudSSO grants to # ``status='revoked' revoke_reason='drift_removed'`` while # the assignments are still active cloud-side. Round-4 logic # review caught this; revert the earlier blanket fallback in # favour of explicit raise-on-transient. directory_id = await self._sso_directory_id() if not directory_id: logger.info( "AliCloud Intl: no Cloud SSO directory — falling back " "to RAM-based grants discovery for scope %s", scope.external_id, ) return await super().list_scope_grants(scope) assignments = await self._sso_list_assignments_for_target( directory_id, scope.external_id, ) if not assignments: return [] out: list[DiscoveredGrant] = [] # Cache group expansions within the call so a group attached # to multiple scopes only triggers one ListGroupMembers per # scope-iteration. group_cache: dict[str, list[dict[str, Any]]] = {} for a in assignments: principal_type = str(a.get("PrincipalType") or "") principal_id = str(a.get("PrincipalId") or "") principal_name = str(a.get("PrincipalName") or "") access_config_name = str(a.get("AccessConfigurationName") or "") access_config_id = str(a.get("AccessConfigurationId") or "") create_time = a.get("CreateTime") if not access_config_name: continue if principal_type == "User": email = principal_name.strip().lower() if "@" not in email: # Cloud SSO usernames in non-Google directories # could be opaque ids — skip without an email # rather than emit unresolvable grants. continue out.append(DiscoveredGrant( principal_email=email, principal_type="user", role_key=access_config_name, role_display_name=access_config_name, external_ref=( f"sso::{access_config_id}::{principal_id}::" f"{scope.external_id}" ), scope=scope, granted_at=None, # CreateTime parsed below if needed raw={ "source": "cloudsso", "AccessConfigurationId": access_config_id, "AccessConfigurationName": access_config_name, "PrincipalId": principal_id, "PrincipalName": principal_name, "TargetType": "RD-Account", "TargetId": scope.external_id, "CreateTime": create_time, }, )) elif principal_type == "Group": members = group_cache.get(principal_id) if members is None: members = await self._sso_list_group_members( directory_id, principal_id, ) group_cache[principal_id] = members for m in members: member_email = str(m.get("Email") or "").strip().lower() if "@" not in member_email: continue member_user_id = str(m.get("UserId") or "") out.append(DiscoveredGrant( principal_email=member_email, principal_type="user", role_key=access_config_name, role_display_name=access_config_name, external_ref=( f"sso::{access_config_id}::{member_user_id}::" f"{scope.external_id}::group::{principal_id}" ), scope=scope, granted_at=None, raw={ "source": "cloudsso", "AccessConfigurationId": access_config_id, "AccessConfigurationName": access_config_name, "GroupId": principal_id, "GroupName": principal_name, "PrincipalId": member_user_id, "PrincipalName": m.get("UserName") or member_email, "TargetType": "RD-Account", "TargetId": scope.external_id, }, )) else: logger.info( "AliCloud Intl: unknown PrincipalType=%s on assignment " "for scope %s — ignoring", principal_type, scope.external_id, ) return out # ── CloudSSO role catalog (override RAM-only parent) ───────── async def _sso_resolve_user_id( self, directory_id: str, email: str, ) -> str | None: """Resolve a CloudSSO User ID by username (= email). PingCAP's CloudSSO directory uses the work email as the SCIM/SAML-provisioned ``UserName`` (validated 2026-05-29 against prod — every assignment we discover via ``ListAccessAssignments`` carries ``PrincipalName=``). Returns None ONLY when the user is genuinely absent from the directory (e.g. SAML/SCIM provisioning lag for a brand-new hire). Raises ``RuntimeError`` on transport / API failure so ``grant_access`` can distinguish "user doesn't exist" from "we couldn't ask Aliyun" — the former tells the requester to wait on IT to provision, the latter is "retry later, this is an Aliyun blip", and they have very different remediation. Round-1 code review caught the prior swallow-and-return-None path that misrouted ops investigation to the wrong system. Aliyun ``ListUsers`` Filter syntax (confirmed against the official docs at https://www.alibabacloud.com/help/en/ cloudsso/developer-reference/api-cloudsso-2021-05-15-listusers) is Aliyun's relaxed grammar — **unquoted literal**: ``Filter=UserName eq `` — NOT strict SCIM v2 RFC 7644 which would mandate ``UserName eq ""``. Filter matching is case-insensitive. Defensively paginate via ``IsTruncated``/``NextToken`` even though ``UserName eq `` returns ≤1 row (UserName is the directory's natural key) — if Aliyun ever changes its default ``MaxResults=10`` or starts ignoring the Filter for a class of users, we still find the match instead of silently missing it past page 1. """ norm = email.strip().lower() if "@" not in norm: return None next_token: str | None = None while True: params: dict[str, Any] = { "DirectoryId": directory_id, "Filter": f"UserName eq {norm}", } if next_token: params["NextToken"] = next_token data = await self._acs3_call( host=_CLOUDSSO_HOST, version=_CLOUDSSO_VERSION, action="ListUsers", method="GET", params=params, ) for u in data.get("Users") or []: if str(u.get("UserName") or "").strip().lower() == norm: user_id = str(u.get("UserId") or "") return user_id or None if not data.get("IsTruncated"): break next_token = data.get("NextToken") or None if not next_token: break return None async def _sso_list_access_configurations( self, directory_id: str, ) -> list[dict[str, Any]]: """Page ``ListAccessConfigurations`` for the whole directory. Raises ``RuntimeError`` on transport / API failure (the caller decides whether to fall back, fail the grant, or retry — different semantics for ``list_scope_roles`` vs ``grant_access``). Caches the result on the driver instance for the rest of its lifetime (typically one sync round or one grant call — ``_load_cloud_driver`` constructs a fresh instance per request). Round-1 code review caught two earlier footguns and fixed both here: * Empty-list (``[]``) was cached as a successful result — if Aliyun ever returned a transient empty payload mid-sync, every subsequent scope in the same loop reused ``[]`` and silently wiped the role catalog. Now we cache ONLY non-empty successful results; an empty payload re-queries on the next call (and the per-call cost is one ROA round trip — cheap). * Failure (transport / 5xx) was swallowed and returned ``None``, conflated with "directory exists but is empty" by the resolver. Now we raise — the caller surface (``list_scope_roles`` / ``grant_access``) has the context to choose RAM fallback vs structured transient error. """ if getattr(self, "_cached_access_configs", None): return self._cached_access_configs out: list[dict[str, Any]] = [] next_token: str | None = None while True: params: dict[str, Any] = {"DirectoryId": directory_id} if next_token: params["NextToken"] = next_token data = await self._acs3_call( host=_CLOUDSSO_HOST, version=_CLOUDSSO_VERSION, action="ListAccessConfigurations", method="GET", params=params, ) out.extend(data.get("AccessConfigurations") or []) # Mirror the defensive pagination pattern from # _sso_resolve_user_id: stop on either IsTruncated=false # OR a missing NextToken. Round-2 code review caught the # asymmetry — some Aliyun endpoints return a NextToken # on the final page (when MaxResults exactly equals the # last-page row count), so a pure NextToken loop could # in theory re-poll the same final page indefinitely. if not data.get("IsTruncated"): break next_token = data.get("NextToken") or None if not next_token: break # Cache ONLY non-empty results so a transient empty payload # doesn't poison the dropdown for the driver instance's # remaining lifetime. if out: self._cached_access_configs = out return out async def _sso_resolve_access_config_id( self, directory_id: str, name: str, ) -> str | None: """Resolve an AccessConfiguration ID by name (the ``role_key`` the request flow stores in ``cmdb_cloud_access_grants``). Raises ``RuntimeError`` on API failure (propagated from :meth:`_sso_list_access_configurations`). Returns ``None`` only when the lookup succeeded but the requested name isn't in the directory — caller surfaces "role not found, refresh catalog" vs a generic transient error accordingly. """ name_norm = name.strip() if not name_norm: return None configs = await self._sso_list_access_configurations(directory_id) # configs is now guaranteed to be a list (may be empty, # meaning the directory legitimately has no permission sets # — treat as "name not found"). Transport errors raised. for cfg in configs: if str(cfg.get("AccessConfigurationName") or "").strip() == name_norm: cfg_id = str(cfg.get("AccessConfigurationId") or "") return cfg_id or None return None async def list_scope_roles( self, scope: CloudScopeRef, ) -> list[CloudRoleInfo]: """Return CloudSSO AccessConfigurations as the role catalog. In a CloudSSO-managed intl tenant, the "role" an employee requests is a CloudSSO permission set (``admin``, ``poweruser``, ``DBaaS-Prod-Admin-Role``, …) bound to one Resource-Directory account via ``CreateAccessAssignment``. The parent ``AliyunCloudDriver.list_scope_roles`` enumerates the master account's RAM policies, which (a) misses these admin-defined permission sets, and (b) surfaces the ``AliyunReservedSSO-*`` Custom RAM policy mirrors that the ``AttachPolicyToUser`` API hard-refuses to attach (fixed in PR #565). Both failure modes wedge the request flow. Falls through to the parent RAM enumeration only when: * ``scope_type`` is not ``aliyun_account`` (CloudSSO scopes are RD-Account targets only), OR * Tenant genuinely has no CloudSSO directory provisioned (legacy single-account intl tenant in pure RAM mode). Round-4 logic review tightened the prior behaviour: a transport-level ``ListDirectories`` / ``ListAccessConfigurations`` failure now PROPAGATES the ``RuntimeError`` instead of falling back to RAM. The orchestrator's per-scope try/except records the scope in ``errors`` and skips drift-sweep for it; falling back to RAM would emit ``Type:Name``-shape keys that pass the catalog's drop-tolerance floor and trigger mark-and-sweep on the un-touched CloudSSO rows. ``role_key`` matches the convention used by :meth:`list_scope_grants` above (``AccessConfigurationName``) so a granted role round-trips: catalog row → request dropdown selection → ``grant_access`` → reconciliation by the next ``cmdb.grants_sync`` cycle. """ if scope.scope_type != "aliyun_account": return await super().list_scope_roles(scope) # Same Round-4 rationale as ``list_scope_grants`` above: do # NOT catch transport-fail on _sso_directory_id / # ListAccessConfigurations and fall back to RAM. The # ``sync_roles_for_integration`` orchestrator has its own # drop-tolerance / every-scope-failed guards # (lifecycle_service.py:6253-6270); raising lets those guards # do their job. Falling back to RAM would emit RAM-shape # ``Type:Name`` keys that pass the drop-tolerance floor and # then mark-and-sweep would soft-delete the (un-touched) # CloudSSO rows — producing a mixed-shape catalog and a # window where employees see RAM keys in the dropdown for a # CloudSSO-only tenant. directory_id = await self._sso_directory_id() if not directory_id: return await super().list_scope_roles(scope) configs = await self._sso_list_access_configurations(directory_id) out: list[CloudRoleInfo] = [] for cfg in configs: name = str(cfg.get("AccessConfigurationName") or "").strip() cfg_id = str(cfg.get("AccessConfigurationId") or "").strip() if not name or not cfg_id: continue out.append(CloudRoleInfo( role_key=name, display_name=name, description=cfg.get("Description") or None, # ``is_custom=True`` — CloudSSO permission sets are # admin-defined (in the Aliyun CloudSSO console), # i.e. customer-managed in the same sense RAM # ``Custom`` policies are. Round-1 code review # caught the earlier ``is_custom=False`` as breaking # ``reset_role_catalog_to_default``: that endpoint # computes ``target_promoted = bool(is_custom) OR # role_key in defaults`` and the # ``cloud.aliyun.default_policies`` setting only # lists RAM-style ``System:`` keys — so an admin # clicking ``重置为默认`` on an intl tenant would # have demoted every CloudSSO role to # ``is_promoted=False`` with no path back except # manual per-row promote. Marking ``is_custom=True`` # makes them survive the reset, matching the # parent's "Custom is promoted by default" rule. is_custom=True, source="cloudsso", scope=scope, raw={ "AccessConfigurationId": cfg_id, "AccessConfigurationName": name, }, is_promoted=True, )) return out # ── Submit-time shape pre-flight (mirror of grant_access) ──── async def validate_request_role( self, scope: CloudScopeRef, role_key: str, ) -> str | None: """Reject a role_key whose shape :meth:`grant_access` would refuse, BEFORE the request enters the approval queue. Same decision tree as ``grant_access`` below, minus the actual ``CreateAccessAssignment`` — so the requester is told immediately instead of the approver hitting the failure at grant time. Pure shape + a single (cached) ``ListDirectories``; fails OPEN on any transient error per the base-class contract — the grant path stays authoritative. * ``scope_type != aliyun_account`` → permit (legacy RAM tenants). * ``ListDirectories`` transient-fail → permit (grant path re-checks once the API recovers). * CloudSSO directory present + RAM-shape ``Type:Name`` key → reject (stale catalog row — request a CloudSSO permission set). * No directory + bare CloudSSO-shape key → reject (non-SSO tenant needs a RAM ``Type:Name`` key). """ if scope.scope_type != "aliyun_account": return None try: directory_id = await self._sso_directory_id() except Exception: # noqa: BLE001 # Honor the base-class "NEVER raise / fail-open" contract. # ``_sso_directory_id`` raises ``RuntimeError`` on an Aliyun # ≥400, but ``_acs3_call`` lets a raw transport error # (httpx ConnectError / ReadTimeout / DNS) propagate — catch # both. The grant path stays the authoritative gate and # re-checks once the API recovers. return None if directory_id and ":" in role_key: return ( f"Cannot grant RAM-shape role {role_key!r} on a " "CloudSSO-managed AliCloud tenant. The role catalog " "appears to be stale — refresh and pick a CloudSSO " "permission set name (e.g. 'admin' / 'poweruser' / " "'DBaaS-Prod-Admin-Role')." ) if not directory_id and ":" not in role_key: return ( f"Cannot grant role {role_key!r} on a non-CloudSSO " "AliCloud tenant. The role catalog appears to be stale " "— refresh and pick a RAM-shape role_key like " "'System:ReadOnlyAccess'." ) return None # ── CloudSSO grant (override RAM-only parent) ──────────────── async def grant_access( self, ctx: CloudAccessContext, ) -> CloudGrantResult: """Grant a CloudSSO permission set to a user on an RD account. Mirrors :meth:`revoke_access` below: works through CloudSSO ``CreateAccessAssignment`` rather than RAM ``AttachPolicyToUser``. The parent path tried to attach an ``AliyunReservedSSO-*`` Custom RAM policy and Aliyun returned 400 ``NotSupport.AliyunReservedPolicy`` — the bug PR #565 closes. Outcomes: * ``scope_type != "aliyun_account"`` — fall through to parent RAM (legacy iam-mode single-account tenants). * ``ListDirectories`` transport-fail — return ``ok=False`` with a clean transient-error message. NEVER fall through to RAM on transport-fail: we cannot tell which surface the tenant uses without a successful directory lookup, and a stray RAM-attach attempt on a CloudSSO tenant could create a shadow RAM user that bypasses CloudSSO governance. * No CloudSSO directory + bare CloudSSO-shape ``role_key`` (no ``:``) — return ``ok=False`` shape-mismatch error (the catalog is stale; surface a guide rather than fall through to parent's ``InvalidRoleKey`` rejection). * No CloudSSO directory + RAM-shape ``role_key`` (``Type:Name``) — fall through to parent RAM (legit non-SSO tenant). * Directory present + RAM-shape ``role_key`` — return ``ok=False`` shape-mismatch error (Round-4 guard against stale ``cmdb_cloud_roles`` rows from before this PR). * Directory present + bare ``role_key`` — normal CloudSSO grant path (``ListUsers`` → ``ListAccessConfigurations`` → ``CreateAccessAssignment``). Returns ``external_ref`` in the same shape :meth:`list_scope_grants` emits so the next reconciliation matches the in-flight grant and ``revoke_access`` finds the right assignment to remove: ``sso::{access_config_id}::{user_id}::{target_id}``. Transport errors at any resolve step (``ListUsers`` / ``ListAccessConfigurations``) surface as structured ``CloudGrantResult(ok=False, error_message=...)`` with a retry-friendly message rather than propagating the raw API exception or misrouting to a "user not provisioned" / "config renamed" path. """ if ctx.scope.scope_type != "aliyun_account": return await super().grant_access(ctx) try: directory_id = await self._sso_directory_id() except RuntimeError as exc: # ListDirectories transient failure. The CloudSSO # directory exists (we've discovered grants from it # previously) but Aliyun is currently unreachable. # Return a clean transient error so the request can # be retried by the requester / approver. NEVER fall # through to parent RAM here: even RAM-shape role_keys # ought to be re-tried via CloudSSO once the API # recovers, since we cannot tell which surface this # tenant truly uses without a successful # ListDirectories response. Round-2 code review tightened # this to distinguish the two semantically-different # None-returns from the prior helper. return CloudGrantResult( ok=False, error_message=( "AliCloud CloudSSO ListDirectories temporarily " f"unavailable ({exc}). Please retry the grant " "in a few minutes." ), ) # On a CloudSSO-managed tenant, a RAM-shape role_key # ("System:ReadOnlyAccess" / "Custom:Foo") is a stale # catalog row from before this PR or a manual override — # the new ``list_scope_roles`` only ever emits bare # CloudSSO names. If we tried to RAM-attach it on a # CloudSSO tenant we'd either silently succeed (creating # a stray RAM user that bypasses CloudSSO governance) or # fail at _sso_resolve_access_config_id with the # misleading "role renamed in console" error. Surface a # clear shape-mismatch error instead. Round-4 logic # review caught this asymmetry vs the no-directory branch # below. if directory_id and ":" in ctx.role_key: return CloudGrantResult( ok=False, error_message=( f"Cannot grant RAM-shape role {ctx.role_key!r} on " "a CloudSSO-managed AliCloud tenant. The role " "catalog appears to be stale — refresh and " "re-apply with a CloudSSO permission set name " "(e.g. 'admin' / 'poweruser' / " "'DBaaS-Prod-Admin-Role')." ), ) if not directory_id: # Genuinely no SSO directory provisioned on this tenant # — _sso_directory_id raised on transport, so reaching # here means the API succeeded and returned an empty # Directories list. Fall through to the parent RAM # path — this is a legitimate non-SSO tenant and a RAM # grant is the intended behaviour. # # Defensive: if role_key has no ':' (a CloudSSO-shape # name like ``poweruser``), the parent's role_key # validation would reject with "must be Type:Name" — # would be confusing for a non-SSO tenant whose # catalog never produced bare-name keys. Surface a # clear error instead. if ":" not in ctx.role_key: return CloudGrantResult( ok=False, error_message=( f"Cannot grant role {ctx.role_key!r} on a " "non-CloudSSO AliCloud tenant. The role " "catalog appears to be stale — refresh and " "re-apply with a RAM-shape role_key like " "'System:ReadOnlyAccess'." ), ) return await super().grant_access(ctx) try: access_config_id = await self._sso_resolve_access_config_id( directory_id, ctx.role_key, ) except RuntimeError as exc: # ListAccessConfigurations transient failure. Don't fall # through to RAM (the role_key is a CloudSSO name, the # parent would reject ``: not in role_key``). Surface a # clean transient-error so the requester / approver # knows to retry rather than chase a fictional config # rename. Round-1 code review: this was previously # collapsed with "config not found". return CloudGrantResult( ok=False, error_message=( "AliCloud CloudSSO ListAccessConfigurations " f"temporarily unavailable ({exc}). Please retry " "the grant in a few minutes." ), ) if not access_config_id: return CloudGrantResult( ok=False, error_message=( f"CloudSSO AccessConfiguration {ctx.role_key!r} not " "found in directory. The role may have been renamed " "or deleted in the Aliyun CloudSSO console — " "refresh the role catalog and re-apply." ), ) try: user_id = await self._sso_resolve_user_id( directory_id, ctx.principal_email, ) except RuntimeError as exc: # ListUsers transient failure. As above, surface a clean # transient-error rather than blaming SAML/SCIM (which # would misroute the IT ops investigation). Round-1 # code review catch. return CloudGrantResult( ok=False, error_message=( "AliCloud CloudSSO ListUsers temporarily " f"unavailable ({exc}). Please retry the grant " "in a few minutes." ), ) if not user_id: return CloudGrantResult( ok=False, error_message=( f"User {ctx.principal_email} not found in CloudSSO " "directory. Ensure SAML/SCIM provisioning has synced " "this user from the IdP before granting CloudSSO " "access." ), ) target_id = ctx.scope.external_id external_ref = f"sso::{access_config_id}::{user_id}::{target_id}" # Detail keys MUST stay in # ``lifecycle_service._ALLOWED_EXTERNAL_DETAIL_KEYS`` (the # allowlist that prevents secrets / leaky internals from # landing in ``cmdb_cloud_access_grants.detail``). Use the # established snake_case names; PR also extends that allowlist # with CloudSSO-specific ``access_configuration_id`` / # ``access_configuration_name`` / ``target_id`` / ``target_type``. # Round-1 code review caught the prior PascalCase emit being # silently filtered to ``external_detail={}``. detail = { "access_configuration_id": access_config_id, "access_configuration_name": ctx.role_key, "principal_id": user_id, "principal_email": ctx.principal_email, "principal_name": ctx.principal_email, "target_id": target_id, "target_type": "RD-Account", } try: await self._acs3_call( host=_CLOUDSSO_HOST, version=_CLOUDSSO_VERSION, action="CreateAccessAssignment", method="POST", params={ "DirectoryId": directory_id, "AccessConfigurationId": access_config_id, "TargetType": "RD-Account", "TargetId": target_id, "PrincipalType": "User", "PrincipalId": user_id, }, ) except RuntimeError as exc: msg = str(exc) # Aliyun returns these on re-grant; treat as idempotent. # ``EntityAlreadyExists.AccessAssignment`` is the literal # code from prod observation 2026-05-29 — keep the broader # substring matches too in case Aliyun varies the wording. if ( "EntityAlreadyExists" in msg or "AccessAssignmentAlreadyExists" in msg or "AlreadyExists.AccessAssignment" in msg ): return CloudGrantResult( ok=True, already_exists=True, external_ref=external_ref, detail=detail, ) return CloudGrantResult(ok=False, error_message=msg) return CloudGrantResult( ok=True, external_ref=external_ref, detail=detail, ) # ── CloudSSO revoke (override RAM-only parent) ─────────────── async def revoke_access( self, ctx: CloudAccessContext, ) -> CloudGrantResult: """Revoke a CloudSSO assignment or a Group-expansion member. ``list_scope_grants`` above emits two ``external_ref`` shapes for SSO-managed tenants: * ``sso::{access_config_id}::{user_id}::{target_id}`` — direct User assignment. Removed via ``DeleteAccessAssignment``. * ``sso::{access_config_id}::{user_id}::{target_id}::group:: {group_id}`` — synthetic per-member row emitted when the assignment is on a Group. Removing the assignment would revoke access for the entire group; instead, remove just this one user from the SSO group via ``RemoveUserFromGroup``. Other group members keep their assignment. Non-``sso::`` refs (legacy RAM-policy attachments) fall through to the parent. The parent's RAM-based revoke crashed loudly before this override (``InvalidParameter.PolicyName.InvalidChars`` — verified 2026-05-13 against prod offboarding for huohao@pingcap.com), because it parsed ``sso::::...`` as ``user_name='sso'`` + ``PolicyName=':'``. """ ref = ctx.existing_external_ref or "" if not ref.startswith("sso::"): return await super().revoke_access(ctx) parts = ref.split("::") # Minimum 4 parts for User direct: ["sso", access_config_id, # user_id, target_id]. Group-expansion adds ["group", group_id]. if len(parts) < 4 or not parts[1] or not parts[2] or not parts[3]: return CloudGrantResult( ok=False, error_message=f"Malformed CloudSSO external_ref: {ref!r}", ) access_config_id = parts[1] principal_id = parts[2] target_id = parts[3] is_group_expansion = ( len(parts) >= 6 and parts[4] == "group" and bool(parts[5]) ) group_id = parts[5] if is_group_expansion else None try: directory_id = await self._sso_directory_id() except RuntimeError as exc: # ListDirectories transient failure. CRITICAL: do NOT # mark already_absent here — that would silently report # a successful revoke when in fact the assignment is # still live cloud-side. Surface a transient-error so # the offboarding flow retries the revoke later. # Round-2 code review caught this latent revoke bug — # the prior swallow-and-return-None path conflated # "no directory exists" (idempotent success) with "API # blip" (must retry) and chose the wrong default for # offboarding correctness. return CloudGrantResult( ok=False, error_message=( "AliCloud CloudSSO ListDirectories temporarily " f"unavailable ({exc}); cannot confirm whether the " "assignment is gone. Please retry the revoke." ), ) if not directory_id: # No SSO directory at all on this tenant — the assignment # cannot exist, treat as already absent rather than block # offboarding on a vanished IdP. return CloudGrantResult(ok=True, already_absent=True) if is_group_expansion and group_id: # Per-member group removal: only this user loses access. try: await self._acs3_call( host=_CLOUDSSO_HOST, version=_CLOUDSSO_VERSION, action="RemoveUserFromGroup", method="POST", params={ "DirectoryId": directory_id, "GroupId": group_id, "UserId": principal_id, }, ) except RuntimeError as exc: msg = str(exc) if ( "EntityNotExist" in msg or "NotExist" in msg or ".NotFound" in msg or "UserNotInGroup" in msg ): return CloudGrantResult(ok=True, already_absent=True) return CloudGrantResult(ok=False, error_message=msg) return CloudGrantResult(ok=True, external_ref=ref) # Direct User assignment removal. try: await self._acs3_call( host=_CLOUDSSO_HOST, version=_CLOUDSSO_VERSION, action="DeleteAccessAssignment", method="POST", params={ "DirectoryId": directory_id, "AccessConfigurationId": access_config_id, "TargetType": "RD-Account", "TargetId": target_id, "PrincipalType": "User", "PrincipalId": principal_id, # Clean up the per-account user shadow if this was # the user's last assignment on the target. Without # this, Aliyun keeps a stale provisioned RAM identity # behind and a subsequent DeleteUser fails with # ``EntityExists.User.AssignedAccessConfiguration``. "DeprovisionStrategy": "DeprovisionForLastAccessAssignmentOnAccount", }, ) except RuntimeError as exc: msg = str(exc) if ( "EntityNotExist" in msg or "NotExist" in msg or ".NotFound" in msg or "AccessAssignmentNotExist" in msg ): return CloudGrantResult(ok=True, already_absent=True) return CloudGrantResult(ok=False, error_message=msg) return CloudGrantResult(ok=True, external_ref=ref)