"""AliCloud International (海外阿里云) — multi-sub-account ECS / SLB / domain / DNS / OSS / CDN / RAM enumeration. The intl tenant uses the same RPC/ACS3 protocol as 国内阿里云, so the single-account behavior is inherited from :class:`AliyunProvider`. On top of that, this subclass knows that the master account is an **umbrella** holding ~17 member accounts under a Resource Directory. Each list method here: 1. Discovers every member account via Resource Directory ``ListAccounts``. 2. STS-AssumeRole's into each member account using ``ResourceDirectoryAccountAccessRole`` (the default role Aliyun creates when an account joins a Resource Directory; trust policy allows the master account to assume). 3. Calls the parent ``AliyunProvider.list_*`` with temp credentials swapped in. 4. Tags every returned object with the member account id (in the ``raw_data`` blob and, when the column exists, in ``cloud_scope_id``) so the UI can render per-sub-account totals. If AssumeRole fails for a sub-account (no trust policy, role removed, network blip), the failure is logged and that sub-account's resources are skipped — the rest of the sync still succeeds. This is critical because partial failure must NOT collapse the whole tenant's data. Master-account-level resources (the tiny set that lives directly on the umbrella account, not under any child) are also fetched so we don't lose data that worked before this refactor. """ from __future__ import annotations import logging from contextlib import asynccontextmanager from datetime import UTC, date, datetime, timedelta from typing import Any from app.domains.cmdb.providers import aliyun as _aliyun from app.domains.cmdb.providers.base import ( CloudCdnDomain, CloudDnsRecord, CloudDomain, CloudLoadBalancer, CloudServer, ProviderCapability, ServerMetricPoint, ServerMetricTarget, register_provider, ) from app.domains.cmdb.providers.exceptions import ( CapabilityNotSupported, ProviderAuthError, ) logger = logging.getLogger(__name__) _INTL_REGIONS = [ "ap-southeast-1", "ap-southeast-2", "ap-southeast-3", "ap-southeast-5", "ap-southeast-6", "ap-southeast-7", "ap-northeast-1", "ap-northeast-2", "ap-south-1", "us-east-1", "us-west-1", "eu-central-1", "eu-west-1", "me-east-1", ] # Aliyun creates this role on every member account that joins a # Resource Directory. Its trust policy whitelists the master account, # so the master AK can ``sts:AssumeRole`` it without any per-account # setup. Override via the ``member_assume_role_name`` integration # config field if your org uses a custom name. _DEFAULT_MEMBER_ROLE = "ResourceDirectoryAccountAccessRole" # 5-minute safety margin: STS tokens are valid for 1 hour by default; # we re-AssumeRole well before expiry so a long-running sync doesn't # trip on an in-flight token rotation. _TOKEN_TTL_SECONDS = 3600 _TOKEN_REFRESH_MARGIN = 300 # Currencies this driver is willing to surface a total for. The VALUES # are no longer used to convert (#2178 — see the monthly path): a native # amount is returned with its native label and the read side does the # one and only conversion, at the live rate. The keys remain the # "is this currency known?" gate, because an unknown currency must # surface ``total_cost_usd=None`` rather than be treated as 1:1 with USD. _FX_TO_USD = { "USD": 1.0, "CNY": 0.14, "EUR": 1.08, } @register_provider class AlicloudIntlProvider(_aliyun.AliyunProvider): """Aliyun driver for the international tenant. Distinct ``provider_type`` from ``aliyun`` so the UI can render 国内 vs 海外 as separate tenants. All capability behavior (sync semantics, scope lifecycle, billing, DNS writes) follows the same RPC contract; the override surface here is: * region default → ``_INTL_REGIONS`` * domain endpoint host + version pinned to the intl variant * per-sub-account resource fanout via STS AssumeRole """ _DOMAIN_API_HOST = "https://domain-intl.aliyuncs.com/" _DOMAIN_API_VERSION = "2017-12-18" # AliCloud Intl tenants must hit the SG BSS host. Using the CN host # (``business.aliyuncs.com``) returns ``NotApplicable: caller site # doesn't match API region``. Same fix as PR #49 applied to daily # billing in the unified driver. _BSS_HOST = "business.ap-southeast-1.aliyuncs.com" @classmethod def provider_type(cls) -> str: return "alicloud_intl" @classmethod def supported_capabilities(cls) -> frozenset[ProviderCapability]: # Add IAM (RAM users) on top of the inherited capabilities. The # base ``AliyunProvider`` only declared resource caps because # 国内阿里云 has been billed-only / single-account; intl has the # full Resource Directory + per-account RAM tree. return frozenset({ ProviderCapability.SERVERS, ProviderCapability.DOMAINS, ProviderCapability.DNS_RECORDS, ProviderCapability.LOAD_BALANCERS, ProviderCapability.CDN, ProviderCapability.OBJECT_STORAGE, ProviderCapability.IAM, ProviderCapability.METRICS, }) async def get_monthly_billing(self, month: date | None = None) -> dict: """Aggregate monthly bill across the master + every RD member. OVERRIDES the base ``AliyunProvider.get_monthly_billing`` which uses ``IsGroupByProduct=true``. That parameter — counter- intuitively — collapses the response to ONLY the calling account's own resources, dropping the cross-account aggregate BSS otherwise computes. For PingCAP's 17-RD-member intl tenant this caused the page to show ¥4,386 (master's own usage) instead of $440k+ (consolidated). See 2026-04-26 bug-report screenshot. With ``IsGroupByProduct=false`` BSS returns one row per (OwnerID, BillingCycle) — i.e. one row per RD member account with that account's TOTAL pretax cost. We sum to get the consolidated bill and surface a per-account breakdown in ``cost_by_service`` so the billing page can show "which sub-account is racking up the bill". AliCloud International tenants bill in USD, NOT CNY (unlike ``aliyun-primary`` which is CNY). We honour whatever ``Currency`` the API returns so the multi-cloud aggregator doesn't silently sum across currencies. """ from datetime import timedelta as _td target = ( month or (date.today().replace(day=1) - _td(days=1)) ).replace(day=1) cycle = target.strftime("%Y-%m") endpoint = f"https://{self._BSS_HOST}/" cost_by_account: dict[str, float] = {} # Track per-currency totals; if more than one is observed the # rows can't be safely summed into a single "total" without an # FX conversion, and silent-summing across currencies would # produce nonsense headline numbers. per_currency_totals: dict[str, float] = {} next_token: str | None = None last_page: dict = {} # Page ceiling + non-advancing-token guard (#2222) — inherited # ``_MAX_BILL_PAGES`` from the CN provider; this walk had neither. pages_walked = 0 while True: pages_walked += 1 if pages_walked > self._MAX_BILL_PAGES: raise RuntimeError( f"AliCloud Intl QueryAccountBill {cycle} hit the " f"{self._MAX_BILL_PAGES}-page ceiling — refusing to " "loop; surfacing as a transient failure rather than " "pinning a scheduled-jobs slot indefinitely.", ) from app.domains.cmdb.providers.aliyun import _common_params params = _common_params("QueryAccountBill", self._ak) params["Version"] = self._BSS_VERSION params["BillingCycle"] = cycle params["Granularity"] = "MONTHLY" # KEY FIX: false (not true) so the response is the # cross-account consolidated bill, not just the master's # own resources. params["IsGroupByProduct"] = "false" params["MaxResults"] = "300" if next_token: params["NextToken"] = next_token data = await self._call(endpoint, params) last_page = data payload = data.get("Data") or {} items = (payload.get("Items") or {}).get("Item") or [] for item in items: amt = float(item.get("PretaxAmount") or 0) # AccountName falls back to OwnerID. The "service" # axis here is per-sub-account totals — most useful # for an org bill (which sub is the spend driver). account_label = ( str(item.get("AccountName") or "") or str(item.get("OwnerID") or "") or "unknown" ) cost_by_account[account_label] = ( cost_by_account.get(account_label, 0.0) + amt ) row_currency = str(item.get("Currency") or "USD") per_currency_totals[row_currency] = ( per_currency_totals.get(row_currency, 0.0) + amt ) prev_token = next_token next_token = payload.get("NextToken") or None if not next_token: break if next_token == prev_token: raise RuntimeError( f"AliCloud Intl QueryAccountBill {cycle} returned a " "non-advancing NextToken — refusing to loop.", ) # Pick canonical currency: largest-spend currency wins as the # native label. Convert all rows to USD using the fixed FX # table for the headline ``total_cost_usd``. Per-currency # breakdown is preserved in ``raw_response`` so audit can # drill in if a tenant ever bills in mixed currencies. # Surface ``total_cost_usd=None`` when the FX rate is unknown # for ANY contributing currency — silently treating an unknown # currency as 1:1 with USD (the previous ``_FX_TO_USD.get(c, 1.0)`` # fallback) inflates JPY/HKD/SGD totals by 7-100x and triggers # bogus budget alerts. Native total + per-currency breakdown # remain on the response so dashboards can render the truth. unknown: list[str] = [ c for c in per_currency_totals if c not in _FX_TO_USD ] if not per_currency_totals: currency = "USD" total_native = 0.0 total_usd: float | None = 0.0 elif unknown: logger.warning( "AliCloud Intl billing has unknown currency/currencies " "%s; surfacing total_cost_usd=None. per_currency=%s", unknown, per_currency_totals, ) currency = max( per_currency_totals.items(), key=lambda kv: kv[1], )[0] total_native = per_currency_totals[currency] total_usd = None elif len(per_currency_totals) == 1: currency, total_native = next(iter(per_currency_totals.items())) # Native value with a native label — NOT ``native × stale # hard-coded rate`` (#2178). The CN parent had exactly this # shape and it was removed in the Round-666 fix, whose # comment spells out the damage: the handler's FX-fallback # persists ``total_cost_usd`` while keeping the native # label, so an already-multiplied value gets divided by the # live rate a SECOND time on read (~51× understatement with # the 0.14 constant against a ~7.2 rate). Surfacing the # native amount lets the single live-FX conversion on the # read side be the only one. The EUR branch was worse still: # it produced a USD number labelled ``EUR``, which the # reader (no EUR rate) drops entirely. total_usd = total_native else: currency = max( per_currency_totals.items(), key=lambda kv: kv[1], )[0] total_native = per_currency_totals[currency] # Mixed currencies: the response carries ONE (value, label) # pair, so there is no honest single native number. Report # the dominant currency's native total and withhold the USD # headline rather than pre-converting with the stale table — # the per-currency breakdown stays in ``raw_response`` for # anyone who needs the full picture. logger.warning( "AliCloud Intl billing spans multiple currencies %s; " "surfacing the dominant currency's native total and " "total_cost_usd=None. per_currency=%s", sorted(per_currency_totals), per_currency_totals, ) total_usd = None raw = dict(last_page) if isinstance(last_page, dict) else {} raw["per_currency_totals"] = per_currency_totals if unknown: raw["fx_unknown_currencies"] = unknown return { "month": target, "total_cost_usd": total_usd, "total_cost_native": total_native, "cost_by_service": cost_by_account, "currency": currency, "raw_response": raw, } def __init__(self, credentials): # type: ignore[no-untyped-def] super().__init__(credentials) # Cache: account_id -> {ak, sk, token, expires_at}. Re-used across # the multiple region-iterations a single ``list_*`` triggers. self._sts_cache: dict[str, dict[str, Any]] = {} # Cached sub-account list (populated lazily by ``_list_member_accounts``). self._member_accounts_cache: list[dict[str, Any]] | None = None def _regions(self) -> list[str]: configured = (self._credentials.fields.get("regions") or "").strip() if configured: return [r.strip() for r in configured.split(",") if r.strip()] return list(_INTL_REGIONS) def _member_role_name(self) -> str: """Custom override or the Aliyun default RD role name.""" v = (self._credentials.fields.get("member_assume_role_name") or "").strip() return v or _DEFAULT_MEMBER_ROLE def _member_external_id(self) -> str | None: """Optional ExternalId for cross-account AssumeRole. Confused-deputy protection per AWS/Aliyun best practice. When the customer's role trust policy is hardened to require ``ExternalId``, returning ``None`` here makes every AssumeRole silently fail with no actionable error. Caller sets ``member_external_id`` in the integration config to opt in. Trust policy that doesn't require ExternalId still works because Aliyun ignores the param when the trust policy lacks the ``oss:ExternalId`` condition. """ v = (self._credentials.fields.get("member_external_id") or "").strip() return v or None # ── Resource Directory listing ───────────────────────────── async def _list_member_accounts(self) -> list[dict[str, Any]]: """Return every active member account in the Resource Directory. Cached for the life of the provider so a single sync that calls ``list_servers`` then ``list_domains`` only hits the API once. Master-account fallback (one synthetic entry pointing at the master AK, no AssumeRole needed) when the tenant is not part of a Resource Directory — keeps the single-account path working for tenants that haven't enabled RD yet. """ if self._member_accounts_cache is not None: return self._member_accounts_cache accounts: list[dict[str, Any]] = [] page = 1 while True: params = _aliyun._common_params("ListAccounts", self._ak) params.update({ "Version": "2020-03-31", "PageSize": "100", "PageNumber": str(page), }) try: data = await self._call( "https://resourcemanager.aliyuncs.com/", params, ) except Exception as exc: # noqa: BLE001 # Pre-launch hardening 2026-05-16: match on the SDK's # structured ``code`` field FIRST so the redactor's # str(exc) reshaping can't mask a real permission error # as "single-account tenant" and silently drop every # sub-account from the fan-out. # Round-1 review follow-up (Review-A P2-3): pre-fix the # substring fallback was guarded by ``not code`` — # future-proof problem: if ``_call`` later attaches a # ``.code`` attribute with an unrelated value, the # substring fallback would be skipped and a real # RD-unsupported error would be re-raised, taking down # the fan-out. Now: code-match OR substring-match # (independent), so adding structured ``.code`` later # is purely additive — never breaks the existing # substring path. code = ( str(getattr(exc, "code", "") or "") or str(getattr(exc, "error_code", "") or "") ) msg = str(exc) rd_unsupported_codes = { "ResourceDirectoryNotInUse", "Forbidden.RAM", "InvalidAction", } # Tenant isn't part of a Resource Directory — single # account mode. Synthesise one entry from the master AK. is_rd_unsupported = ( code in rd_unsupported_codes or "ResourceDirectoryNotInUse" in msg or "Forbidden.RAM" in msg or "InvalidAction" in msg ) if is_rd_unsupported: logger.info( "AlicloudIntl: ListAccounts unavailable (code=%s) — " "treating as single-account tenant", code or msg[:140], ) self._member_accounts_cache = [] return [] # Real error — propagate so the sync job marks failed # rather than silently producing 0 resources. raise items = ( (data.get("Accounts") or {}).get("Account") or data.get("Accounts") or [] ) if not items: break for acct in items: state = str(acct.get("Status") or acct.get("JoinMethod") or "") # Only sync into accounts in a usable state. ``CreateSuccess`` # is the steady-state per Aliyun docs. if state and "success" not in state.lower() and state.lower() not in ( "active", "joinedsuccess", "createsuccess", ): continue accounts.append({ "account_id": str(acct.get("AccountId") or ""), "display_name": ( acct.get("DisplayName") or acct.get("AccountName") or "" ), "type": acct.get("Type"), "folder_id": acct.get("FolderId"), }) # Terminate on the authoritative signal — a short/last page — # not on TotalCount. When the response omits or zeroes # TotalCount but still returns a full page, ``page*100 >= 0`` # is always true and pages 2+ (and all their sub-accounts) # would be silently dropped from the RD fan-out. if len(items) < 100: break page += 1 self._member_accounts_cache = accounts return accounts # ── STS AssumeRole ───────────────────────────────────────── async def _assume_role( self, account_id: str, ) -> tuple[str, str, str] | None: """Return ``(ak, sk, token)`` temp credentials for a sub-account, or ``None`` if AssumeRole failed. Cached for ~55 minutes so a per-region fanout doesn't burn the STS rate limit. Refresh before the 5-minute expiry margin to ride out clock skew between Helix and Aliyun. """ cached = self._sts_cache.get(account_id) if cached and datetime.now(UTC) < cached["expires_at"]: return cached["ak"], cached["sk"], cached["token"] role_arn = f"acs:ram::{account_id}:role/{self._member_role_name()}" session_name = f"helix-cmdb-{account_id}"[:32] params = _aliyun._common_params("AssumeRole", self._ak) params.update({ "Version": "2015-04-01", "RoleArn": role_arn, "RoleSessionName": session_name, "DurationSeconds": str(_TOKEN_TTL_SECONDS), }) # Optional ExternalId for confused-deputy protection. When the # customer's trust policy uses ``oss:ExternalId`` to bind the # role to a specific external token, omitting the param makes # every AssumeRole silently fail. Aliyun ignores the param when # the trust policy lacks the condition, so unconditionally # passing when configured is safe. external_id = self._member_external_id() if external_id: params["ExternalId"] = external_id try: data = await self._call("https://sts.aliyuncs.com/", params) except Exception as exc: # noqa: BLE001 # Redact: the raw exception string can carry the signed STS # request URL with AccessKeyId + Signature in query params. # Route through the shared redactor before logging so the # AK fragment doesn't land in operator logs. from app.domains.cmdb.providers.exceptions import ( redact_provider_error, ) logger.info( "AlicloudIntl: AssumeRole %s failed: %s", role_arn, redact_provider_error(exc, max_len=200), ) return None creds = (data.get("Credentials") or {}) ak = str(creds.get("AccessKeyId") or "") sk = str(creds.get("AccessKeySecret") or "") token = str(creds.get("SecurityToken") or "") if not ak or not sk or not token: logger.info( "AlicloudIntl: AssumeRole %s returned incomplete creds", role_arn, ) return None self._sts_cache[account_id] = { "ak": ak, "sk": sk, "token": token, "expires_at": datetime.now(UTC) + timedelta(seconds=_TOKEN_TTL_SECONDS - _TOKEN_REFRESH_MARGIN), } return ak, sk, token @asynccontextmanager async def _impersonate(self, account_id: str): """Swap ``self._ak / self._sk / self._security_token`` for the sub-account's temp credentials for the duration of the block, then restore the master credentials. Yields the sub-account_id (so callers can tag results) when the AssumeRole succeeds, and ``None`` when it doesn't — callers should ``continue`` past ``None`` rather than calling further Aliyun APIs with broken auth. """ creds = await self._assume_role(account_id) if creds is None: yield None return ak, sk, token = creds orig_ak, orig_sk, orig_token = self._ak, self._sk, self._security_token self._ak = ak self._sk = sk self._security_token = token try: yield account_id finally: self._ak = orig_ak self._sk = orig_sk self._security_token = orig_token # ── Per-sub-account fanout helpers ───────────────────────── async def _fanout(self, fetcher_name: str) -> list[Any]: """Run a parent-class ``list_*`` once per sub-account (plus the master, in case the master itself owns resources directly). ``fetcher_name`` is the unbound method name on :class:`AliyunProvider` — e.g. ``"list_servers"``, ``"list_domains"``. Each item returned by the parent has ``raw["__sub_account_id"]`` injected so downstream sync code can later upsert ``cloud_scope_id`` into the CMDB tables. """ members = await self._list_member_accounts() if not members: # Single-account tenant: just call the master once. results = await getattr(_aliyun.AliyunProvider, fetcher_name)(self) for r in results: _stamp_account_id(r, str(self._ak)) return list(results) out: list[Any] = [] # 1. Master account — covers things that live on the umbrella # (e.g. domain registrations are typically registered on the # master, not member accounts). try: master_results = await getattr(_aliyun.AliyunProvider, fetcher_name)(self) for r in master_results: _stamp_account_id(r, "__master__") out.extend(master_results) except CapabilityNotSupported: # Service is not subscribed at the master account level # (e.g. CDN never opened on the org). Propagate so the sync # pipeline marks the capability ``skipped`` instead of # ``failed`` — historical rows survive untouched and the # scheduled-job history doesn't go red over a paid-feature # gap. Crucially this is a SEPARATE branch from # ``ProviderAuthError`` below: a real bad-AK fault must # still propagate as an auth failure, not get silently # converted into a skip. raise except ProviderAuthError: # Master-account auth failures must NOT be swallowed — # if the root AK is broken every member-account # AssumeRole will also fail, the fan-out returns ``[]``, # and mark-and-sweep nukes every previously-known # resource. Re-raise so sync marks the capability # ``failed`` and operator gets the credential error # in the alert pipeline. raise except Exception as exc: # noqa: BLE001 logger.warning( "AlicloudIntl %s on master failed (non-auth): %s", fetcher_name, str(exc)[:200], ) # 2. Each member account. # Pre-launch hardening 2026-05-16: every "skip this sub-account" # path records a partial_failure so the sync job's # capability_results surface "lost 1 of 17 subs" instead of # silently dropping ~6% of resources and letting safe_sweep # treat them as real deletions. The drop-ratio floor is now # configurable via ``platform_settings['cmdb_sync.sweep_drop_tolerance_percent']`` # (default 20% — see ``sweep_settings.py``), so drops smaller # than the configured tolerance are absorbed silently by # design — the ``partial_failures`` channel is what flips the # capability to ``partial_failure`` status and pages the # failure-notifier even when the floor itself stays green. for m in members: account_id = m["account_id"] async with self._impersonate(account_id) as scope_id: if scope_id is None: self.partial_failures.append({ "leg": f"sub_account:{account_id}", "capability": fetcher_name, "reason": "assume_role_failed", }) continue try: sub_results = await getattr(_aliyun.AliyunProvider, fetcher_name)(self) except CapabilityNotSupported as exc: # Service unsubscribed at the sub-account level — # log and skip just this sub-account, don't fail # the whole fan-out (other sub-accounts may have # the service active). Service unsubscribed is the # ONLY skip path that doesn't count as a partial # failure (it's expected for sub-accounts that don't # use the capability). logger.info( "AlicloudIntl %s on %s: service not subscribed " "(skipping sub-account): %s", fetcher_name, account_id, str(exc)[:200], ) continue except ProviderAuthError as exc: logger.info( "AlicloudIntl %s on %s: auth error (skipping): %s", fetcher_name, account_id, str(exc)[:200], ) self.partial_failures.append({ "leg": f"sub_account:{account_id}", "capability": fetcher_name, "reason": "auth_error", }) continue except Exception as exc: # noqa: BLE001 logger.warning( "AlicloudIntl %s on %s: %s — sub-account skipped", fetcher_name, account_id, str(exc)[:200], ) self.partial_failures.append({ "leg": f"sub_account:{account_id}", "capability": fetcher_name, "reason": f"unexpected:{type(exc).__name__}", }) continue for r in sub_results: _stamp_account_id(r, account_id) out.extend(sub_results) return out # ── Capability-method overrides — all delegate to _fanout ───── async def list_servers(self) -> list[CloudServer]: return await self._fanout("list_servers") async def list_domains(self) -> list[CloudDomain]: return await self._fanout("list_domains") async def list_dns_records(self) -> list[CloudDnsRecord]: return await self._fanout("list_dns_records") async def list_load_balancers(self) -> list[CloudLoadBalancer]: return await self._fanout("list_load_balancers") async def list_cdn_domains(self) -> list[CloudCdnDomain]: return await self._fanout("list_cdn_domains") async def list_object_storage(self) -> list[Any]: return await self._fanout("list_object_storage") def _cms_endpoint(self) -> str: # Intl tenant's CloudMonitor lives in the ap-southeast-1 region # (same site as the BSS host); the cn-hangzhou central endpoint # the mainland base uses is not reachable with intl credentials. return "https://metrics.ap-southeast-1.aliyuncs.com/" async def list_server_metrics( self, targets: list[ServerMetricTarget], *, lookback_days: int = 14, period_seconds: int = 3600, ) -> list[ServerMetricPoint]: """Fan CloudMonitor across master + every RD member. The parent query filters by the target instance-id set, and under impersonation only that sub-account's instances return data — so each sub contributes its own slice with no cross-account overlap. ``_fanout`` can't be reused (it calls zero-arg ``list_*``); this mirrors its master + per-sub-account structure with the ``targets`` argument threaded through. Deduped on (instance_id, timestamp) for safety.""" members = await self._list_member_accounts() parent = _aliyun.AliyunProvider.list_server_metrics seen: set[tuple[str, Any]] = set() out: list[ServerMetricPoint] = [] def _absorb(points: list[ServerMetricPoint]) -> None: for p in points: key = (p.instance_id, p.timestamp) if key in seen: continue seen.add(key) out.append(p) if not members: return await parent( self, targets, lookback_days=lookback_days, period_seconds=period_seconds, ) # Master account first. try: _absorb(await parent( self, targets, lookback_days=lookback_days, period_seconds=period_seconds, )) except (CapabilityNotSupported, ProviderAuthError): raise except Exception as exc: # noqa: BLE001 logger.warning( "AlicloudIntl list_server_metrics on master failed: %s", str(exc)[:200], ) for m in members: account_id = m["account_id"] async with self._impersonate(account_id) as scope_id: if scope_id is None: self.partial_failures.append({ "leg": f"sub_account:{account_id}", "capability": "metrics", "reason": "assume_role_failed", }) continue try: _absorb(await parent( self, targets, lookback_days=lookback_days, period_seconds=period_seconds, )) except CapabilityNotSupported: continue except Exception as exc: # noqa: BLE001 self.partial_failures.append({ "leg": f"sub_account:{account_id}", "capability": "metrics", "reason": f"unexpected:{type(exc).__name__}", }) continue return out # ── Server lifecycle — explicitly NOT inherited from AliyunProvider. # # PR #559 added power_on / power_off / restart_server / destroy_server # to AliyunProvider, hardcoded to ``https://ecs.aliyuncs.com/`` and # signed with the parent's single AK. AliCloud Intl tenants: # * use international endpoints (``ecs..aliyuncs.com``), # * have resources spread across N member sub-accounts that the # master AK does NOT directly control, # so blindly inheriting the parent methods would hit the wrong # endpoint with the wrong credentials — admin-pressed Stop on an # intl instance would either 4xx (endpoint mismatch) or, worse, # silently target a same-id instance on the domestic account. # # Until a proper multi-account lifecycle wiring lands (resolve the # owning sub-account from raw_data['__sub_account_id'] → AssumeRole # → regional ECS endpoint), surface the capability as unsupported # so the router returns 501 and the frontend grays out the menu. # The HTTP 501 path is the same one Azure / GCP / Tencent / KSyun # use, so admins know to operate via the AliCloud console. async def power_on(self, **_kw: Any) -> None: raise CapabilityNotSupported( "alicloud_intl server power-on not implemented (multi-account " "AssumeRole lifecycle pending) — operate via the AliCloud console", ) async def power_off(self, **_kw: Any) -> None: raise CapabilityNotSupported( "alicloud_intl server power-off not implemented (multi-account " "AssumeRole lifecycle pending) — operate via the AliCloud console", ) async def restart_server(self, **_kw: Any) -> None: raise CapabilityNotSupported( "alicloud_intl server restart not implemented (multi-account " "AssumeRole lifecycle pending) — operate via the AliCloud console", ) async def destroy_server(self, **_kw: Any) -> None: raise CapabilityNotSupported( "alicloud_intl server destroy not implemented (multi-account " "AssumeRole lifecycle pending) — operate via the AliCloud console", ) # ── RAM users (IAM capability) — newly declared above ───────── async def list_iam_users(self) -> list[dict[str, Any]]: """Per-sub-account RAM users. For each member account, AssumeRole then ``ListUsers`` against ``ram.aliyuncs.com``. Master-account RAM users (typically the Cloud SSO service account + a few admins) are also included, tagged ``__sub_account_id="__master__"``. """ members = await self._list_member_accounts() out: list[dict[str, Any]] = [] # Per-account failures are collected and re-raised at the end so an # incomplete snapshot DISABLES any downstream mark-and-sweep rather # than returning a truncated ``out`` that looks complete. A throttled # AssumeRole / ListUsers on one account previously produced 0 users # for that account with ``runs.status == ok``; a CMDB IAM sweep would # then treat that account's active RAM users as "vanished" and could # revoke in-use users. Mirrors the raise-on-failure posture of the # UCloud / Aliyun master drivers (feedback_handler_status_must_raise). failures: list[str] = [] # Master. try: out.extend(await self._list_ram_users_current_credentials("__master__")) except Exception as exc: # noqa: BLE001 logger.warning( "AlicloudIntl list_iam_users master: %s", str(exc)[:200], ) failures.append(f"__master__: {str(exc)[:200]}") for m in members: account_id = m["account_id"] async with self._impersonate(account_id) as scope_id: if scope_id is None: # AssumeRole into this member account failed — an # unreadable account is an incomplete snapshot, not an # empty one. Record it so the sweep is disabled. failures.append(f"{account_id}: AssumeRole failed") continue try: out.extend( await self._list_ram_users_current_credentials(account_id), ) except Exception as exc: # noqa: BLE001 logger.warning( "AlicloudIntl list_iam_users %s: %s", account_id, str(exc)[:200], ) failures.append(f"{account_id}: {str(exc)[:200]}") continue if failures: raise RuntimeError( "AlicloudIntl list_iam_users: " f"{len(failures)} account(s) failed — refusing to return a " "partial RAM-user snapshot (would let a downstream sweep " f"revoke in-use users). Failures: {'; '.join(failures)}", ) return out async def _list_ram_users_current_credentials( self, account_id: str, ) -> list[dict[str, Any]]: """RAM ``ListUsers`` against the credentials currently set on ``self`` (master AK or a swapped-in temp token). Page-level errors propagate to the outer per-sub-account isolation layer in :meth:`list_iam_users` rather than being swallowed here. Pre-fix this method caught any exception during pagination, logged at ``info``, and broke the loop — callers received partial page-1 results with no signal that pages 2+ were lost. A transient 429 on page 2 would silently drop ~50% of a sub-account's RAM users from the IAM-sync output and the cron's ``runs.status`` landed ``ok`` regardless. Per ``feedback_handler_status_must_raise``: handlers must raise on non-trivial failure so the cron run is marked failed and the failure-notifier can dedup-page the operator. """ users: list[dict[str, Any]] = [] marker: str | None = None while True: params = _aliyun._common_params("ListUsers", self._ak) params.update({"Version": "2015-05-01", "MaxItems": "100"}) if marker: params["Marker"] = marker data = await self._call("https://ram.aliyuncs.com/", params) items = (data.get("Users") or {}).get("User") or [] for u in items: user_name = str(u.get("UserName") or "") user_id_external = str(u.get("UserId") or "") attached, groups = await self._ram_user_policies_and_groups( user_name, ) # Aliyun timestamps are ISO 8601 with ``Z`` suffix. create_date = _parse_iso8601(u.get("CreateDate")) last_login = _parse_iso8601(u.get("LastLoginDate")) users.append({ "user_name": user_name, "arn": ( f"acs:ram::{account_id}:user/{user_name}" if account_id != "__master__" else None ), "user_id_external": user_id_external, "create_date": create_date, "last_login_at": last_login, "attached_policies": attached, "groups": groups, "mfa_enabled": False, # RAM exposes via separate API # NOT "disabled" — nobody asked RAM. Saying # "off" about a question never put to the # provider is what made the Tencent tenant # report 25 users with no MFA while several # had it on (2026-08-12). "mfa_state": "unknown", "tags": {"__sub_account_id": account_id} if account_id else {}, "raw": { "Email": u.get("Email"), "DisplayName": u.get("DisplayName"), "MobilePhone": u.get("MobilePhone"), "__sub_account_id": account_id, }, }) if not data.get("IsTruncated"): break marker = data.get("Marker") if not marker: break return users async def _ram_user_policies_and_groups( self, user_name: str, ) -> tuple[list[dict], list[str]]: """Best-effort per-user policy + group enrichment. Failures are demoted to debug logs — a user with unreadable policies still produces a CMDB IAM-user row, just without the policies array. Doing this per-user keeps RAM rate-limit hits bounded; we don't need 100% coverage to surface grants. """ attached: list[dict] = [] groups: list[str] = [] # ListPoliciesForUser try: params = _aliyun._common_params("ListPoliciesForUser", self._ak) params.update({ "Version": "2015-05-01", "UserName": user_name, }) data = await self._call("https://ram.aliyuncs.com/", params) for p in (data.get("Policies") or {}).get("Policy", []) or []: attached.append({ "name": str(p.get("PolicyName") or ""), "arn": str(p.get("PolicyName") or ""), # RAM has no ARN; reuse name }) except Exception as exc: # noqa: BLE001 logger.debug( "AlicloudIntl ListPoliciesForUser %s: %s", user_name, exc, ) # ListGroupsForUser try: params = _aliyun._common_params("ListGroupsForUser", self._ak) params.update({ "Version": "2015-05-01", "UserName": user_name, }) data = await self._call("https://ram.aliyuncs.com/", params) for g in (data.get("Groups") or {}).get("Group", []) or []: gn = str(g.get("GroupName") or "") if gn: groups.append(gn) except Exception as exc: # noqa: BLE001 logger.debug( "AlicloudIntl ListGroupsForUser %s: %s", user_name, exc, ) return attached, groups def _stamp_account_id(obj: Any, account_id: str) -> None: """Write ``__sub_account_id`` into the object's ``raw`` dict in-place. Resource dataclasses (CloudServer / CloudDomain / etc.) carry the untyped ``raw`` payload through to ``cmdb_*.raw_data`` so the sync upsert can read this back when adding ``cloud_scope_id`` resolution in a later step. For ``list_object_storage`` (which returns dicts rather than dataclasses), stamp the dict directly. """ if isinstance(obj, dict): # Specialized syncs persist only ``item['raw']``. Stamping the # envelope itself used to discard the account id for every OSS # bucket, leaving hundreds of resources unable to inherit their # Resource Directory account owner. raw = obj.get("raw") if not isinstance(raw, dict): raw = {} obj["raw"] = raw raw.setdefault("__sub_account_id", account_id) # Preserve the historical envelope stamp for callers that inspect # provider output before specialized persistence. obj.setdefault("__sub_account_id", account_id) return raw = getattr(obj, "raw", None) if isinstance(raw, dict): raw.setdefault("__sub_account_id", account_id) # Also mirror onto tags for resources that have a tags dict — UI # filtering can pick this up cheaply. tags = getattr(obj, "tags", None) if isinstance(tags, dict): tags.setdefault("__sub_account_id", account_id) def _parse_iso8601(s: Any) -> datetime | None: if not s or not isinstance(s, str): return None try: return datetime.fromisoformat(s.replace("Z", "+00:00")) except ValueError: return None