"""Aliyun cloud driver. Wraps the existing :class:`AliyunProvider` for CMDB sync AND extends it with the multi-cloud account lifecycle methods (list_scopes, list_scope_roles, grant_access, revoke_access, list_billing_daily). The lifecycle methods hit Aliyun REST endpoints directly via httpx using **ACS3-HMAC-SHA256** signing — we deliberately don't pull the ``alibabacloud-*`` SDK family (20+ packages, 100MB+) just for two endpoints per service. Each product has its own base domain, so the signer accepts ``host``, ``version``, and ``action`` and routes. Scope semantics depend on config: * Tenant mode "sso" — use Resource Directory ListAccounts to enumerate member accounts; grants are RAM users on the member + SSO role attachment (``ram.AttachPolicyToUser``). * Tenant mode "iam" (rare; for Aliyun single-account tenants) — one scope = self, grants create RAM users directly on this tenant. """ from __future__ import annotations import asyncio import datetime as _dt import hashlib import hmac import json import logging import re import urllib.parse import uuid from datetime import date, datetime, timedelta from typing import Any import httpx from app.domains.cmdb.providers.aliyun import AliyunProvider from app.domains.cmdb.providers.exceptions import ProviderAuthError from app.integrations.driver_base import ( COST_BUCKET_ADJUSTMENT, COST_BUCKET_OTHER, COST_BUCKET_USAGE, CloseScopeResult, CloudAccessContext, CloudAccountRevokeContext, CloudDailyCost, CloudGrantResult, CloudRoleInfo, CloudScopeInfo, CloudScopeRef, CostFact, DiscoveredGrant, ProvisioningResult, ResourceCreatorInfo, ResourceCreatorWindowExceeded, ScopeClosureState, ) from app.integrations.drivers._cloud_shared import ( _CloudDriverBase, account_id_candidates, ) from app.integrations.registry import register_driver from app.integrations.schemas import IntegrationFieldInfo logger = logging.getLogger(__name__) # ── BSS / OpenAPI 限流重试（2026-07-29）───────────────────────────── # # 阿里云 OpenAPI 用 **HTTP 400 + ``Code=Throttling.User``** 表达限流 # （不是 429），且 ``_acs3_call`` 此前对任何 >=400 直接抛 —— 于是限流 # 一次就整块失败。2026-07-29 的 9 个月回填实测：4 路并行拉 BSS # ``DescribeInstanceBill``（25 个子账号 × 按天分页）触发 219 次 # ``Throttling.User``，绝大多数子账号当块无数据落库。本仓库其它云驱动 # （AWS/Azure/腾讯）都有限流重试，阿里云是唯一缺口。 # # 限流语义是「请求被拒、未执行」，所以重试对**读写都安全**（不是 # "可能已生效" 的超时那类）。退避固定阶梯（不引随机，便于测试断言）， # 上限 10s;耗尽后抛与原来完全一致的 RuntimeError。 _ACS3_THROTTLE_MAX_ATTEMPTS = 5 _ACS3_THROTTLE_BACKOFF_SEC = (1.0, 2.0, 4.0, 8.0) def _is_acs3_throttled(status_code: int, body_text: str) -> bool: """阿里云 OpenAPI 的限流形状：429，或 400/403 且 ``Code`` 以 ``Throttling`` 开头（``Throttling.User`` / ``Throttling.Api`` / ``Throttling`` 都算）。只看 ``Code`` 字段,不拿 Message 做模糊匹配 （避免把「参数里带 throttling 字样」的业务错误误判成限流）。""" if status_code == 429: return True if status_code not in (400, 403): return False try: code = str((json.loads(body_text) or {}).get("Code") or "") except (json.JSONDecodeError, TypeError, AttributeError): return False return code.startswith("Throttling") # 只读 action 前缀。阿里云 OpenAPI 的动词命名很规整:读取一律是 # Describe* / List* / Query* / Get*,变更是 Create/Delete/Attach/ # Update/Set/Modify/Add/Remove/… _ACS3_READONLY_ACTION_PREFIXES = ("Describe", "List", "Query", "Get") # 可重试的临时服务端错误。**只对只读 action 重试** —— 5xx 语义是 # 「结果未知」(请求可能已在服务端生效),对变更类 action 重试会有重复 # 下发的风险(例如 AttachPolicyToUser 收到 503 但实际已生效)。限流 # (上面那个)不同:那是明确「被拒、未执行」,读写都能安全重试。 # # 2026-07-29 生产回填实测:9 个月阿里云回填里有 1 次 # ``503 ServiceUnavailable``(DescribeInstanceBill 2026-07-14), # 直接让那个子账号的整个 30 天窗口没有数据落库,只能人工补跑。 # Azure 的 ``_post_with_retry`` 早已覆盖 429+5xx;这里对齐,但更保守 # （限定只读）。 _ACS3_TRANSIENT_STATUSES = frozenset({500, 502, 503, 504}) def _is_acs3_retryable_5xx(status_code: int, action: str) -> bool: """临时 5xx **且** action 是只读的 → 可重试。""" if status_code not in _ACS3_TRANSIENT_STATUSES: return False return action.startswith(_ACS3_READONLY_ACTION_PREFIXES) # ── Default-promoted Aliyun RAM system policies ───────────────────── # Curated set mirrors Azure's "Privileged administrator roles" idea: # the 5 roles a regular employee would realistically request as their # *primary* role. Everything else (~250 system policies) lives behind # the ``[built-in]``-tagged extra-roles multi-select. # # Override via the ``cloud.aliyun.default_policies`` platform setting # (migration 071 seeds this as the canonical source of truth; this # driver-level list is the no-DB fallback for greenfield tests). # # The ``role_key`` shape is ``":"`` — same key # the rest of the driver uses to call ``AttachPolicyToUser``. _ALIYUN_DEFAULT_FALLBACK: list[dict[str, Any]] = [ { "role_key": "System:AdministratorAccess", "display_name": "AdministratorAccess", "description": "管理员 — 全资源读写 + RAM 管理。慎重授予。", }, { "role_key": "System:ReadOnlyAccess", "display_name": "ReadOnlyAccess", "description": "只读全部资源。", }, { "role_key": "System:AliyunRAMFullAccess", "display_name": "AliyunRAMFullAccess", "description": "RAM 用户 / 角色 / 策略全权管理。", }, { "role_key": "System:AliyunResourceManagerFullAccess", "display_name": "AliyunResourceManagerFullAccess", "description": "资源目录全权管理(账号 / 资源夹 / 控制策略)。", }, { "role_key": "System:AliyunBSSFullAccess", "display_name": "AliyunBSSFullAccess", "description": "账单 / 发票 / 订单 / 优惠券全权管理。", }, ] _ALIYUN_TAG_PAIR_RE = re.compile( # ``key: value:`` pairs separated by `;`. Values may # contain spaces, colons, hyphens, dots — anything except a # literal ``;`` (which is the chunk separator) or the literal # ``key:`` / ``value:`` prefixes that mark the next field. r"key:(?P[^\s;]+)\s+value:(?P[^;]*?)(?=\s*(?:;|$))", ) def _parse_aliyun_tags( raw: Any, *, instance_id: str = "", ) -> dict[str, str]: """Coerce Aliyun's ``Tag`` field into a clean ``{k: v}`` dict. Aliyun's BSS API emits tags in two shapes depending on the endpoint version: 1. ``"key:env value:prod; key:owner value:alice"`` — a flat string with ``key:`` / ``value:`` markers separated by ``;``. Values can contain spaces (regex captures up to the next ``;`` or end of string). 2. ``[{"TagKey": "env", "TagValue": "prod"}, ...]`` — array of objects (newer APIs). Malformed chunks (missing ``key:`` or ``value:`` markers, or a ``key:`` with no value field) are dropped and a single INFO line is emitted naming the affected resource so an operator can audit. The driver does NOT silently fabricate keys or values. """ if not raw: return {} out: dict[str, str] = {} if isinstance(raw, str): matched = list(_ALIYUN_TAG_PAIR_RE.finditer(raw)) for m in matched: k = m.group("k").strip() v = m.group("v").strip() if k: out[k] = v # Was there content not captured by the regex? Surface it. if not matched and raw.strip(): logger.info( "Aliyun tags on %s did not match key:/value: format: %r", instance_id or "", raw[:200], ) return out if isinstance(raw, list): for tag in raw: if not isinstance(tag, dict): continue k = str(tag.get("Key") or tag.get("TagKey") or "").strip() v = str(tag.get("Value") or tag.get("TagValue") or "").strip() if k: out[k] = v return out logger.info( "Aliyun tags on %s have unexpected type %s; ignoring", instance_id or "", type(raw).__name__, ) return {} def _safe_float(value: Any) -> float | None: """Parse a non-money BSS field without letting it kill the run. ``Usage`` is a usage QUANTITY, not money — a dirty value there (BSS occasionally returns ``"-"`` or a unit-suffixed string) used to raise straight out of the row loop and fail the whole sync. The driver contract is partial-tolerant, and the Tencent / UCloud readers of the same field already guard it. Money fields keep their strict parsing on purpose: a silently-dropped amount is a wrong number, while a dropped quantity is a missing annotation. """ if value in (None, ""): return None try: return float(value) except (TypeError, ValueError): logger.warning("Aliyun bill: unparseable Usage value %r", value) return None # Page ceiling + same-token guard for every BSS bill walk (#2153). # A degenerate BSS response that keeps returning the SAME non-empty # ``NextToken`` loops forever: each call succeeds quickly, so the 30s # per-request httpx timeout never fires and the worker watchdog (whose # own comment admits it cannot see a hung loop) never trips — the job # just pins a scheduled-jobs concurrency slot and hammers the BSS API # indefinitely. The intl subclass wrote these two guards for itself and # documented the failure as real; they belong in the parent so every # walk (and the intl class, which inherits ``list_cost_facts``) gets # them. 500 pages × 300 rows = 150k line items per day, comfortably # above any realistic daily bill. _MAX_BILL_PAGES = 500 # 1000 × 100 = 100k accounts, far above any realistic Resource Directory. # The cap is a last-resort guard against a corrupt/malicious TotalCount. _MAX_SCOPE_PAGES = 1000 # ── Resource Directory member closure ───────────────────────── # Endpoint for the Resource Directory (资源目录) control plane. Global # service: the CN host is correct for the 海外 tenant too — verified # against prod's alicloud_intl RD master 2026-09-01 (GetAccount / # GetResourceDirectory / GetAccountDeletionCheckResult all answer). # Unlike BSS and ActionTrail (which the intl subclass has to override # because they are calling-region pinned), this one needs no override. _RD_HOST = "resourcemanager.aliyuncs.com" _RD_VERSION = "2020-03-31" # Member statuses that mean the member is already gone or on its way # out — the only ones for which ``already_closed=True`` is honest. # Deliberately CONSERVATIVE: anything not listed falls through to a real # close attempt, which fails loudly and changes nothing if the member # turns out to be unclosable. The opposite error — claiming "already # closed" for a live member — silently flips the Helix row to ``closed`` # and cascade-disables every grant while the account keeps running and # keeps billing. So this set must never grow on a guess. # # Two families are pointedly EXCLUDED, both of which the pre-2026-09-01 # test (``status in {...PromoteVerifying...} or "Remov" in status or # "Delete" in status``) wrongly matched: # * ``Promote*`` / ``Switch*`` — ResourceAccount ⇄ CloudAccount type # conversions. A member mid-conversion is fully alive. Every one of # prod's 14 CloudAccount members reports ``SwitchSuccess``. # * ``RemoveFailed`` / ``DeleteFailed`` — the removal FAILED, i.e. the # member is still very much there. _RD_GONE_STATUSES = frozenset({ "Removed", "RemoveSuccess", "Deleting", "DeleteSuccess", }) _RD_NOT_FOUND_CODES = ( "EntityNotExist", "Account.NotExist", "AccountId.NotExist", ) _RD_DENIED_CODES = ("Forbidden", "NoPermission") _RD_RESOURCES_EXISTING_CODES = ( "NotAllowedDelete.HasResource", "NotAllowedDelete.HasUndeleted", "Member.HasResource", "Member.NotAllowedDelete.HasResource", "Member.NotAllowedDelete.HasUndeletedResource", "PrePaidResource", "AccountHasPrePaidResource", ) _RD_REAL_NAME_CODES = ( "NotPassRealNameAuth", "Member.NotAllowedToDoAction.NotPassRealNameAuth", "RealNameAuth.NotPass", ) _RD_PENDING_CHARGES_CODES = ( "Account.Arrears", "Member.Arrears", "HasUnpaidBill", "HasUnsettledBill", "Member.NotAllowedDelete.HasUnpaidBill", "Member.NotAllowedDelete.HasUnsettledBill", "Account.HasUnpaidBill", "Account.HasUnsettledBill", "PaymentMethod.NotSet", "Member.NotAllowedDelete.PaymentMethodAbsent", ) # Aliyun documents the member-deletion pre-check as taking 3–5 minutes, # so this poll is a courtesy for the rare fast case, NOT the mechanism # that carries the close to completion — that is the resumability of # ``_close_rd_account`` itself. Keep the budget short: the admin's HTTP # request is what is being held open, and returning "come back in a few # minutes" costs nothing because re-clicking resumes rather than # restarts. 4 × 4s ≈ 16s, comfortably inside nginx's 300s proxy timeout # and the 600s decommission Redis lock. _RD_PRECHECK_POLL_ATTEMPTS = 4 _RD_PRECHECK_POLL_INTERVAL_SEC = 4.0 def _acs3_truthy(value: Any) -> bool: """Aliyun's JSON booleans arrive as ``True`` or as ``"true"``. Coerce both. Defaults to False for anything unrecognised, which is the safe direction here: the one caller gates a real deletion on it. """ if isinstance(value, bool): return value return str(value or "").strip().lower() == "true" def _rd_check_ids(raw: Any) -> list[str]: """CheckIds out of ``AbandonableChecks``, for ``AbandonableCheckId``. Ids look like ``SP_oss_oss_intl`` / ``SP_pai_pai_intl``. Only items that actually carry a ``CheckId`` are returned — an entry without one cannot be abandoned, and inventing an id would make the whole DeleteAccount call fail on a malformed parameter instead of just that one item going unabandoned. """ items = raw if isinstance(items, dict): for key in ("AbandonableCheck", "AbandonableChecks"): if isinstance(items.get(key), list): items = items[key] break if not isinstance(items, list): return [] out: list[str] = [] for item in items: if not isinstance(item, dict): continue check_id = str(item.get("CheckId") or "").strip() if check_id and check_id not in out: out.append(check_id) return out def _rd_check_items(raw: Any) -> list[str]: """Flatten ``NotAllowReason`` / ``AbandonableChecks`` for humans.""" items = raw if isinstance(items, dict): # Aliyun sometimes wraps repeated elements one level deep. for key in ("NotAllowReason", "AbandonableCheck", "AbandonableChecks"): if isinstance(items.get(key), list): items = items[key] break if not isinstance(items, list): return [] out: list[str] = [] for item in items: if not isinstance(item, dict): continue name = str(item.get("CheckName") or item.get("CheckId") or "").strip() desc = str(item.get("Description") or "").strip() label = " — ".join(p for p in (name, desc) if p) if label: out.append(label) return out @register_driver class AliyunCloudDriver(_CloudDriverBase): provider = "aliyun" display_name = "阿里云" description = ( "阿里云(ECS / 域名 / 云解析 DNS / CDN / OSS / SLB + " "资源目录子账号 + RAM 用户管理 + 离职全量撤销)。" ) provider_cls = AliyunProvider supported_capabilities = frozenset({ "servers", "domains", "dns_records", "cdn", "object_storage", "load_balancers", "user_revoke", }) capabilities = [ "servers", "domains", "dns_records", "cdn", "object_storage", "load_balancers", "user_revoke", ] # Aliyun RAM users + attached policies are tenant-wide: one RAM # user's policies apply across every account in the Resource # Directory. Run grants-sync once per integration, tagging the # tenant-wide (user, policy) pairs with the first scope the # service happens to hit. grants_scope_wide = False # Aliyun AdministratorAccess is the system policy that represents # full-control ownership; custom admin policies can be added here # by the admin via platform_settings if a tenant renames it. owner_role_keys = frozenset({"AdministratorAccess"}) # Role governance (see app/integrations/cloud_role_governance.py). # RAM / ResourceManager full access = IAM/org-write escalation → # restricted (explicit, overriding the "FullAccess" elevated keyword). # System:AdministratorAccess is restricted by admin keyword. Other # System:*FullAccess fall to elevated via keyword; BSS billing admin # listed explicitly. Role keys are "System:Name"/"Custom:Name". restricted_role_keys = frozenset({ "System:AdministratorAccess", "System:AliyunRAMFullAccess", "System:AliyunResourceManagerFullAccess", }) elevated_role_keys = frozenset({ "System:AliyunBSSFullAccess", }) # CNY-native billing: list_cost_facts needs a CNY→USD rate. requires_cny_fx = True # Per-instance cache of ``GetResourceDirectory`` (see # ``_get_resource_directory``) — RD Root folder + 管理账号 id. _rd_directory_cache: dict[str, Any] | None = None # ``account_mode`` is computed from config via the property below; # no class-level override needed (parent CloudDriver defaults to "sso"). config_schema = [ IntegrationFieldInfo( key="access_key_id", label="AccessKey ID", is_secret=True, required=True, ), IntegrationFieldInfo( key="access_key_secret", label="AccessKey Secret", is_secret=True, required=True, ), IntegrationFieldInfo( key="account_mode", label="账号模式", is_secret=False, required=False, default="sso", placeholder="sso", ), # CloudSSO metadata — optional. Only recorded for display / # audit; the RAM-user flow above doesn't call the CloudSSO # AssumeRole endpoint. Provided so ops can point future admins # at the right directory / access configuration without having # to dig into Alibaba Cloud console. IntegrationFieldInfo( key="directory_id", label="CloudSSO Directory ID (可选)", is_secret=False, required=False, placeholder="d-xxxxxxxxxxxxxxxx", ), IntegrationFieldInfo( key="access_configuration_id", label="CloudSSO AccessConfiguration ID (可选)", is_secret=False, required=False, placeholder="ac-xxxxxxxxxxxxxxxx", ), IntegrationFieldInfo( key="role_arn", label="AssumeRole ARN (可选)", is_secret=False, required=False, placeholder="acs:ram:::role/", ), IntegrationFieldInfo( key="region", label="Default Region (ECS/SLB)", is_secret=False, required=False, default="cn-shanghai", placeholder="cn-shanghai", ), # Resource Directory folder under which ``CreateResourceAccount`` # will place new sub-accounts. OPTIONAL — Aliyun's # ``CreateResourceAccount`` treats ``ParentFolderId`` as optional # and, when omitted, creates the member directly under the RD # **Root** folder (same default as AWS Organizations). So # ``create_scope`` no longer hard-requires this field: set it # only when the org wants new sub-accounts filed under a specific # 资源夹. The credential admin can find folder IDs on # https://resourcemanager.console.aliyun.com under "资源夹". IntegrationFieldInfo( key="rd_parent_folder_id", label="Resource Directory 资源夹 ID (新建子账号用,留空=Root)", is_secret=False, required=False, placeholder="fd-xxxxxxxxxx (留空则建在 Root 资源夹)", ), # Reseller / billing model for the new account. Aliyun supports # "ResellAccount" (paid by master) and "NormalAccount" (separate # billing). Default "ResellAccount" matches what most teams want # — the new account is billed against the master credential, no # extra paperwork. Override here only if the org uses isolated # billing per sub-account. IntegrationFieldInfo( key="rd_account_type", label="子账号计费类型 (新建子账号用)", is_secret=False, required=False, default="ResellAccount", placeholder="ResellAccount", ), # 结算账号 (console: 「修改结算方式」). Aliyun's # ``CreateResourceAccount`` treats ``PayerAccountId`` as optional # and DEFAULTS TO 「新成员自主付款」 when omitted — which is never # what PingCAP wants (every sub-account must settle against the # RD 管理账号 sa@pingcap.com). Leave this blank and the driver # resolves the RD master via ``GetResourceDirectory`` and sends # it as the payer (=「使用管理账号为新成员付款」). Set an explicit # member AccountId for 「使用已有成员为新成员付款」, or the literal # ``self`` to deliberately create a self-paying account. IntegrationFieldInfo( key="rd_payer_account_id", label="新建子账号的结算账号 (留空=管理账号代付, self=自主付款)", is_secret=False, required=False, placeholder="留空即管理账号代付", ), # Optional per-integration FX override. Normally the platform # uses ``platform_settings.ai_spending.cny_to_usd_rate`` # (refreshed weekly from frankfurter.dev / ECB). Set this only # if finance needs a fixed monthly rate that's independent of # the live rate — leaving it blank routes through the # platform value. See app.core.fx_rate.resolve_cny_per_usd. IntegrationFieldInfo( key="cny_per_usd", label="CNY→USD 汇率覆盖 (可选,留空走平台值)", is_secret=False, required=False, placeholder="7.20", ), ] @property def account_mode(self) -> str: # type: ignore[override] return str(self.config.get("account_mode") or "sso").strip() or "sso" # ── ACS3 signer (HMAC-SHA256) ────────────────────────────── async def _acs3_call( self, *, host: str, version: str, action: str, method: str = "POST", params: dict[str, Any] | None = None, body: dict[str, Any] | None = None, ) -> dict[str, Any]: ak = str(self.config.get("access_key_id") or "").strip() sk = str(self.config.get("access_key_secret") or "").strip() if not ak or not sk: raise RuntimeError("Aliyun access key/secret not configured") now_utc = _dt.datetime.now(tz=_dt.UTC) iso_time = now_utc.strftime("%Y-%m-%dT%H:%M:%SZ") nonce = uuid.uuid4().hex # List values are sent as a JSON array in ONE query parameter — # ``AbandonableCheckId=["SP_oss_oss_intl","SP_pai_pai_intl"]`` — # matching the Aliyun CLI's own # ``--AbandonableCheckId '["SP_oss_oss"]'``. # # **Measured, not assumed** (2026-09-02, prod account # 5995487234252122). This first shipped as the classic Aliyun RPC # repeated form (``AbandonableCheckId.1`` / ``.2``) because that # is what the older SDKs emit for array params, and a unit test # asserting exactly that shape passed. It was wrong: Aliyun # ignored the parameter entirely, the deletion was accepted and # then died at ``CheckFailed`` listing the very items we had # tried to abandon. Re-submitted byte-for-byte identically except # for the encoding, the same account went ``Checking`` → # ``Deleting`` with an empty ``FailReasonList`` and a scheduled # ``DeletionTime``. A unit test can only prove the code does what # you told it to; only the provider can tell you what it accepts. query: dict[str, str] = {} for key, value in (params or {}).items(): if isinstance(value, (list, tuple)): query[key] = json.dumps(list(value), separators=(",", ":")) else: query[key] = str(value) body_bytes = ( json.dumps(body, separators=(",", ":")).encode() if body is not None else b"" ) content_sha = hashlib.sha256(body_bytes).hexdigest() canonical_query = "&".join( f"{urllib.parse.quote(k, safe='-_.~')}=" f"{urllib.parse.quote(v, safe='-_.~')}" for k, v in sorted(query.items()) ) headers = { "host": host, "x-acs-action": action, "x-acs-content-sha256": content_sha, "x-acs-date": iso_time, "x-acs-signature-nonce": nonce, "x-acs-version": version, } if body_bytes: headers["content-type"] = "application/json; charset=utf-8" canonical_headers = "".join( f"{k}:{v}\n" for k, v in sorted(headers.items()) ) signed_headers = ";".join(sorted(headers.keys())) canonical_request = ( f"{method}\n/\n{canonical_query}\n{canonical_headers}\n" f"{signed_headers}\n{content_sha}" ) hashed_canonical = hashlib.sha256( canonical_request.encode(), ).hexdigest() string_to_sign = f"ACS3-HMAC-SHA256\n{hashed_canonical}" signature = hmac.new( sk.encode(), string_to_sign.encode(), hashlib.sha256, ).hexdigest() auth = ( f"ACS3-HMAC-SHA256 Credential={ak}," f"SignedHeaders={signed_headers},Signature={signature}" ) headers["authorization"] = auth url = f"https://{host}/?{canonical_query}" if canonical_query else f"https://{host}/" # 限流重试:见模块顶 ``_is_acs3_throttled``。签名已算好且与 # 时间戳绑定,但阿里云的签名有效期为 15 分钟,最坏累计退避 15s # 远在窗口内,所以可以复用同一份 headers 重发。 async with httpx.AsyncClient(timeout=30, trust_env=False) as client: for attempt in range(_ACS3_THROTTLE_MAX_ATTEMPTS): resp = await client.request( method, url, headers={k: str(v) for k, v in headers.items()}, content=body_bytes or None, ) retryable = ( _is_acs3_throttled(resp.status_code, resp.text) or _is_acs3_retryable_5xx(resp.status_code, action) ) if not retryable: break if attempt >= len(_ACS3_THROTTLE_BACKOFF_SEC): break # 重试预算耗尽 —— 落到下面的抛错分支 delay = _ACS3_THROTTLE_BACKOFF_SEC[attempt] logger.warning( "Aliyun %s transient failure (status=%s attempt=%d/%d) " "— backing off %.0fs", action, resp.status_code, attempt + 1, _ACS3_THROTTLE_MAX_ATTEMPTS, delay, ) await asyncio.sleep(delay) if resp.status_code >= 400: # Round-42 audit: Aliyun BSS occasionally echoes the # AccessKey-id in signature-mismatch (400) diagnostics. The # raw response text lands in capability_results['error'], # failure-notifier cards, and 30+ day log retention; wrap # through the shared redactor so credential fragments are # masked before format-stringing. from app.domains.cmdb.providers.exceptions import ( redact_provider_error, ) raise RuntimeError( f"Aliyun {action} {resp.status_code}: " f"{redact_provider_error(resp.text, max_len=400)}", ) try: return resp.json() except json.JSONDecodeError: # Pre-launch P1 fix (Round 45, CR-207): scrub raw text # before surfacing — Aliyun OpenAPI errors sometimes echo # the request signature in metadata, which may contain # credential fragments. Same family bug as the SF token # endpoint echo that Round-44 fixed. from app.domains.cmdb.providers.exceptions import ( # noqa: PLC0415 redact_provider_error, ) return {"raw": redact_provider_error(resp.text, max_len=400)} # ── Scope sync: Resource Directory ListAccounts ───────── async def _own_account_id(self) -> str: """Resolve the canonical Aliyun AccountId via STS. Same role AWS's ``sts:GetCallerIdentity`` plays on the AWS driver: it's the only thing we trust to identify the tenant for ``cmdb_cloud_scopes.external_id`` and downstream billing joins. There is **no fallback to the Helix integration UUID** — production incident 2026-04-23 (AWS SG) showed how that creates ghost scope rows that collect stale grants. STS failure raises ``ProviderAuthError`` and the sync job fails loudly so the admin fixes credentials first. AccountId is a decimal-digit string (typically 16 digits but Aliyun has not historically guaranteed exact width); validate digits-only with a length floor that excludes obvious misconfigurations. """ try: data = await self._acs3_call( host="sts.aliyuncs.com", version="2015-04-01", action="GetCallerIdentity", method="POST", ) except RuntimeError as exc: raise ProviderAuthError( f"Aliyun STS GetCallerIdentity failed — cannot resolve " f"AccountId for integration {self.integration_id}: {exc}", ) from exc account = str(data.get("AccountId") or "").strip() if not account.isdigit() or len(account) < 10: raise ProviderAuthError( f"Aliyun STS returned a non-canonical AccountId " f"({account!r}) for integration {self.integration_id}; " f"refusing to write a scope row.", ) return account async def _master_account_id(self) -> str | None: """Resolve + cache the RD master (payer) AccountId for billing. BSS ``DescribeInstanceBill`` on a Resource Directory master returns the consolidated bill where each row carries ``OwnerID`` (resource owner = sub-account). Some rows have an EMPTY ``OwnerID`` — those are master-account-allocated charges (the master's own OSS buckets, consolidated platform fees, etc.; their ``BillAccountID`` is the master/payer). The per-scope billing methods fan out per scope and filter client-side, so an empty-OwnerID row must be attributed to EXACTLY ONE scope — the master — or it gets counted once per sub-account scope and over-reports tenant spend (verified prod 2026-06-17: ~$16.8K / 30d duplicated across 23 alicloud_intl scopes). The master AccountId == the credential's own account (``_own_account_id`` / the ``BillAccountID`` on every row). Cached on the instance because the cost / billing crons call the per-scope methods once per scope (23×) on a single driver instance — without the cache each scope would re-issue an STS ``GetCallerIdentity`` call. Returns ``None`` (not raising) when STS resolution fails, so the billing methods degrade to the legacy keep-for-every-scope posture (tolerate the over-count) rather than dropping ALL billing for the window. Scopes_sync resolves the same id via ``list_scopes`` earlier in the cron, so a transient STS outage almost never reaches this path. """ cached = getattr(self, "_master_account_id_value", "__unset__") if cached != "__unset__": return cached resolved: str | None try: resolved = await self._own_account_id() except Exception as exc: # noqa: BLE001 — degrade, don't drop billing logger.warning( "Aliyun integration %s: could not resolve RD master " "AccountId for empty-OwnerID billing attribution (%s); " "falling back to legacy keep-for-every-scope posture " "(may over-count master-allocated lines until STS " "recovers).", self.integration_id, exc, ) resolved = None self._master_account_id_value = resolved return resolved @staticmethod def _billing_row_belongs_to_scope( owner_id: str, owner_filter: str | None, master_id: str | None, ) -> bool: """Decide whether a BSS bill row is attributed to this scope. Single source of truth shared by ``list_cost_facts`` and the ``list_billing_daily`` overrides so cost_facts and billing_daily stay reconciled by construction. * ``owner_filter is None`` → tenant-wide call (scope=None): keep every row (single pass, no fan-out). * row's ``OwnerID`` matches this scope → keep. * empty ``OwnerID`` (master-allocated) → keep ONLY for the master scope. When ``master_id`` is unknown (STS failed) we cannot identify the master, so keeping the row for EVERY scope fans it out N× → the org total and budget alerts over-report by ~(N-1)× the master-allocated portion (and the per-window re-ingest only partially self-heals — dates that roll out of the moving window stay corrupted). SKIP it instead: an outage-window under-count that self-heals cleanly on the next successful sync is far safer than a silent, partially-permanent N× over-count. """ if not owner_filter: return True if owner_id == owner_filter: return True if owner_id == "": return master_id is not None and owner_filter == master_id return False async def list_scopes(self) -> list[CloudScopeInfo]: if self.account_mode == "iam": alias = str(self.config.get("_alias") or self.integration_id) account_id = await self._own_account_id() return [CloudScopeInfo( scope_type="standalone_iam", external_id=account_id, display_name=alias, status="active", raw={"mode": "iam", "aliyun_account_id": account_id}, )] # Resolve the RD master (management/payer) AccountId up front so we # can authoritatively mark the master row in-place — whether it # arrives via ``ListAccounts`` (some directories DO return the # management account, with ``Type='ResourceDirectoryMaster'``) or # only via the STS backstop append below. The ``rd_master`` marker # is what every owner-governance surface keys off to treat the # master as owner-exempt infra, so it must be set regardless of # which path emits the row. This identity is part of the authoritative # snapshot, not optional enrichment: ListAccounts commonly omits the # management account, so swallowing an STS failure would return a # member-only snapshot and let the downstream sweep close the master. master_id: str = await self._own_account_id() out: list[CloudScopeInfo] = [] seen_account_ids: set[str] = set() expected_total: int | None = None page = 1 while page <= _MAX_SCOPE_PAGES: try: data = await self._acs3_call( host="resourcemanager.aliyuncs.com", version="2020-03-31", action="ListAccounts", params={"PageNumber": page, "PageSize": 100}, ) except RuntimeError as e: # Not part of a Resource Directory — one-account tenant. # Resolve the real Aliyun AccountId via STS instead of # silently using the integration UUID. STS failure # raises ProviderAuthError (no fallback — see # _own_account_id docstring). if ( page == 1 and "Aliyun ListAccounts 400:" in str(e) and "ResourceDirectoryNotInUse" in str(e) ): alias = str(self.config.get("_alias") or self.integration_id) # Reuse the master_id already resolved above (same STS # GetCallerIdentity) instead of re-calling — and DON'T # mark rd_master here: a single-account tenant's only # scope is a normal account that must stay owner-governed. account_id = master_id or await self._own_account_id() return [CloudScopeInfo( scope_type="aliyun_account", external_id=account_id, display_name=alias, status="active", raw={"note": "no-rd", "aliyun_account_id": account_id}, )] raise if "Accounts" not in data or "TotalCount" not in data: raise RuntimeError( "Aliyun ListAccounts omitted Accounts or TotalCount; " "refusing a non-authoritative scope snapshot", ) accounts_payload = data.get("Accounts") if isinstance(accounts_payload, dict): accounts_payload = accounts_payload.get("Account") if accounts_payload is None: accounts: list[dict[str, Any]] = [] elif isinstance(accounts_payload, list) and all( isinstance(account, dict) for account in accounts_payload ): accounts = accounts_payload else: raise RuntimeError( "Aliyun ListAccounts returned an invalid Accounts shape; " "refusing a non-authoritative scope snapshot", ) total_value = data["TotalCount"] # Alibaba documents TotalCount as an integer. Do not coerce: # ``int(True) == 1`` and ``int(1.9) == 1`` would otherwise let a # malformed authority field satisfy the destructive-sweep guard. if ( not isinstance(total_value, int) or isinstance(total_value, bool) or total_value < 0 ): raise RuntimeError( "Aliyun ListAccounts returned a non-integral or negative " "TotalCount; " "refusing a non-authoritative scope snapshot", ) total = total_value if expected_total is None: expected_total = total elif total != expected_total: raise RuntimeError( "Aliyun ListAccounts changed TotalCount during pagination " f"({expected_total} -> {total} at page={page}); refusing " "a non-atomic scope snapshot", ) already_seen = len(seen_account_ids) expected_page_size = min(100, max(total - already_seen, 0)) if len(accounts) != expected_page_size: if not accounts and already_seen < total: raise RuntimeError( "Aliyun ListAccounts returned an empty page before " f"TotalCount was exhausted (page={page}, total={total}); " "refusing a partial scope snapshot", ) raise RuntimeError( "Aliyun ListAccounts page cardinality disagrees with " f"TotalCount (page={page}, rows={len(accounts)}, " f"expected_rows={expected_page_size}, total={total}, " f"already_seen={already_seen}); refusing a partial " "or oversized scope snapshot", ) if not accounts: break for acct in accounts: state = str(acct.get("Status", "CreateSuccess")).lower() acct_id = str(acct.get("AccountId") or "").strip() if not acct_id.isdigit() or len(acct_id) < 10: raise RuntimeError( "Aliyun ListAccounts returned a non-canonical " f"AccountId at page={page}; refusing an untrusted " "scope snapshot", ) if acct_id in seen_account_ids: raise RuntimeError( "Aliyun ListAccounts returned a repeated " f"AccountId at page={page}; refusing a duplicated " "scope snapshot", ) seen_account_ids.add(acct_id) # Mark the management/payer account as owner-exempt infra if # ListAccounts returns the STS identity. The provider type is # checked as consistency evidence only; it cannot establish a # second identity. This keeps the backstop append from being # the only place the marker is set without trusting a weaker # signal than STS. directory_claims_master = ( acct.get("Type") == "ResourceDirectoryMaster" ) if directory_claims_master and acct_id != master_id: raise RuntimeError( "Aliyun ListAccounts returned a ResourceDirectoryMaster " "AccountId that disagrees with STS; refusing an " "inconsistent scope snapshot", ) # STS is the identity authority. The directory type is useful # consistency evidence, but cannot independently exempt an # unrelated member scope from owner governance. is_master = acct_id == master_id raw: dict[str, Any] = { "type": acct.get("Type"), "folder_id": acct.get("FolderId"), } if is_master: raw["rd_master"] = True out.append(CloudScopeInfo( scope_type="aliyun_account", external_id=acct_id, display_name=( acct.get("DisplayName") or acct.get("AccountName") ), status="active" if "success" in state else state, raw=raw, )) if len(seen_account_ids) == total: break page += 1 else: raise RuntimeError( "Aliyun ListAccounts exceeded the " f"{_MAX_SCOPE_PAGES}-page safety ceiling; refusing a partial " "scope snapshot", ) # Backstop: Aliyun ``ListAccounts`` enumerates the directory's # MEMBER accounts and does NOT always return the management # account that owns the directory — so without this the master's # own resources + every empty-OwnerID (master-allocated) billing # line have no scope home. Pre-fix those master lines fanned out # across all member scopes and over-reported tenant spend (prod # 2026-06-17). Append the STS-resolved master if it wasn't in the # listing (idempotent dedup on external_id) so # ``_billing_row_belongs_to_scope`` can attribute empty-OwnerID # rows to exactly this scope. When it WAS in the listing it's # already marked rd_master in the loop above. if master_id and not any(s.external_id == master_id for s in out): alias = str(self.config.get("_alias") or self.integration_id) out.append(CloudScopeInfo( scope_type="aliyun_account", external_id=master_id, display_name=f"{alias} (RD master)", status="active", raw={"type": "ResourceDirectoryMaster", "rd_master": True}, )) return out # ── Scope creation: Resource Directory CreateResourceAccount ─ async def create_scope( self, *, name: str, description: str | None, owner_email: str, budget_usd_monthly: float, metadata: dict[str, Any] | None = None, idempotency_key: str | None = None, # noqa: ARG002 ) -> CloudScopeInfo: """Aliyun Resource Directory ``CreateResourceAccount`` — production path for the self-service "申请新阿里云子账号" flow. Requires the integration's master credentials to belong to the Resource Directory **master account** with the ``AliyunResourceDirectoryServiceLinkedRolePolicy`` SLR provisioned. ``rd_parent_folder_id`` is OPTIONAL: when set, the new sub-account is filed under that 资源夹; when blank (the common case — verified against prod 2026-06-03, neither the domestic nor the 海外 RD master had it configured) we resolve the RD **Root** folder id via ``GetResourceDirectory`` and file the member there, exactly like AWS Organizations' default. We resolve Root explicitly rather than relying on Aliyun's undocumented omit-ParentFolderId behaviour, so the call always carries a concrete folder id. The previous behaviour hard-raised on a blank folder id, which permanently blocked the self-service "申请新阿里云子账号" flow even though Aliyun never required the parameter. ``DisplayName`` is the human-readable name (hyphens / spaces / Alibaba brand words all allowed); ``AccountNamePrefix`` is the alphanumeric-only, brand-token-stripped reduction of ``name`` and seeds the auto-generated ``@.aliyunid.com`` login email (Aliyun rejects brand words like ``aliyun`` there — see the prefix-build comment below). The two are independent — admins can rename the account later in the console without breaking Helix's external_id binding (we store the AccountId, not the email). Returns a :class:`CloudScopeInfo` with ``scope_type= "aliyun_account"`` and the new account's AccountId. The scope-request service writes the corresponding ``cmdb_cloud_scopes`` row with the requester as owner + platform-default budget / alert rules pre-applied. Idempotency: ``CreateResourceAccount`` is NOT idempotent on the Aliyun side (no client token in the same request id), but the scope-request state machine guards against double-create — a request can only enter ``provisioning`` once. If the same DisplayName already exists under the parent folder, Aliyun rejects with ``EntityAlreadyExists.Account`` which we surface verbatim so admin can rename and retry. """ # Optional: blank → file the account under the RD **Root** folder. # We do NOT raise here. Rather than betting on Aliyun's # omit-ParentFolderId default behaviour (the API doc marks the # parameter optional but does not document the default folder), # resolve the Root folder id explicitly via GetResourceDirectory # and always send ParentFolderId. ``configured`` records whether # the folder came from operator config or the resolved Root, so # ``raw`` stays honest for later inspection. configured_folder = str( self.config.get("rd_parent_folder_id") or "", ).strip() if configured_folder: parent_folder = configured_folder else: parent_folder = await self._resolve_rd_root_folder_id() account_type = str( self.config.get("rd_account_type") or "ResellAccount", ).strip() or "ResellAccount" if account_type not in {"ResellAccount", "NormalAccount"}: raise RuntimeError( f"rd_account_type must be ResellAccount or NormalAccount, " f"got {account_type!r}", ) # 结算账号. Omitting PayerAccountId makes the new member # SELF-PAYING (console 「新成员自主付款」) — see the helper's # docstring for why that default is wrong for us. payer_account_id, payer_source = ( await self._resolve_rd_payer_account_id() ) # AccountNamePrefix is what Aliyun bakes into the auto-issued # login email (``@.aliyunid.com``). Two live # constraints, both confirmed empirically against the prod # 2020-03-31 endpoint on 2026-06-03 (request # ``dev-offline-afterworld-aliyun``): # # 1. **No Alibaba brand tokens.** A prefix containing # ``aliyun`` is rejected with # ``IllegalParameter.Account.AccountName`` ("the account # name is illegal") — the login email would otherwise read # ``…aliyun@…aliyunid.com``, impersonating the brand. PROVEN: # prefix ``devofflineafterworldaliyun`` was refused; the # identical request with only ``aliyun`` removed # (``devofflineafterworld``) created the account. We strip # the documented Alibaba brand family below. # 2. **Alphanumeric only.** We reduce to lowercase ASCII # alnum (drop spaces / hyphens / dots) — the safest superset # of the documented rule (start/end alnum, no consecutive # specials), so we never have to reason about edge cases in # the live endpoint's stricter-than-documented behaviour. # # ``DisplayName`` is independent — it keeps the readable, hyphen- # and ``aliyun``-bearing name (every existing PingCAP sub-account # has a hyphenated DisplayName, and this very request's DisplayName # ``dev-offline-afterworld-aliyun`` was accepted). Aliyun's prefix # bound is 2–37 chars; we cap at 37 and pad a too-short prefix. slug = "".join( c for c in name.lower() if c.isascii() and c.isalnum() ) # Strip Alibaba brand tokens (substring, case-insensitive — slug is # already lowercased). Order longest-first so e.g. ``alibaba`` is # removed whole rather than leaving fragments. for brand in ("alibaba", "aliyun", "alipay", "hichina"): slug = slug.replace(brand, "") # Aliyun's AccountNamePrefix must START WITH A LOWERCASE LETTER (it # seeds the login-email local-part). Names like ``123`` or # ``01-datalake`` pass the upstream alnum-only validator and the FE # rule but would otherwise yield a digit-leading slug Aliyun rejects # with IllegalParameter.Account.AccountName. Brand-stripping can also # turn a letter-leading name into a digit-leading slug # (``aliyun123`` -> ``123``), so the checks MUST run AFTER stripping. if not slug: # Reduced to nothing (e.g. "Aliyun") — use the readable default. slug = "helix" elif not (slug[0].isascii() and slug[0].isalpha()): # Digit/other-leading — prepend a letter to satisfy Aliyun. slug = "h" + slug slug = slug[:37] if len(slug) < 2: slug = (slug + "helix")[:37] params: dict[str, Any] = { "DisplayName": name, "AccountNamePrefix": slug, "ResellAccountType": account_type, "ParentFolderId": parent_folder, } if payer_account_id: params["PayerAccountId"] = payer_account_id try: data = await self._acs3_call( host="resourcemanager.aliyuncs.com", version="2020-03-31", action="CreateResourceAccount", params=params, ) except RuntimeError as e: # Aliyun's create can take a few seconds; the request always # returns synchronously with the new AccountId though, so # any RuntimeError here is genuine (auth / quota / dup name). raise RuntimeError( f"Aliyun CreateResourceAccount rejected: {e}" ) from e account = data.get("Account") or data account_id = str(account.get("AccountId") or "").strip() if not account_id: raise RuntimeError( f"Aliyun CreateResourceAccount returned no AccountId: " f"{data!r}", ) return CloudScopeInfo( scope_type="aliyun_account", external_id=account_id, display_name=account.get("DisplayName") or name, status="active", raw={ "rd_parent_folder_id": parent_folder, "rd_parent_folder_source": ( "config" if configured_folder else "resolved_root" ), "account_type": account_type, "payer_account_id": payer_account_id, "payer_source": payer_source, "owner_email": owner_email, "description": description, "metadata": metadata or {}, "budget_usd_monthly": budget_usd_monthly, "aliyun_response": account, }, ) async def _resolve_rd_root_folder_id(self) -> str: """Return the Resource Directory **Root** folder id via ``GetResourceDirectory``. Used by :meth:`create_scope` when the integration has no explicit ``rd_parent_folder_id`` configured — new sub-accounts then land in the RD Root folder. We resolve it explicitly (rather than relying on Aliyun's undocumented omit-ParentFolderId default) so the ``CreateResourceAccount`` call always carries a concrete folder id and the failure mode is a clean error here instead of a surprising placement on the Aliyun side. Raises ``RuntimeError`` if the call fails or returns no ``RootFolderId`` — the caller surfaces it as the scope-request ``failure_reason``. A blank RootFolderId almost always means the credentials don't belong to the RD **master** account (a member account can't create sub-accounts at all), so the loud failure is the correct behaviour. """ data = await self._get_resource_directory() rd = data.get("ResourceDirectory") or data root_folder_id = str(rd.get("RootFolderId") or "").strip() if not root_folder_id: raise RuntimeError( "Aliyun GetResourceDirectory returned no RootFolderId — " "the integration's credentials must belong to the Resource " "Directory master account to create sub-accounts. " f"response={data!r}", ) return root_folder_id async def _get_resource_directory(self) -> dict[str, Any]: """``GetResourceDirectory``, cached for the driver instance. A single ``create_scope`` call needs two facts out of it (RD Root folder id + RD 管理账号 id, the latter being the default 结算账号), and the driver instance is short-lived — one HTTP call is enough. """ cached: dict[str, Any] | None = self._rd_directory_cache if cached is not None: return cached data = await self._acs3_call( host="resourcemanager.aliyuncs.com", version="2020-03-31", action="GetResourceDirectory", params={}, ) self._rd_directory_cache = data return data async def _resolve_rd_payer_account_id(self) -> tuple[str | None, str]: """Resolve the 结算账号 for a new Resource Directory member. Returns ``(payer_account_id, source)``; a ``None`` payer means "send no ``PayerAccountId``", i.e. Aliyun's 「新成员自主付款」. **Why this exists** (prod, 2026-08-07): Aliyun's ``CreateResourceAccount`` treats ``PayerAccountId`` as optional and silently defaults to 「新成员自主付款」. Helix never sent the parameter, so every self-service 阿里云子账号 landed self-paying — confirmed on ``data-integration-team-internal-alicloud`` (AccountId 5083883504403318, created 2026-07-08), whose 当前结算账号 read 自主付款 and had to be flipped to 「使用管理账号为新成员付款」 by hand in the console. A self-paying member is not merely a cosmetic setting: it sits outside the master's consolidated bill, so Helix's per-scope cost/billing sync (which reads the master's BSS consolidated bill) under-reports it, and the account can suspend on an unpaid balance nobody owns. Resolution order: 1. ``rd_payer_account_id`` = ``self``/``none``/``own`` → no payer param (deliberate 自主付款; an explicit operator decision). 2. ``rd_payer_account_id`` = a numeric AccountId → that account pays (console 「使用已有成员为新成员付款」). 3. blank (the default) → the RD **管理账号** pays (console 「使用管理账号为新成员付款」), resolved via ``GetResourceDirectory``. Raises ``RuntimeError`` when the master id can't be resolved. Failing the request is deliberate: falling back to "send nothing" would silently recreate exactly the bug above, and the scope-request state machine surfaces the message as ``failure_reason`` so admin can retry after fixing credentials. """ configured = str( self.config.get("rd_payer_account_id") or "", ).strip() if configured.lower() in {"self", "none", "own"}: return None, "config_self" if configured: if not configured.isdigit(): raise RuntimeError( f"rd_payer_account_id must be a numeric Aliyun " f"AccountId (or 'self' for 自主付款), got " f"{configured!r}", ) return configured, "config" data = await self._get_resource_directory() rd = data.get("ResourceDirectory") or data master_id = str(rd.get("MasterAccountId") or "").strip() if not master_id: raise RuntimeError( "Aliyun GetResourceDirectory returned no MasterAccountId, " "so the new sub-account's 结算账号 cannot be set to the " "Resource Directory 管理账号 — refusing to create a " "self-paying member. Check that the integration's " "credentials belong to the RD master account, or set " "rd_payer_account_id explicitly ('self' = 自主付款). " f"response={data!r}", ) return master_id, "resolved_master" # ── Role catalog: RAM policies on the master ───────────── async def list_scope_roles( self, scope: CloudScopeRef, ) -> list[CloudRoleInfo]: """Admin-curated system policies (promoted) + every Custom policy (promoted) + every other System policy (NOT promoted). Aliyun's RAM has ~250 ``System`` policies and a tenant typically has 5-50 ``Custom`` ones. Pre-068, all of them landed in the request dropdown and the employee had to scan ~300 entries to find "ReadOnlyAccess". Now we mirror the GCP/Azure behavior: only curated System policies get promoted to the default view; Custom policies are always promoted (the customer chose to create them); the rest stay in the catalog so admins can promote/demote them later. """ from app.integrations.drivers._cloud_defaults import ( load_default_roles, ) out: list[CloudRoleInfo] = [] promoted_keys: set[str] = set() defaults = await load_default_roles( "cloud.aliyun.default_policies", _ALIYUN_DEFAULT_FALLBACK, ) for entry in defaults: key = str(entry.get("role_key") or "").strip() if not key: continue promoted_keys.add(key) # Hard page ceiling + marker-unchanged guard. A degenerate # ``IsTruncated=True`` response with the same / empty Marker would # otherwise spin forever, pinning a worker. 200 pages × 200 items = # 40k policies, well above any realistic tenant. Mirrors the # ListUsers guard below (list_iam_users). _MAX_POLICY_PAGES = 200 marker: str | None = None for policy_type in ("System", "Custom"): pages_walked = 0 while True: pages_walked += 1 if pages_walked > _MAX_POLICY_PAGES: raise RuntimeError( "Aliyun ListPolicies(PolicyType=" f"{policy_type}): hit {_MAX_POLICY_PAGES}-page ceiling " "— refusing to loop or silently truncate; caller " "surfaces as transient failure (no state overwrite)", ) params = {"PolicyType": policy_type, "MaxItems": 200} if marker: params["Marker"] = marker try: data = await self._acs3_call( host="ram.aliyuncs.com", version="2015-05-01", action="ListPolicies", params=params, ) except RuntimeError as e: # ListPolicies failure is logged but doesn't fail the # whole sync — partial catalog is better than no # catalog at all (the admin-curated defaults loaded # above are still useful). The next ``cmdb.roles_sync`` # cycle will retry. logger.warning( "Aliyun ListPolicies(PolicyType=%s) failed: %s", policy_type, e, ) break for p in (data.get("Policies") or {}).get("Policy") or []: policy_name = p.get("PolicyName") or "" # ── Aliyun CloudSSO mirror filter ────────────── # CloudSSO-enabled accounts auto-materialize every # permission set as a ``Custom`` RAM policy named # ``AliyunReservedSSO--``. Aliyun # marks these reserved — ``AttachPolicyToUser`` # always returns 400 ``NotSupport.AliyunReservedPolicy``. # Surfacing them in the request dropdown is a # footgun (PR #565 — mohangjie/dengshuan landed a # poweruser request that wedged provisioning). # Granting these correctly goes through the # CloudSSO ``CreateAccessAssignment`` path on the # intl driver, NOT RAM. # # Narrow the prefix to ``AliyunReservedSSO-`` (the # specific reserved CloudSSO mirror family Aliyun # documents — Round-1 code review caught the # earlier ``AliyunReserved`` prefix as over-broad: # an admin who legitimately named a Custom policy # ``AliyunReservedHelixOpsLockdown`` or any future # ``AliyunReservedX``-but-attachable family would # vanish from the dropdown for no real reason). if policy_name.startswith("AliyunReservedSSO-"): continue role_key = f"{policy_type}:{policy_name}" is_custom = policy_type == "Custom" out.append(CloudRoleInfo( role_key=role_key, display_name=policy_name, description=p.get("Description"), is_custom=is_custom, source=f"aliyun_{policy_type.lower()}", scope=scope, is_promoted=is_custom or role_key in promoted_keys, )) if not data.get("IsTruncated"): break new_marker = data.get("Marker") if not new_marker or new_marker == marker: # IsTruncated=True but the Marker didn't advance (or came # back empty) — a degenerate response. Raise rather than # loop forever or silently truncate the policy catalog. raise RuntimeError( "Aliyun ListPolicies(PolicyType=" f"{policy_type}): IsTruncated=True with unchanged/empty " "Marker — refusing to loop or truncate; caller " "surfaces as transient failure (no state overwrite)", ) marker = new_marker marker = None return out # ── Grant / revoke: RAM user + policy attach ──────────── def _ram_user_name_for(self, email: str) -> str: local = email.split("@", 1)[0] # RAM user name allowed chars: letters/digits/- /. /_ / up to 64 sanitized = "".join( c for c in local if c.isascii() and (c.isalnum() or c in "-._") ) if sanitized: return sanitized[:64] # Round-20 audit (A1F5): a fully non-ASCII local-part (e.g. # ``张三@x.com``) collapses to empty, then to the literal # ``helix-user`` — two such employees collide on one RAM user. # Mirror the AWS driver's sha1-based fallback so the mapping # stays 1:1 and deterministic across runs. import hashlib # Compatibility identity, not secret protection: this digest is part # of already-provisioned external RAM usernames. Changing it would # orphan revoke lookups and create duplicate cloud identities. digest = hashlib.sha1( email.encode("utf-8"), usedforsecurity=False, ).hexdigest()[:12] return f"helix-{digest}" async def grant_access( self, ctx: CloudAccessContext, ) -> CloudGrantResult: if ":" not in ctx.role_key: return CloudGrantResult( ok=False, error_message=( "Aliyun role_key must be 'System:PolicyName' or " "'Custom:PolicyName'" ), ) policy_type, policy_name = ctx.role_key.split(":", 1) # ── CloudSSO mirror guard ────────────────────────────────── # ``AliyunReservedSSO--`` Custom RAM policies # are CloudSSO permission set mirrors and cannot be attached # to RAM users — Aliyun rejects with # ``NotSupport.AliyunReservedPolicy``. The ``list_scope_roles`` # filter (PR #565) keeps them out of the request dropdown, # but legacy ``cmdb_cloud_roles`` rows and direct API callers # could still land here. Surface a clear error instead of # letting the user see Aliyun's raw 400. # # Narrow prefix ``AliyunReservedSSO-`` (mirrors the catalog # filter); over-broad ``AliyunReserved`` would block any # admin-created Custom policy whose name happens to begin # with that string. if policy_name.startswith("AliyunReservedSSO-"): return CloudGrantResult( ok=False, error_message=( f"Policy {policy_name!r} is reserved by Alibaba " "Cloud (CloudSSO permission set mirror) and cannot " "be attached to RAM users. Use the CloudSSO grant " "path or pick a non-reserved policy." ), ) user_name = self._ram_user_name_for(ctx.principal_email) # Ensure RAM user exists try: await self._acs3_call( host="ram.aliyuncs.com", version="2015-05-01", action="CreateUser", params={ "UserName": user_name, "DisplayName": ctx.principal_display_name or user_name, "Email": ctx.principal_email, "Comments": "managed by helix", }, ) created = True except RuntimeError as e: if "EntityAlreadyExists.User" in str(e): created = False else: return CloudGrantResult(ok=False, error_message=str(e)) # Attach policy try: await self._acs3_call( host="ram.aliyuncs.com", version="2015-05-01", action="AttachPolicyToUser", params={ "PolicyType": policy_type, "PolicyName": policy_name, "UserName": user_name, }, ) except RuntimeError as e: if "EntityAlreadyExists" in str(e): return CloudGrantResult( ok=True, already_exists=True, external_ref=f"{user_name}::{policy_type}:{policy_name}", detail={"user_name": user_name, "created": created}, ) return CloudGrantResult(ok=False, error_message=str(e)) return CloudGrantResult( ok=True, external_ref=f"{user_name}::{policy_type}:{policy_name}", detail={"user_name": user_name, "created": created}, ) async def revoke_access( self, ctx: CloudAccessContext, ) -> CloudGrantResult: # CloudSSO assignments live under a different API surface and # must be revoked by the ``alicloud_intl`` subclass override # (which calls ``DeleteAccessAssignment`` / ``RemoveUserFromGroup``). # If we somehow reach the RAM-based path with an ``sso::`` ref, # the legacy ``split(":", 1)`` would derive a garbage PolicyName # with ``:`` in it and Aliyun rejects with # ``InvalidParameter.PolicyName.InvalidChars`` — surface a clear # routing error instead. if (ctx.existing_external_ref or "").startswith("sso::"): return CloudGrantResult( ok=False, error_message=( "Aliyun RAM revoke received a CloudSSO external_ref " f"({ctx.existing_external_ref!r}). The intl/SSO subclass " "should be handling this — check the driver registration " "for this integration." ), ) if ctx.existing_external_ref and "::" in ctx.existing_external_ref: user_name, role_part = ctx.existing_external_ref.split("::", 1) else: user_name = self._ram_user_name_for(ctx.principal_email) role_part = ctx.role_key if ":" not in role_part: return CloudGrantResult( ok=False, error_message="Invalid role_key for Aliyun revoke", ) policy_type, policy_name = role_part.split(":", 1) try: await self._acs3_call( host="ram.aliyuncs.com", version="2015-05-01", action="DetachPolicyFromUser", params={ "PolicyType": policy_type, "PolicyName": policy_name, "UserName": user_name, }, ) except RuntimeError as e: if "EntityNotExist" in str(e): return CloudGrantResult(ok=True, already_absent=True) return CloudGrantResult(ok=False, error_message=str(e)) return CloudGrantResult(ok=True, external_ref=ctx.existing_external_ref) async def get_last_login( self, ctx: CloudAccessContext, ) -> datetime | None: user_name = self._ram_user_name_for(ctx.principal_email) try: data = await self._acs3_call( host="ram.aliyuncs.com", version="2015-05-01", action="GetUser", params={"UserName": user_name}, ) except RuntimeError: return None user = data.get("User") or {} last = user.get("LastLoginDate") if not last: return None try: return datetime.fromisoformat(str(last).replace("Z", "+00:00")) except ValueError: return None # ── Daily billing: BSS DescribeInstanceBill aggregation ────── # # The BSS API host is region-specific. ``business.aliyuncs.com`` # is the CN endpoint and rejects intl-tenant credentials with # ``NotApplicable: caller site doesn't match API region``. The # international tenants must use # ``business.ap-southeast-1.aliyuncs.com``. # ``AlicloudIntlCloudDriver`` overrides this class-level constant. _BSS_HOST = "business.aliyuncs.com" 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]: """Aggregate per-(scope, day) total cost via BSS DescribeInstanceBill. Round-34 (2026-05-18) switched from ``QueryAccountBill`` to ``DescribeInstanceBill`` after Round-33's ``BillOwnerId`` patch was found insufficient on a Resource Directory **Reseller** master (the default Aliyun ``rd_account_type``). In Reseller mode the master pays for every sub-account's resources, so a sub-account's own ``QueryAccountBill`` row only contains cross- account residual fees (Data Transfer / K8s mgmt) — every ECS / EBS / OSS / SLB line item is INVISIBLE because master paid it. Prod evidence (alicloud-intl-primary, 2026-05-18): post-Round-33 ``QueryAccountBill BillOwnerId=`` returned $7/day for a sub whose true spend (per DescribeInstanceBill) was $4500/day. ``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). Aggregating per-day per-(filtered scope) produces ``CloudDailyCost`` rows that are 1:1 with ``SUM(cost_facts.amount_usd) GROUP BY date, scope`` by construction (same upstream API feed, same per-row conversion via ``_convert_pretax_amount``, only the output grain differs). Works correctly for all account modes: * Single-account IAM: ``OwnerID == caller``, client filter matches. * RD master + NormalAccount sub: ``OwnerID == sub``, payer == sub. * RD master + ResellAccount sub: ``OwnerID == sub``, payer == master. ``QueryAccountBill`` is preserved only for the alternative monthly- aggregator path (``AlicloudIntlProvider.get_monthly_billing`` calls it with ``IsGroupByProduct=false`` to surface the cross-RD total — that path is OK because it deliberately asks for the master-level aggregate, not per-sub). Aliyun BSS's ``BillingDate`` field is Asia/Shanghai-anchored (verified against console rendering). Helix's ``_month_bounds`` in scope_billing_detail uses UTC. The mismatch means a row dated 2026-05-01 (Beijing) covers Beijing 2026-05-01 00:00 → 2026-05-02 00:00 (≈ UTC 2026-04-30 16:00 → 2026-05-01 16:00). Cross-provider monthly views are accurate to ±8 hours at the month boundary. Documented convention; pixel-precise alignment requires the hourly-grain BSS API + re-bucket (Phase-2 surgery). """ del bq_billing # AWS-driver-specific kwarg; accept and ignore if scope is None: return [] # Parent emits CNY RAW (no FX conversion in this path — the # downstream ``_to_usd`` in lifecycle_service.py reads the # ``currency`` column and converts on read via the platform's # ``resolve_cny_per_usd``). Intl override below handles per-row # currency conversion to USD via ``_convert_pretax_amount``. owner_filter = scope.external_id # Mirror list_cost_facts: empty-OwnerID (master-allocated) rows # attribute to the master scope only, so billing_daily and # cost_facts stay reconciled and neither fans out. master_id = await self._master_account_id() out: list[CloudDailyCost] = [] cur = start_date rows_missing_gross = 0 while cur <= end_date: by_service: dict[str, float] = {} total_cny = 0.0 list_cny = 0.0 list_seen = False next_token: str | None = None pages_walked = 0 while True: pages_walked += 1 if pages_walked > _MAX_BILL_PAGES: raise RuntimeError( f"Aliyun DescribeInstanceBill {cur} hit the " f"{_MAX_BILL_PAGES}-page ceiling — refusing to " "loop; surfacing as a transient failure rather " "than hanging the worker.", ) 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``. "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"Aliyun DescribeInstanceBill {cur} failed: {e}", ) from e raw_data = data.get("Data") or {} # ``DescribeInstanceBill`` returns ``Items`` as a plain # list (unlike ``QueryAccountBill`` which wraps in # ``{"Item": [...]}``). Mirrors ``list_cost_facts``. items = raw_data.get("Items") or [] for item in items: if not self._billing_row_belongs_to_scope( str(item.get("OwnerID") or ""), owner_filter, master_id, ): continue native = float(item.get("PretaxAmount") or 0) svc = str( item.get("ProductName") or item.get("ProductCode") or "Other", ) # 双口径:仅正常用量行计入 list(牌价);调账/退款 # 行不计;缺 PretaxGrossAmount 的行计数告警。 # # #2223: the list accumulation runs BEFORE the # zero-invoice skip. A usage row fully covered by a # voucher / RI / spending-threshold discount has # ``PretaxAmount == 0`` but ``PretaxGrossAmount > 0`` # — the same "invoice zero, list non-zero" shape #2175 # fixed on the AWS BQ path and #2180 on the principal-tag # path, in the same PR. Skipping first silently dropped # those rows from the engineering (list) measure, which # is exactly the measure that must NOT move when a # discount lands. if self._classify_bss_bucket(item, native) == COST_BUCKET_USAGE: gross = self._bss_gross_amount(item) if gross is None: if native != 0: rows_missing_gross += 1 else: list_cny += gross list_seen = True if native == 0: continue by_service[svc] = by_service.get(svc, 0.0) + native total_cny += native prev_token = next_token next_token = raw_data.get("NextToken") or None if not next_token: break if next_token == prev_token: raise RuntimeError( f"Aliyun DescribeInstanceBill {cur} returned a " "non-advancing NextToken — refusing to loop.", ) out.append(CloudDailyCost( date=cur, scope=scope, # CNY raw. Legacy field name ``total_cost_usd`` is a # misnomer — downstream ``_to_usd`` in lifecycle_service # checks ``currency`` and divides by cny_per_usd when # CNY. Keeping the raw value here avoids silently # converting at FX rates that drift from reality. total_cost_usd=total_cny, # 同为 CNY raw(与 total_cost_usd 同一 currency 语义); # 阿里的牌价即 PretaxGrossAmount → public == list。 # 全日无可计行 → None(口径未接入),不伪造 0。 total_cost_usd_list=(list_cny if list_seen else None), total_cost_usd_public=(list_cny if list_seen else None), currency="CNY", cost_by_service=by_service, )) cur += timedelta(days=1) if rows_missing_gross: logger.warning( "Aliyun billing_daily: %d rows missing 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 # ── Line-item cost facts (production-grade FinOps ingestion) ──── # # Aliyun's BSS DescribeInstanceBill is the line-item endpoint: # one row per (BillingDate × InstanceID × billing-item) with rich # dimensions (Region, ResourceGroup, ProductCode, Tag map). It's # the canonical FinOps source for Aliyun and what the Resource # Manager / Cost Center console reads from. # # The query is BillingCycle-scoped (YYYY-MM); BillingDate narrows # to a single day. We loop day-by-day across the requested window, # paginate via NextToken, and convert PretaxAmount (CNY) to USD # via the FX rate the cost-ingestion service resolves through # :func:`app.core.fx_rate.resolve_cny_per_usd` (single source of # truth = ``platform_settings.ai_spending.cny_to_usd_rate``; # per-integration ``config_public.cny_per_usd`` is an optional # ops-pinned override). The driver receives the rate as a kwarg — # it does NOT read FX from DB itself, since this method runs in # the async event loop and the sync billing API calls happen in # ``_acs3_call`` via httpx. When FX is unavailable the service # raises before invoking the driver and the cron records a per- # integration ``failed`` status; the driver self-defends with # the ``cny_per_usd is None or <= 0`` guard for direct callers # that bypass the service layer. def _convert_pretax_amount( self, item: dict[str, Any], native: float, fx_rate: float, ) -> tuple[float, str] | None: """Convert a single BSS line-item ``PretaxAmount`` to USD. Returns ``(amount_usd, native_currency_label)`` or ``None`` to drop the row (unknown / unsupported currency). The 国内阿里云 (CN-primary) tenant bills exclusively in CNY, so the parent implementation hardcodes the conversion. The 海外 (international / SG) subclass overrides this hook to honour the per-row ``Currency`` field returned by the SG BSS endpoint — intl tenants bill in USD by default, and dividing a USD ``PretaxAmount`` by ``cny_per_usd`` (the parent's behaviour) would silently undercount their spend by ~7×, breaking finance dashboards and budget alerts. Round-18 audit fix. """ return native / fx_rate, "CNY" # ── 业务折前成本 (list cost) — BSS 行分类 ────────────────────── # # 研发口径 = ``PretaxGrossAmount``(税前原价/牌价,优惠前应付) — # 白名单同 AWS/GCP 思路:只计正常用量行;退款/调账(负数,2026 YTD # −$21.5 万,集中在 5/25、7/21 处理日)归 adjustment 桶、list 恒 # None。``Item`` 字段(SubscriptionOrder / PayAsYouGoBill / Refund / # Adjustment)是主分类信号;缺失时回退按金额符号(负数 → adjustment, # 保守)。两个待云管理确认的口径点(抢占式实例在 PretaxGrossAmount # 里按 spot 实价还是标准价记账;PretaxGrossAmount 是否为纯牌价、无 # 价内合同折扣)——答案回来只需调整此处,白名单结构不变。 # ``PretaxGrossAmount`` 缺失的行不得用实付兜底 —— list 落 None, # 调用方聚合缺数计数、超 0 记 WARN(云管理团队的规则:缺失行应 # 单独告警核查)。 _BSS_USAGE_ITEMS: frozenset[str] = frozenset({ "SubscriptionOrder", "PayAsYouGoBill", }) _BSS_ADJUSTMENT_ITEMS: frozenset[str] = frozenset({ "Refund", "Adjustment", }) @classmethod def _classify_bss_bucket(cls, item: dict[str, Any], native: float) -> str: item_type = str(item.get("Item") or "").strip() if item_type in cls._BSS_USAGE_ITEMS: return COST_BUCKET_USAGE if item_type in cls._BSS_ADJUSTMENT_ITEMS: return COST_BUCKET_ADJUSTMENT if item_type: # 未知新类型:白名单思路,进 other,绝不误入 usage。 return COST_BUCKET_OTHER # 老响应缺 Item 字段:按符号保守分类(负数=调账形状)。 return COST_BUCKET_USAGE if native >= 0 else COST_BUCKET_ADJUSTMENT @staticmethod def _bss_gross_amount(item: dict[str, Any]) -> float | None: """``PretaxGrossAmount``（牌价）；缺失/不可解析 → None（绝不 用实付兜底——None 由调用方计数告警）。""" raw = item.get("PretaxGrossAmount") if raw in (None, ""): return None try: return float(raw) except (TypeError, ValueError): return None async def list_cost_facts( self, start_date: date, end_date: date, *, scope: CloudScopeRef | None = None, cny_per_usd: float | None = None, principal_tag_keys: list[str] | None = None, bq_billing: dict[str, Any] | None = None, ) -> list[CostFact]: # ``principal_tag_keys`` is AWS-specific (IAM principal tag # cost-allocation) — Aliyun BSS has no equivalent dimension, # so we accept the kwarg for signature uniformity and ignore. # ``bq_billing`` likewise AWS-driver-specific (CUR-in-BigQuery). del principal_tag_keys, bq_billing # Scope is optional. Aliyun's DescribeInstanceBill returns # rows tagged with OwnerID (member account) so we can filter # client-side; passing scope.external_id lets us short-circuit # by tagging only matching rows. None = full Resource # Directory aggregate. owner_filter = scope.external_id if scope is not None else None # FX rate guard — behaviour depends on ``self.requires_cny_fx``: # # * True (CN-primary, i.e. AliyunCloudDriver itself): # The cost-ingestion service MUST resolve the rate before # invoking this method. None or ≤0 means the service failed # to resolve → raise loudly so the cron records a hard # failure instead of silently writing wrong USD aggregates. # # * False (intl tenant, AlicloudIntlCloudDriver): # BSS defaults to USD; CNY appears only on rare legacy lines. # The service skips FX resolution (``requires_cny_fx=False``), # so ``cny_per_usd`` arrives as None. Normalise to 0.0 — # ``_convert_pretax_amount`` already treats fx_rate ≤ 0 as # "drop CNY row with WARNING", which is the correct behaviour # for an intl tenant where CNY is incidental. if cny_per_usd is None or cny_per_usd <= 0: if self.requires_cny_fx: raise RuntimeError( f"Aliyun integration {self.integration_id}: " "``cny_per_usd`` not provided. The cost-ingestion " "service must resolve the rate via " "app.core.fx_rate.resolve_cny_per_usd before calling " "list_cost_facts.", ) # Intl: CNY rows will be dropped by _convert_pretax_amount. cny_per_usd = 0.0 fx_rate = float(cny_per_usd) # Resolve the RD master once per call so empty-OwnerID # (master-allocated) rows are attributed to exactly the master # scope instead of fanning out across every sub-account scope. master_id = await self._master_account_id() out: list[CostFact] = [] integration_id = self.integration_id cur = start_date rows_missing_gross = 0 while cur <= end_date: next_token: str | None = None # #2222: the page ceiling belongs on THIS walk too. The # same-token guard alone is defeated by a BSS response that # alternates tokens (A→B→A→B) or mints a fresh bogus token per # page — each call succeeds fast, so no httpx timeout fires and # the worker watchdog (whose own comment admits it cannot see a # hung loop) never trips; the job just pins a scheduled-jobs # concurrency slot for ever. The intl subclass inherits this # method, so prod's intl cost-facts path was exposed as well. pages_walked = 0 while True: pages_walked += 1 if pages_walked > _MAX_BILL_PAGES: raise RuntimeError( f"Aliyun DescribeInstanceBill (cost_facts) {cur} hit " f"the {_MAX_BILL_PAGES}-page ceiling — refusing to " "loop; surfacing as a transient failure rather than " "hanging the worker.", ) params: dict[str, Any] = { "BillingCycle": cur.strftime("%Y-%m"), "BillingDate": cur.isoformat(), "Granularity": "DAILY", # IsHideZeroCharge=true drops rows where Aliyun has # PRE-NETTED a paired charge + compensating credit # to PretaxAmount=0 (RI cancellations, CSR true-ups). # Finance can no longer reconcile those events. # Keep them: the per-row ``if native == 0: continue`` # filter below still drops pure metering-noise zero # rows that have no compensating pair. "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"Aliyun DescribeInstanceBill {cur} failed: {e}", ) from e raw_data = data.get("Data") or {} items = raw_data.get("Items") or [] for item in items: if not self._billing_row_belongs_to_scope( str(item.get("OwnerID") or ""), owner_filter, master_id, ): continue native = float(item.get("PretaxAmount") or 0) # #2223: a usage row whose invoice amount is zero can # still carry a non-zero LIST price (voucher / RI / # full discount). Dropping it here lost the whole row # — both measures — from cost_facts. Emit it with # ``amount_usd = 0`` and the real list price; a row # with NEITHER measure is still skipped below. converted = self._convert_pretax_amount( item, native, fx_rate, ) if converted is None: continue amount_usd, currency = converted tag_map = _parse_aliyun_tags( item.get("Tag"), instance_id=str(item.get("InstanceID") or ""), ) # 双口径:牌价(PretaxGrossAmount)→ list/public, # 与实付走同一 FX 换算钩子(intl 子类按行币种, # CN 父类 CNY/fx_rate),保证可比。调账/退款行 # list 恒 None;usage 行缺牌价 → None + 计数告警。 bucket = self._classify_bss_bucket(item, native) list_native: float | None = None list_usd: float | None = None if bucket == COST_BUCKET_USAGE: gross = self._bss_gross_amount(item) if gross is None: # A zero-invoice row with no gross column # carries no measure at all — that IS the # metering-noise row the old skip targeted, # and it is not worth an "undercount" warning. if native != 0: rows_missing_gross += 1 else: list_native = gross converted_gross = self._convert_pretax_amount( item, gross, fx_rate, ) if converted_gross is not None: list_usd = converted_gross[0] if native == 0 and not list_native: # No invoice money AND no list money — pure # metering noise (or a pre-netted pair with no # gross column). Nothing to record. continue out.append(CostFact( date=cur, integration_id=integration_id, service=str( item.get("ProductName") or item.get("ProductCode") or "Other", ), amount_usd=amount_usd, amount_native=native, amount_usd_list=list_usd, amount_native_list=list_native, # 阿里的牌价即 PretaxGrossAmount → public == list。 amount_usd_public=list_usd, cost_bucket=bucket, currency=currency, scope=scope, resource_group=str( item.get("ResourceGroup") or "", ) or None, resource_id=str(item.get("InstanceID") or "") or None, resource_name=str( item.get("NickName") or item.get("InstanceID") or "", ) or None, resource_type=str( item.get("ProductDetail") or item.get("ProductType") or "", ) or None, region=str(item.get("Region") or "") or None, usage_type=str( item.get("BillingItem") or item.get("BillingType") or "", ) or None, tags=tag_map, usage_quantity=_safe_float(item.get("Usage")), usage_unit=str(item.get("UsageUnit") or "") or None, raw=item, )) prev_token = next_token next_token = raw_data.get("NextToken") or None if not next_token: break if next_token == prev_token: raise RuntimeError( f"Aliyun DescribeInstanceBill {cur} returned a " "non-advancing NextToken — refusing to loop.", ) cur += timedelta(days=1) if rows_missing_gross: logger.warning( "Aliyun cost_facts: %d usage rows missing " "PretaxGrossAmount (scope=%s %s..%s) — list-cost measure " "undercounts; NEVER backfilled from the post-discount " "amount per the whitelist rule. Investigate the BSS " "response shape.", rows_missing_gross, scope.external_id if scope else "", start_date, end_date, ) return out # ── IAM user inventory (tenant-wide) ────────────────────────── async def list_iam_users(self) -> list[dict[str, Any]]: """Enumerate every RAM user on the master account. Used for orphan detection (RAM users with no matching Helix identity). Not a scope-bound method — Aliyun RAM users live on the Resource Directory master, not on member accounts. """ out: list[dict[str, Any]] = [] marker: str | None = None # Round-49: hard cap iteration. A malformed Aliyun response # that always returns ``IsTruncated=True`` with the same # Marker would otherwise loop indefinitely under the # throttle, pinning a worker thread. 100 pages × 100 items # = 10k RAM users, comfortably above any realistic tenant. pages_walked = 0 while pages_walked < 100: pages_walked += 1 params: dict[str, Any] = {"MaxItems": 100} if marker: params["Marker"] = marker # Pre-launch P0 fix (mass-revoke safety): a transient # RAM 5xx / throttle / auth flap MUST raise, never silently # `break` with an empty/partial list — `list_scope_grants` # treats `users == []` as "nothing exists in cloud" and the # grants_sync reconciliation revokes every Helix-tracked # grant. Mirrors the 8th-audit Azure fix (see # azure_cloud.py:1007). The caller (grants_sync / # CloudLifecycleService) catches and surfaces the error # without overwriting state. data = await self._acs3_call( host="ram.aliyuncs.com", version="2015-05-01", action="ListUsers", params=params, ) users = (data.get("Users") or {}).get("User") or [] for u in users: out.append({ "user_id": str(u.get("UserId") or ""), "user_name": str(u.get("UserName") or ""), "display_name": str(u.get("DisplayName") or ""), "email": str(u.get("Email") or ""), "status": "active", "created_at": u.get("CreateDate"), "raw": u, }) if not data.get("IsTruncated"): break new_marker = data.get("Marker") if new_marker == marker: # Aliyun returned IsTruncated=True with an unchanged # Marker — sign of a degenerate response. Raise like # other 5xx-equivalent failures rather than silently # truncating the list (which would feed a mass-revoke). raise RuntimeError( "Aliyun ListUsers: IsTruncated=True with " "unchanged Marker — refusing to loop or " "truncate; caller will surface as transient " "failure (no state overwrite)" ) marker = new_marker else: raise RuntimeError( "Aliyun ListUsers: hit 100-page ceiling — refusing " "to silently truncate. Caller will treat this as " "a transient failure." ) return out # ── Reverse grants discovery (cloud → Helix) ───────────────── async def list_scope_grants( self, scope: CloudScopeRef, ) -> list[DiscoveredGrant]: """Enumerate every RAM user × policy pair tenant-wide. Aliyun RAM is tenant-wide: a user's attached policies apply across every account in the Resource Directory. The grants- sync service calls this once per integration (see ``grants_scope_wide=False``), passing whichever scope it chose as the canonical attachment target. We don't filter on scope — the tenant-wide list is what's accurate.""" users = await self.list_iam_users() if not users: return [] out: list[DiscoveredGrant] = [] for u in users: user_name = str(u.get("user_name") or "") email = str(u.get("email") or "").strip().lower() if not email and "@" in user_name: email = user_name.lower() if not email or not user_name: # No email attribute on the RAM user and username isn't # an email — the service can't resolve to a Helix # person, so don't emit a row (prevents orphan rows). continue # Aliyun RAM ``ListPoliciesForUser`` is intentionally not # paginated here: the official 2015-05-01 API exposes only # ``UserName`` as the input and returns the policies array # directly with no ``Marker`` / ``IsTruncated`` markers. # RAM caps managed-policy attachments at 10 per user # (Aliyun quota ``ram-user-managed-policy``), so the # response always fits one page. Pagination has nothing to # turn — see the UCloud equivalent for the truly-paginated # shape. try: data = await self._acs3_call( host="ram.aliyuncs.com", version="2015-05-01", action="ListPoliciesForUser", params={"UserName": user_name}, ) except RuntimeError as exc: if "EntityNotExist" in str(exc): continue # Transient/unexpected error — raise so the caller records # the scope as errored and skips drift-sweep for it. # Swallowing here would leave the user with no policies in # this scan's result; the next drift-sweep would then revoke # their grants as "stale-discovered", silently removing # legitimate access on a transient Aliyun API failure. raise for p in (data.get("Policies") or {}).get("Policy") or []: policy_name = str(p.get("PolicyName") or "") if not policy_name: continue policy_type = str(p.get("PolicyType") or "Custom") # ``role_key`` shape aligns with existing grant flows: # ":" so the revoke side can # split it back into the two fields Aliyun needs. role_key = f"{policy_type}:{policy_name}" out.append(DiscoveredGrant( principal_email=email, principal_type="user", role_key=role_key, role_display_name=policy_name, external_ref=f"{user_name}::{policy_type}:{policy_name}", scope=scope, granted_at=None, raw={ "user_name": user_name, "policy_type": policy_type, "policy_name": policy_name, "attach_date": p.get("AttachDate"), }, )) return out # ── Account revocation (delete RAM user, kill switch) ───────── async def revoke_account( self, ctx: CloudAccountRevokeContext, ) -> ProvisioningResult: """Delete a RAM user on the master account for the given email. Deletion order (Aliyun RAM requires child resources purged before the user is removable): 1. Detach every policy from the user 2. Delete every AK/SK 3. Delete login profile (console password) 4. Delete MFA device (if any) 5. Remove user from every group 6. ``DeleteUser`` Each step swallows ``EntityNotExist`` so a re-run on a partially-cleaned user still succeeds (idempotent). ``account_id`` should be the RAM user name. For convenience, if an email is passed, we derive the user name using the same rule as :meth:`_ram_user_name_for`. """ raw_id = (ctx.account_id or "").strip() if not raw_id: return ProvisioningResult( ok=False, error_message="Aliyun revoke_account requires account_id (RAM user name or email)", ) # Literal handle first — see ``account_id_candidates``. Deriving # the local part unconditionally makes the chain a silent no-op # whenever the RAM user is genuinely named after an address. candidates = account_id_candidates(raw_id, self._ram_user_name_for) user_name = "" policy_data: dict[str, Any] = {} probe_error: str | None = None for candidate in candidates: try: policy_data = await self._acs3_call( host="ram.aliyuncs.com", version="2015-05-01", action="ListPoliciesForUser", params={"UserName": candidate}, ) user_name = candidate break except RuntimeError as exc: if "EntityNotExist" in str(exc): continue probe_error = str(exc) break if not user_name: if probe_error is not None: return ProvisioningResult(ok=False, error_message=probe_error) return ProvisioningResult( ok=True, external_ref=raw_id, raw_response={ "note": "user already absent", "tried": candidates, }, ) steps: list[str] = [] async def _swallow_nse( label: str, action: str, params: dict[str, Any], ) -> bool: try: await self._acs3_call( host="ram.aliyuncs.com", version="2015-05-01", action=action, params=params, ) steps.append(label) return True except RuntimeError as exc: msg = str(exc) if "EntityNotExist" in msg or ".NotFound" in msg: return False raise # 1. Policies — already fetched by the resolution probe above, # so the user is known to exist and this costs no second call. data = policy_data for p in (data.get("Policies") or {}).get("Policy") or []: await _swallow_nse( f"detach_policy:{p.get('PolicyName')}", "DetachPolicyFromUser", { "PolicyType": p.get("PolicyType") or "Custom", "PolicyName": p["PolicyName"], "UserName": user_name, }, ) # 2. Access keys try: data = await self._acs3_call( host="ram.aliyuncs.com", version="2015-05-01", action="ListAccessKeys", params={"UserName": user_name}, ) for k in (data.get("AccessKeys") or {}).get("AccessKey") or []: await _swallow_nse( f"delete_ak:{k.get('AccessKeyId')}", "DeleteAccessKey", { "UserAccessKeyId": k["AccessKeyId"], "UserName": user_name, }, ) except RuntimeError as exc: # list failed → likely a permission issue; keep going with # the remaining steps, surface error at the end if user # still can't be deleted. logger.warning("Aliyun ListAccessKeys(%s): %s", user_name, exc) # 3. Console login await _swallow_nse( "delete_login_profile", "DeleteLoginProfile", {"UserName": user_name}, ) # 4. MFA await _swallow_nse( "unbind_mfa", "UnbindMFADevice", {"UserName": user_name}, ) # 5. Group memberships try: data = await self._acs3_call( host="ram.aliyuncs.com", version="2015-05-01", action="ListGroupsForUser", params={"UserName": user_name}, ) for g in (data.get("Groups") or {}).get("Group") or []: await _swallow_nse( f"remove_from_group:{g.get('GroupName')}", "RemoveUserFromGroup", {"UserName": user_name, "GroupName": g["GroupName"]}, ) except RuntimeError as exc: logger.warning("Aliyun ListGroupsForUser(%s): %s", user_name, exc) # 6. Delete user try: await self._acs3_call( host="ram.aliyuncs.com", version="2015-05-01", action="DeleteUser", params={"UserName": user_name}, ) steps.append("delete_user") except RuntimeError as exc: if "EntityNotExist" in str(exc): return ProvisioningResult( ok=True, external_ref=user_name, raw_response={ "note": "user already absent", "steps": steps, }, ) return ProvisioningResult( ok=False, error_message=f"Aliyun DeleteUser failed: {exc}", raw_response={"steps": steps}, ) return ProvisioningResult( ok=True, external_ref=user_name, raw_response={"steps": steps}, ) # ── Scope-level closure (admin "下线") ───────────────────── async def close_scope_on_provider( self, scope: CloudScopeRef, *, iam_account_ids: list[str] | None = None, ) -> CloseScopeResult: """Real cloud-side scope closure. Dispatches by ``scope_type``: * ``aliyun_account`` → Resource Directory close path (:meth:`_close_rd_account`). ``GetAccount`` reads ``Type``; a ``CloudAccount`` is first switched back to a ``ResourceAccount`` (``UpdateAccount``), then every member goes through the mandatory ``CheckAccountDelete`` pre-check and a real ``DeleteAccount``. Resumable: the pre-check takes 3–5 minutes on Aliyun's side, so a run that catches it mid-flight returns ``deletion_precheck_running`` with Helix state untouched and the next click continues where it left off. * ``standalone_iam`` → evacuate every Helix-tracked RAM user on this scope by running the same multi-step revoke chain that offboarding uses. Real ``DeleteUser`` per RAM user. Idempotent: an already-removed / not-found account returns ``already_closed=True``. Anything else raises ``NotImplementedError``. """ if scope.scope_type == "aliyun_account": return await self._close_rd_account(scope.external_id) if scope.scope_type == "standalone_iam": return await self._evacuate_ram_users(iam_account_ids or []) raise NotImplementedError( f"Aliyun driver cannot close scope_type={scope.scope_type!r}", ) async def _close_rd_account(self, account_id: str) -> CloseScopeResult: """Really close one Resource Directory member account. Aliyun only deletes members of type ``ResourceAccount`` via ``DeleteAccount``; a ``CloudAccount`` is refused outright with ``MemberTypeError.DeleteAccount``. Until 2026-09-01 this driver gave up at that fork and called ``RemoveCloudAccount`` instead, which merely *unlinks* the member from the directory — the account stays alive, stays billable, and drops out of every Helix sync, while Helix flipped the scope row to ``closed`` and cascade-disabled every grant. A button captioned 下线 that produces a live unmanaged account is the one outcome an irreversible admin action must never have. There is a real path, and this is it: 1. ``GetAccount`` — read ``Type`` / ``Status``. 2. ``CloudAccount`` → ``UpdateAccount`` with ``NewAccountType=ResourceAccount`` (「云账号切换为资源账号」). Aliyun permits the switch when the member's legal entity matches the management account's, the member has security info (安全手机/邮箱) set, and it has no *enabled* root AccessKey. The switch disables the member's root user, which is a step toward closure, not away from it. 3. ``CheckAccountDelete`` + ``GetAccountDeletionCheckResult`` — the pre-check ``DeleteAccount`` requires. Aliyun documents it as taking **3–5 minutes**, so we kick it off, poll briefly, and return ``deletion_precheck_running`` rather than holding the admin's request open for minutes. 4. ``DeleteAccount`` — the real thing, **abandoning every abandonable pre-check item**. Withholding those ids (the original 2026-09-01 shape) makes the closure unfinishable: Aliyun accepts the call and then kills it at ``CheckFailed`` on exactly the items it offered to let you pass. See the comment at the call site. ``DeletionType`` ``"0"`` deletes directly; ``"1"`` parks the member in Aliyun's silence period first and deletes when it expires. A previous attempt that died (``CheckFailed`` / ``DeleteFailed``) is detected up front from the deletion task itself — ``GetAccount`` cannot tell you, a member with a dead deletion looks exactly like a live one — and forces a fresh pre-check, because re-using the dead one's result just refails. **The whole method is resumable and idempotent**, which is what makes the 3–5 minute pre-check survivable without new infrastructure: every step re-derives its own precondition from live provider state, so clicking 下线 again simply continues. Step 2 self-skips once ``Type`` is ``ResourceAccount``; step 3 reads the existing check task instead of restarting it (a restart would re-arm the 3–5 minute wait on every click). ``RemoveCloudAccount`` is deliberately NOT reachable from here. Unlinking a member is a legitimate operation, but it is not a closure, so it must not hide behind this button. When a member genuinely cannot be closed (an invited third-party account whose legal entity differs from the master's), we fail loudly and say what the admin has to do instead. """ try: describe = await self._acs3_call( host=_RD_HOST, version=_RD_VERSION, action="GetAccount", params={"AccountId": account_id}, ) except RuntimeError as exc: msg = str(exc) if any(p in msg for p in _RD_NOT_FOUND_CODES): return CloseScopeResult( ok=True, already_closed=True, provider_state="not_in_directory", detail={"reason": "Account not in this Resource Directory"}, ) if "ResourceDirectoryNotInUse" in msg or "Forbidden.RAM" in msg: return CloseScopeResult( ok=False, error_code="rd_unavailable", error_message=( "Aliyun Resource Directory is not configured on this " "tenant or the integration credentials are not the RD " "master. Configure RD master credentials with " "ResourceDirectoryFullAccess to enable cloud-side close." ), ) return CloseScopeResult( ok=False, error_code="describe_failed", error_message=f"Aliyun GetAccount: {exc}", ) account = describe.get("Account") or {} acct_type = str(account.get("Type") or "").strip() existing_status = str(account.get("Status") or "").strip() # A previous deletion may already be in flight, finished, or dead. # ``GetAccount.Status`` does NOT say — a member whose deletion # check failed still reports ``SwitchSuccess`` / ``CreateSuccess`` # like any live member, which is exactly how prod account # 5995487234252122 looked while its deletion sat at # ``CheckFailed``. Only the deletion task knows, so ask it first. prior = await self._rd_deletion_task_status(account_id) if prior == "Success": return CloseScopeResult( ok=True, already_closed=True, provider_state="deleted", detail={"reason": "Aliyun already deleted this member"}, ) if prior in {"Checking", "Deleting"}: return CloseScopeResult( ok=False, error_code="deletion_precheck_running", error_message=( "A deletion is already running on this member " f"(Aliyun status {prior}). Nothing further was " "submitted; Helix re-reads the provider daily and " "will report the outcome." ), detail={"aliyun_deletion_status": prior}, ) # ``CheckFailed`` / ``DeleteFailed`` — the previous attempt died. # Its check result is stale and re-using it would fail the same # way, so force a fresh ``CheckAccountDelete`` below. retry_after_failure = prior in {"CheckFailed", "DeleteFailed"} if existing_status in _RD_GONE_STATUSES: return CloseScopeResult( ok=True, already_closed=True, provider_state=existing_status.lower(), detail={"prior_status": existing_status, "type": acct_type}, ) # ── Step 2: CloudAccount → ResourceAccount ────────────── converted = False if acct_type == "CloudAccount": failure = await self._switch_cloud_account_to_resource_account( account_id, account, ) if failure is not None: return failure converted = True # Confirm the switch actually landed before pre-checking. # Aliyun runs member type changes through a verifying state # (the ``Switch*`` / ``Promote*`` status family), so a 200 on # UpdateAccount does not by itself mean ``Type`` has flipped # yet. Pre-checking too early gets ``MemberTypeError. # DeleteAccount``, which reads as a hard failure when the # truth is "wait a moment". Everything here is resumable, so # the honest answer is to say so and let the next click # continue — by which point the switch is a no-op skip. settled = await self._rd_account_type(account_id) if settled is not None and settled != "ResourceAccount": return CloseScopeResult( ok=False, error_code="cloud_account_switch_pending", error_message=( "Aliyun accepted the switch from CloudAccount to " "ResourceAccount but has not finished applying it " f"yet (member type still {settled!r}). The member's " "root login is already disabled. Nothing has been " "deleted and no Helix state has changed — click " "下线 again shortly to continue." ), detail={"type": acct_type, "settled_type": settled}, ) elif acct_type and acct_type != "ResourceAccount": # ``ResourceDirectoryMaster`` lands here. Closing the master # would decapitate the directory the credentials themselves # depend on; Aliyun refuses it too, but refusing locally # keeps the blast radius off the provider entirely. return CloseScopeResult( ok=False, error_code="unclosable_member_type", error_message=( f"Aliyun member type {acct_type!r} cannot be closed " "through the Resource Directory. Only members of type " "ResourceAccount (and CloudAccount members that can be " "switched back to ResourceAccount) support DeleteAccount." ), detail={"type": acct_type}, ) # ── Step 3: mandatory deletion pre-check (3–5 min) ────── check = await self._rd_deletion_precheck( account_id, force_fresh=retry_after_failure, ) if check.get("_error"): precheck_error = str(check["_error"]) if converted and "MemberTypeError" in precheck_error: # Same race as the confirmation read above, one step # later: Aliyun reported the new type but the deletion # subsystem has not caught up. Retriable, not fatal. return CloseScopeResult( ok=False, error_code="cloud_account_switch_pending", error_message=( "Aliyun switched this member to a ResourceAccount " "but its deletion pre-check has not picked the new " "type up yet. Nothing has been deleted and no Helix " "state has changed — click 下线 again shortly to " "continue." ), detail={ "converted": True, "type": acct_type, "provider_message": precheck_error[:400], }, ) return CloseScopeResult( ok=False, error_code="deletion_precheck_failed", error_message=( "Aliyun deletion pre-check failed: " + precheck_error ), detail={"converted": converted, "type": acct_type}, ) if str(check.get("Status") or "") != "PreCheckComplete": # Not an error — Aliyun's pre-check genuinely runs for # minutes. Helix's row is untouched; the admin re-clicks and # this method resumes from the finished check. return CloseScopeResult( ok=False, error_code="deletion_precheck_running", error_message=( "Aliyun's member-deletion pre-check is still running " "(it takes 3–5 minutes). " + ( "The account has already been switched from " "CloudAccount to ResourceAccount. " if converted else "" ) + "Nothing has been closed yet and no Helix state has " "changed. Click 下线 again in a few minutes to resume " "— the check is not restarted." ), detail={ "converted": converted, "type": acct_type, "precheck_status": check.get("Status"), }, ) if not _acs3_truthy(check.get("AllowDelete")): blockers = _rd_check_items(check.get("NotAllowReason")) return CloseScopeResult( ok=False, error_code="deletion_not_allowed", error_message=( "Aliyun's pre-check refuses to delete this member. " "Resolve these items in the Aliyun console and retry: " + ("; ".join(blockers) or "(no reason returned)") ), detail={ "converted": converted, "type": acct_type, "not_allow_reason": check.get("NotAllowReason"), "abandonable_checks": check.get("AbandonableChecks"), }, ) # ── Step 4: the real deletion ─────────────────────────── # **Every abandonable check is abandoned.** These are the items # Aliyun is willing to proceed past if you say so explicitly — # "this account still has 5 ModelStudio inference instances, an # OSS bucket, a function…". Until 2026-09-02 this driver sent no # ``AbandonableCheckId`` at all, on the reasoning that # overriding a provider's safety check on the admin's behalf was # not the driver's decision. That reasoning produced a close # that CANNOT COMPLETE: Aliyun accepts the DeleteAccount, then # its asynchronous check fails on exactly those items and stops # dead at ``CheckFailed`` with no retry. Prod proof — account # 5995487234252122, submitted 2026-09-01 14:57 with 16 # abandonable items, ``CheckFailed`` an hour later, account # still alive. # # Withholding them was not caution, it was handing the admin an # unfinishable job: they clicked an irreversible button that # says "所有资源与数据会被永久删除", and then the closure quietly # refused to proceed because those resources exist. The consent # for this already happened at the confirm dialog, and # ``DeleteAccount`` destroys every resource in the member # account anyway ("all resources and data in the member account # are permanently removed" — Aliyun DeleteAccount reference), so # abandoning a check that says "there are resources" cannot # destroy anything the delete itself was not already going to. # # What we owe instead is a RECORD: the exact ids and their # human-readable descriptions go into ``detail`` and from there # into the decommission audit row, so "what did Helix wave # through on my behalf" has a precise answer forever. abandonable = check.get("AbandonableChecks") abandon_ids = _rd_check_ids(abandonable) delete_params: dict[str, Any] = {"AccountId": account_id} if abandon_ids: delete_params["AbandonableCheckId"] = abandon_ids try: deleted = await self._acs3_call( host=_RD_HOST, version=_RD_VERSION, action="DeleteAccount", params=delete_params, ) except RuntimeError as exc: return self._classify_rd_delete_failure( exc, acct_type=acct_type, converted=converted, ) deletion_type = str(deleted.get("DeletionType") or "").strip() # "0" → deleted outright. "1" → Aliyun parks the member in a # silence period (it had recent pay-as-you-go resources) and # deletes it when that expires. Report them apart: "已删除" and # "静默期中" are different facts for anyone auditing this later. direct = deletion_type == "0" return CloseScopeResult( ok=True, provider_state="deleted" if direct else "deletion_silence_period", detail={ "type": acct_type, "converted_from_cloud_account": converted, "deletion_type": deletion_type, "abandoned_check_ids": abandon_ids, "abandoned_checks": _rd_check_items(abandonable), "note": ( "Aliyun deleted the Resource Directory member outright." if direct else "Aliyun moved the member into its deletion silence " "period; the account is unusable now and is deleted " "automatically when the period expires." ), }, ) async def _switch_cloud_account_to_resource_account( self, account_id: str, account: dict[str, Any], ) -> CloseScopeResult | None: """「云账号切换为资源账号」 — returns ``None`` on success. This is the step that makes a real Aliyun closure possible at all. A non-``None`` return is the failure to hand straight back to the caller; Helix state stays untouched either way. The three documented preconditions (legal entity matches the management account, security info set, no enabled root AccessKey) are NOT pre-checked here on purpose. Aliyun is the authority on all three, it reports each with a distinct error code, and a local predicate that disagreed with the provider would block closures that would actually have succeeded — the expensive direction of being wrong. """ try: await self._acs3_call( host=_RD_HOST, version=_RD_VERSION, action="UpdateAccount", params={ "AccountId": account_id, "NewAccountType": "ResourceAccount", }, ) except RuntimeError as exc: msg = str(exc) if "LegalEntityDifferent" in msg: return CloseScopeResult( ok=False, error_code="foreign_legal_entity", error_message=( "This Aliyun member is a CloudAccount whose legal " "entity (实名信息) differs from the Resource " "Directory master's, so it cannot be switched back " "to a ResourceAccount and Aliyun will not let the " "master delete it. Helix cannot close this account. " "It has to be cancelled (账号注销) from its own " "Aliyun console by the account holder. Note that " "removing it from the Resource Directory would NOT " "close it — the account would stay alive and " "billable, just invisible to Helix." ), detail={ "type": "CloudAccount", "member_identity": account.get("IdentityInformation"), "provider_message": msg[:400], }, ) if any(p in msg for p in _RD_DENIED_CODES): return CloseScopeResult( ok=False, error_code="access_denied", error_message=( "Aliyun denied UpdateAccount. The integration " "credentials must be the Resource Directory master " "with AliyunResourceDirectoryFullAccess." ), detail={"provider_message": msg[:400]}, ) return CloseScopeResult( ok=False, error_code="cloud_account_switch_failed", error_message=( "Aliyun refused to switch this CloudAccount back to a " "ResourceAccount, which is the only way it can be " "deleted. Aliyun requires all three of: the member's " "legal entity (实名信息) matches the management " "account's; the member has security info (安全手机号 or " "安全邮箱) configured; and the member has no enabled " "root AccessKey. Fix the failing one in the Aliyun " "console and retry. Aliyun original error: " + msg[:200] ), detail={"provider_message": msg[:400]}, ) return None async def _rd_account_type(self, account_id: str) -> str | None: """Re-read a member's ``Type``; ``None`` when it can't be read. ``None`` deliberately means "don't block on this" — the caller uses it only to detect a not-yet-applied type switch, and a read hiccup must not turn a successful switch into a reported failure. """ try: data = await self._acs3_call( host=_RD_HOST, version=_RD_VERSION, action="GetAccount", params={"AccountId": account_id}, ) except RuntimeError: return None account = data.get("Account") or {} return str(account.get("Type") or "").strip() or None async def _rd_deletion_task_status(self, account_id: str) -> str | None: """``GetAccountDeletionStatus.Status``; ``None`` when no task. Read-only. ``None`` also covers "the read failed" — the caller uses this only to decide whether a previous attempt is in flight or dead, and a read hiccup must degrade to "proceed normally", never to a claim about the deletion. """ try: data = await self._acs3_call( host=_RD_HOST, version=_RD_VERSION, action="GetAccountDeletionStatus", params={"AccountId": account_id}, ) except RuntimeError: return None info = data.get("RdAccountDeletionStatus") or {} return str(info.get("Status") or "").strip() or None async def _rd_deletion_precheck( self, account_id: str, *, force_fresh: bool = False, ) -> dict[str, Any]: """Drive ``CheckAccountDelete`` → ``GetAccountDeletionCheckResult``. Returns the ``AccountDeletionCheckResultInfo`` payload, or a dict carrying ``_error`` when the provider rejected the check outright. A result whose ``Status`` is still ``PreChecking`` is a normal, expected return — the caller turns it into a resumable "come back in a few minutes", not a failure. Reads the existing check task BEFORE starting one. Aliyun's check takes 3–5 minutes, so a second decommission click must attach to the check already in flight; kicking off a fresh one each time would reset the clock forever and the close could never complete. ``force_fresh`` is the exception: after a deletion that died, the stored result describes the attempt that failed, so attaching to it would reproduce the failure exactly. """ # Every provider error in here becomes ``_error`` rather than # propagating: ``close_scope_on_provider`` is contracted to # return a :class:`CloseScopeResult`, and a bare RuntimeError # escaping to the service turns an actionable 502 into an # opaque 500 (the service only special-cases # ``NotImplementedError``). try: # After a dead attempt the stored check result is stale — # attaching to it just refails. Start a new one. info = None if force_fresh else await self._rd_read_deletion_check( account_id, ) if info is None: await self._acs3_call( host=_RD_HOST, version=_RD_VERSION, action="CheckAccountDelete", params={"AccountId": account_id}, ) info = await self._rd_read_deletion_check(account_id) for _ in range(_RD_PRECHECK_POLL_ATTEMPTS): if info is not None and str( info.get("Status") or "", ) == "PreCheckComplete": return info await asyncio.sleep(_RD_PRECHECK_POLL_INTERVAL_SEC) info = await self._rd_read_deletion_check(account_id) except RuntimeError as exc: return {"_error": str(exc)[:400]} return info if info is not None else {"Status": "PreChecking"} async def _rd_read_deletion_check( self, account_id: str, ) -> dict[str, Any] | None: """``GetAccountDeletionCheckResult``; ``None`` when no task exists.""" try: data = await self._acs3_call( host=_RD_HOST, version=_RD_VERSION, action="GetAccountDeletionCheckResult", params={"AccountId": account_id}, ) except RuntimeError as exc: if "EntityNotExists.CheckAccount" in str(exc): return None raise return data.get("AccountDeletionCheckResultInfo") or None def _classify_rd_delete_failure( self, exc: RuntimeError, *, acct_type: str, converted: bool, ) -> CloseScopeResult: """Map a ``DeleteAccount`` rejection onto actionable admin guidance.""" msg = str(exc) if any(p in msg for p in _RD_NOT_FOUND_CODES): return CloseScopeResult( ok=True, already_closed=True, provider_state="not_in_directory", detail={"reason": "Account already removed", "type": acct_type}, ) if any(p in msg for p in _RD_DENIED_CODES): return CloseScopeResult( ok=False, error_code="access_denied", error_message=( "Aliyun denied DeleteAccount. The integration " "credentials must be the Resource Directory master " "with ResourceDirectoryFullAccess." ), ) # Aliyun rejects DeleteAccount when the member still owns active # resources, has outstanding charges, or has not completed # real-name verification. The provider returns symbolic error # codes for each bucket — anchor on those codes (prefixed forms # like ``Member.NotAllowedDelete.HasUnsettledBill``) rather than # the 4-char ``"Bill"`` substring that the Round-52 security + # logic scans flagged as too loose (would mis-route any error # mentioning "Bill" in free-text). if "MemberTypeError.DeleteAccount" in msg: return CloseScopeResult( ok=False, error_code="unclosable_member_type", error_message=( "Aliyun still classifies this member as a CloudAccount " "or as invited-into-the-directory, so DeleteAccount is " "refused. Only members created inside the Resource " "Directory and currently typed ResourceAccount can be " "deleted. Aliyun original error: " + msg[:200] ), detail={"type": acct_type, "converted": converted}, ) if any(p in msg for p in _RD_RESOURCES_EXISTING_CODES): return CloseScopeResult( ok=False, error_code="resources_existing", error_message=( "Aliyun refuses to delete a Resource Directory member " "that still owns resources. Empty the member account " "first — release every ECS / RDS / OSS / SLB / " "cloud-disk asset (and wait for any pre-paid instances " "to expire or request a refund) — then retry " "decommission. Aliyun original error: " + msg[:200] ), detail={"provider_message": msg[:400]}, ) if any(p in msg for p in _RD_REAL_NAME_CODES): return CloseScopeResult( ok=False, error_code="not_real_name_authed", error_message=( "Aliyun requires the member account to complete " "real-name verification before DeleteAccount is " "allowed. Finish real-name auth on the member's Aliyun " "console (个人/企业实名认证) and retry." ), detail={"provider_message": msg[:400]}, ) if any(p in msg for p in _RD_PENDING_CHARGES_CODES): return CloseScopeResult( ok=False, error_code="pending_charges", error_message=( "Aliyun refuses to delete a member with outstanding " "billing obligations. Settle or refund the balance on " "Aliyun billing console and retry. Aliyun original " "error: " + msg[:200] ), detail={"provider_message": msg[:400]}, ) if "NotSupport.HasPayerAccount" in msg: return CloseScopeResult( ok=False, error_code="payer_relationship_active", error_message=( "This member is still in a financial (结算) " "relationship with another account. Release the payer " "relationship in the Aliyun billing console, then " "retry. Aliyun original error: " + msg[:200] ), detail={"provider_message": msg[:400]}, ) if "DelegatedAdministator" in msg or "DelegatedAdministrator" in msg: return CloseScopeResult( ok=False, error_code="delegated_administrator", error_message=( "This member is registered as a delegated " "administrator for one or more Aliyun services. " "Deregister it first, then retry. Aliyun original " "error: " + msg[:200] ), detail={"provider_message": msg[:400]}, ) return CloseScopeResult( ok=False, error_code="delete_failed", error_message=f"Aliyun DeleteAccount: {exc}", ) # ── Post-close reconciliation (read-only) ────────────────── async def get_scope_closure_state( self, scope: CloudScopeRef, ) -> ScopeClosureState: """Re-read Aliyun's view of a member Helix already closed. ``GetAccountDeletionStatus`` is the authority: it is the only surface that distinguishes "the deletion is running" from "the deletion FAILED", and the failure case is the one that matters — Aliyun's ``CheckFailed`` / ``DeleteFailed`` leave the account alive with no further automatic progress, while Helix's row already says ``closed``. Documented ``Status`` vocabulary: ``Checking`` / ``Deleting`` (in flight), ``Success`` (deleted), ``CheckFailed`` / ``DeleteFailed`` (dead in the water). When no deletion task exists at all, fall back to ``GetAccount``: gone from the directory ⇒ closed; still there and typed ``CloudAccount`` ⇒ someone switched it back and it is alive again. """ if scope.scope_type != "aliyun_account": raise NotImplementedError( "Aliyun driver has no post-close state for " f"scope_type={scope.scope_type!r}", ) account_id = (scope.external_id or "").strip() try: data = await self._acs3_call( host=_RD_HOST, version=_RD_VERSION, action="GetAccountDeletionStatus", params={"AccountId": account_id}, ) except RuntimeError as exc: msg = str(exc) if "EntityNotExists" in msg or any( p in msg for p in _RD_NOT_FOUND_CODES ): # No deletion task on record. Either it finished and # Aliyun dropped the task, or one was never created. # GetAccount tells those apart. return await self._rd_closure_state_from_account(account_id) return ScopeClosureState( disposition="unknown", error_message=f"Aliyun GetAccountDeletionStatus: {msg[:300]}", ) info = data.get("RdAccountDeletionStatus") or {} status = str(info.get("Status") or "").strip() fail_reasons = _rd_check_items(info.get("FailReasonList")) detail = { "aliyun_status": status, "deletion_type": info.get("DeletionType"), "deletion_time": info.get("DeletionTime") or None, "fail_reasons": fail_reasons, } if status == "Success": return ScopeClosureState( disposition="closed", provider_state="deleted", detail=detail, ) if status in {"CheckFailed", "DeleteFailed"}: return ScopeClosureState( disposition="failed", provider_state="deletion_failed", detail=detail, error_message=( f"Aliyun reports the member deletion {status}. The " "account is still alive and will not progress on its " "own. " + ("Blocking items: " + "; ".join(fail_reasons) if fail_reasons else "Aliyun returned no reason.") ), ) if status in {"Checking", "Deleting"}: return ScopeClosureState( disposition="pending", provider_state=( "deletion_silence_period" if status == "Checking" else "deleting" ), detail=detail, ) # Unrecognised status: do NOT guess. Leaving it ``unknown`` keeps # the scope in the reconcile queue and changes nothing, which is # the only safe reading of a vocabulary we do not know. return ScopeClosureState( disposition="unknown", detail=detail, error_message=( f"Aliyun returned an unrecognised deletion status {status!r}" ), ) async def _rd_closure_state_from_account( self, account_id: str, ) -> ScopeClosureState: """Fallback when Aliyun has no deletion task for this member.""" try: data = await self._acs3_call( host=_RD_HOST, version=_RD_VERSION, action="GetAccount", params={"AccountId": account_id}, ) except RuntimeError as exc: msg = str(exc) if any(p in msg for p in _RD_NOT_FOUND_CODES): return ScopeClosureState( disposition="closed", provider_state="not_in_directory", detail={"reason": "Member no longer in the directory"}, ) return ScopeClosureState( disposition="unknown", error_message=f"Aliyun GetAccount: {msg[:300]}", ) account = data.get("Account") or {} acct_type = str(account.get("Type") or "").strip() status = str(account.get("Status") or "").strip() detail = {"type": acct_type, "aliyun_status": status} if status in _RD_GONE_STATUSES: return ScopeClosureState( disposition="closed", provider_state=status.lower(), detail=detail, ) # Present, no deletion task, not in a gone state — this member is # running. The common cause is a switch back to CloudAccount # (``PromoteResourceAccount``), which someone can do from the RD # console and which silently cancels the whole close. return ScopeClosureState( disposition="alive", provider_state="active", detail=detail, error_message=( "Aliyun still lists this member as a live Resource " f"Directory account (type={acct_type or '?'}, " f"status={status or '?'}) with no deletion task on " "record, but Helix has it marked closed and has already " "disabled its grants." ), ) async def _evacuate_ram_users( self, iam_account_ids: list[str], ) -> CloseScopeResult: """Run the per-user RAM revoke chain for every Helix-tracked user.""" if not iam_account_ids: return CloseScopeResult( ok=True, provider_state="users_evacuated", revoked_users=0, detail={"note": "No Helix-tracked RAM users on this scope."}, ) evacuated: list[dict[str, Any]] = [] failures: list[dict[str, Any]] = [] for user_name in iam_account_ids: ctx = CloudAccountRevokeContext( provider=self.provider, account_id=user_name, ) try: result = await self.revoke_account(ctx) except RuntimeError as exc: # config / credential errors failures.append({"user": user_name, "error": str(exc)}) continue if result.ok: evacuated.append({ "user": user_name, "raw": result.raw_response or {}, }) else: failures.append({ "user": user_name, "error": result.error_message or "no error message returned by driver", }) if failures: return CloseScopeResult( ok=False, error_code="evacuation_partial", error_message=( f"Aliyun evacuation partial: {len(failures)} of " f"{len(iam_account_ids)} RAM user(s) failed to delete. " "Inspect the audit detail for per-user errors." ), revoked_users=len(evacuated), detail={"evacuated": evacuated, "failures": failures}, ) return CloseScopeResult( ok=True, provider_state="users_evacuated", revoked_users=len(evacuated), detail={"evacuated": evacuated}, ) # ── Resource-creator lookup via ActionTrail ───────────────── # # Aliyun ActionTrail provides a CloudTrail-equivalent API for # who-did-what audit. Default 90-day retention on the # "default" Trail (matches AWS); customers with longer-term # OSS-backed Trails can extend. # # API: ``actiontrail.aliyuncs.com`` ``2020-07-06`` ``LookupEvents`` # Params (ACS3-style): # * ``LookupAttribute.1.Key=ResourceName`` # * ``LookupAttribute.1.Value=`` # * ``StartTime`` / ``EndTime`` (ISO-8601) # * ``MaxResults`` (default 50, max 50) # Response shape (Aliyun OpenAPI v3 asymmetry — the wrapper is # PascalCase but event objects are camelCase): # ``Events: [{eventName, eventTime, userIdentity: {type, # accessKeyId, userName, principalId, accountId}, ...}]`` # ``NextToken: "..."`` # # We post-filter ``eventName`` to a narrow create-style allowlist # (RunInstances / CreateBucket / CreateDBInstance / ...) so # noisy modify events don't pollute the result. All event-field # reads go through :meth:`_ev_get` so a future endpoint variant # that flips back to PascalCase doesn't regress to 0% resolution. # ActionTrail is REGION-PINNED on Aliyun. The bare global host # ``actiontrail.aliyuncs.com`` does NOT resolve in DNS (verified # 2026-05-25 against prod EC2 in ap-southeast-1: ``getent hosts # actiontrail.aliyuncs.com`` returns empty; same query for the # regional ``actiontrail.cn-shanghai.aliyuncs.com`` / # ``actiontrail.ap-southeast-1.aliyuncs.com`` returns IPs). Events # land in whichever region the resource lives in, so we have to # walk the integration's region list — first non-DNS-error region # whose result is non-empty wins; if all regions are clean but # empty we treat it as ``None`` (terminal not_found, same as # AWS / GCP / Azure semantics). CN default below targets the # CN-mainland tenant; the ``AlicloudIntlCloudDriver`` subclass # overrides to the intl region list. _ACTIONTRAIL_REGIONS: tuple[str, ...] = ("cn-shanghai",) # ActionTrail OpenAPI v3 (``2020-07-06``) returns its top-level # wrapper in PascalCase (``Events``, ``NextToken``, ``RequestId``) # but each event object's fields are camelCase # (``eventName``, ``eventTime``, ``userIdentity``, …). This # asymmetry was the prod root cause of 0% resolution on the # alicloud_intl integration (3743 rows all stuck at ``not_found``): # ``resp.get("Events")`` returned the list, but ``ev.get("EventName")`` # was always None so every event was filtered out, leaving the # earliest-create tracker empty. # # Aliyun's own SDKs work around this by exposing both shapes; # we do the same with a case-tolerant accessor used everywhere # we read an event field. PascalCase is kept as a fallback so a # future v3 endpoint variant (or a re-signed legacy endpoint) # that flips back to PascalCase doesn't silently regress. @staticmethod def _ev_get( obj: dict[str, Any] | None, *keys: str, ) -> Any: """Return the first non-None value among ``keys`` (case-tolerant). For each requested key, try the key as-given AND with its first letter case-flipped. ``_ev_get(ev, "EventName")`` looks up ``EventName`` then ``eventName``. Returns ``None`` if nothing matched. """ if not obj: return None for key in keys: if key in obj and obj[key] is not None: return obj[key] # case-flip first character if key and key[0].islower(): alt = key[0].upper() + key[1:] elif key: alt = key[0].lower() + key[1:] else: continue if alt in obj and obj[alt] is not None: return obj[alt] return None _ALIYUN_CREATE_EVENT_NAMES = frozenset({ # ECS "RunInstances", "CreateInstance", # OSS "CreateBucket", # RDS "CreateDBInstance", # Slb / Alb / Nlb "CreateLoadBalancer", # OSS / Object services "PutBucket", # CDN / DCDN "AddCdnDomain", "AddDcdnDomain", # DNS "AddDomain", "AddDomainRecord", # FunctionCompute "CreateFunction", "CreateService", # MongoDB / Redis / PolarDB / ADB "CreateDBCluster", "CreateMongoDBInstance", # MaxCompute / DataWorks projects "CreateProject", # CS (Container Service) "CreateCluster", # ROS (Resource Orchestration) "CreateStack", # RAM "CreateUser", "CreateRole", # KMS "CreateKey", # ApiGateway "CreateApi", "CreateApiGroup", }) @classmethod def _aliyun_extract_lookup_name( cls, resource_id: str, ) -> str | None: """Reduce a Helix-stored Aliyun resource_id to the value ActionTrail's ResourceName attribute expects. Aliyun resource ids are typically already short (``i-bp1...``, bucket name, ``rm-...``, ``ali-uid:...``). Synthetic ids (``aliyun-account:...`` / ``aliyun-bailian:...``) have no ActionTrail counterpart. """ if not resource_id: return None if resource_id.startswith(( "aliyun-account:", "aliyun-bailian:", "aliyun-aggregated:", )): return None # Bare Aliyun resource ids pass through. return resource_id async def get_resource_creator( self, resource_id: str, *, resource_type: str | None = None, scope: CloudScopeRef | None = None, ) -> ResourceCreatorInfo | None: """Resolve "who created this resource" via ActionTrail. 90-day retention on the default Trail. Returns the EARLIEST matching create event within the window; ``None`` when no create event is found. """ import datetime as _dt lookup_name = self._aliyun_extract_lookup_name(resource_id) if not lookup_name: return None # Defence: a subclass that overrides _ACTIONTRAIL_REGIONS to # an empty tuple would silently skip the entire region walk # and return None → service writes terminal not_found on # every resource regardless of whether it's attributable. # That's an operator-invisible misconfiguration; refuse with # ``NotImplementedError`` (not ``RuntimeError``) so the service # layer's ``except NotImplementedError`` arm at # ``resolve_resource_creator`` maps it to terminal # ``status='unsupported'`` — the cron sweep stops retrying # immediately (no 5-attempt waste) and the HTTP creator # endpoint returns a clean ``unsupported`` payload instead # of a 500. The operator sees the misconfiguration via the # row's ``last_error`` in the admin UI and can fix the # subclass. if not self._ACTIONTRAIL_REGIONS: raise NotImplementedError( f"{type(self).__name__} has no ActionTrail regions " "configured; subclass must set _ACTIONTRAIL_REGIONS " "to at least one region.", ) now_utc = _dt.datetime.now(tz=_dt.UTC) start = (now_utc - _dt.timedelta(days=90)).strftime( "%Y-%m-%dT%H:%M:%SZ", ) end = now_utc.strftime("%Y-%m-%dT%H:%M:%SZ") # Walk ActionTrail's ``NextToken`` paging the same way the # Azure driver walks ``nextLink``. ActionTrail returns # events newest-first (mirrors AWS CloudTrail). For a # heavily-mutated resource the create event is on the LAST # page, so we walk forward bounded by ``_PAGE_CAP=50`` # (50 × 50 ≈ 2500 events / 90-day window — covers typical # lifecycle). When the cap is hit, we return the oldest # event seen so far — biased newer than true create but # still useful; the cron sweep can re-attempt with # ``force=True``. _PAGE_CAP = 50 earliest: dict[str, Any] | None = None earliest_at_iso: str | None = None # Walk the integration's region list (see # ``_ACTIONTRAIL_REGIONS`` docstring above for why this has # to be multi-region). Track ``regions_reached`` to # distinguish "every region was DNS/network-broken" (raise) # from "every region responded with no matching events" # (terminal ``None`` → ``not_found``). regions_reached = 0 last_transport_error: Exception | None = None for region in self._ACTIONTRAIL_REGIONS: host = f"actiontrail.{region}.aliyuncs.com" page_params = { "LookupAttribute.1.Key": "ResourceName", "LookupAttribute.1.Value": lookup_name, "StartTime": start, "EndTime": end, "MaxResults": "50", } pages_walked = 0 region_reached = False while pages_walked < _PAGE_CAP: try: resp = await self._acs3_call( host=host, version="2020-07-06", action="LookupEvents", method="POST", params=page_params, ) except RuntimeError as exc: msg = str(exc) if "OutOfRange" in msg or "InvalidStartTime" in msg: raise ResourceCreatorWindowExceeded( f"Aliyun ActionTrail rejected window for " f"{lookup_name!r}: {msg}", ) from exc raise except ( httpx.TransportError, httpx.DecodingError, OSError, TimeoutError, ) as exc: # TRANSPORT-only failures for THIS region — record # and try the next. ``httpx.TransportError`` covers # ConnectError / Timeout (Connect/Read/Write/Pool) / # NetworkError / ProtocolError (Local+Remote) / # ProxyError / UnsupportedProtocol. ``DecodingError`` # catches malformed-body cases from a single # region. ``OSError`` catches socket-level errors # not wrapped by httpx, including ``ssl.SSLError`` # (TLS handshake / cert mismatch is socket-level). # ``TimeoutError`` covers ``asyncio.TimeoutError`` # which httpx may pass through verbatim. # # Explicitly NOT caught here: ``RuntimeError`` (4xx # / 5xx wrapped by ``_acs3_call`` — auth gaps and # validation errors MUST propagate so the operator # sees them rather than seeing the driver silently # walk to the next region and report ``not_found`` # on a misconfigured RAM policy). last_transport_error = exc break region_reached = True pages_walked += 1 # ActionTrail v3: top-level wrapper is PascalCase # but event objects are camelCase — read both via # ``_ev_get`` so a single asymmetric response doesn't # silently filter every event out. events = resp.get("Events") or resp.get("events") or [] for ev in events: event_name = self._ev_get(ev, "eventName", "EventName") if event_name not in self._ALIYUN_CREATE_EVENT_NAMES: continue ev_time = self._ev_get(ev, "eventTime", "EventTime") if not ev_time: continue if ( earliest_at_iso is None or ev_time < earliest_at_iso ): earliest = ev earliest_at_iso = ev_time next_token = resp.get("NextToken") or resp.get("nextToken") if not next_token: break # Successive page request: keep filters + add NextToken. page_params = dict(page_params) page_params["NextToken"] = next_token if region_reached: regions_reached += 1 if regions_reached == 0 and last_transport_error is not None: # Every region failed at transport layer — surface so # service layer records ``status='error'`` instead of a # false ``not_found``. raise last_transport_error if earliest is None: return None return self._aliyun_event_to_creator_info(earliest) @classmethod def _aliyun_event_to_creator_info( cls, event: dict[str, Any], ) -> ResourceCreatorInfo: """Parse one ActionTrail event → ResourceCreatorInfo. ActionTrail v3 (``2020-07-06``) returns event objects in camelCase (``eventName``, ``userIdentity.userName``, …). Some legacy endpoint variants may still emit PascalCase; use :meth:`_ev_get` to accept both shapes so a single asymmetric response doesn't silently drop the attribution. """ user_identity = ( cls._ev_get(event, "userIdentity", "UserIdentity") or {} ) # ``principalId`` is the canonical RAM principal id; userName # is a friendlier handle that's often (but not always) an # email. principal_id = ( cls._ev_get(user_identity, "principalId", "PrincipalId") or cls._ev_get(user_identity, "userName", "UserName") or cls._ev_get(user_identity, "accessKeyId", "AccessKeyId") or "" ) user_name = ( cls._ev_get(user_identity, "userName", "UserName") or "" ) email: str | None = None if "@" in user_name: email = user_name display_name: str | None = None if email: display_name = email.split("@", 1)[0].replace(".", " ").title() elif user_name: display_name = user_name created_at_cloud: datetime | None = None ev_time_iso = cls._ev_get(event, "eventTime", "EventTime") if isinstance(ev_time_iso, str): import datetime as _dt try: created_at_cloud = _dt.datetime.fromisoformat( ev_time_iso.replace("Z", "+00:00"), ) except ValueError: created_at_cloud = None _RAW_CAP_BYTES = 16 * 1024 # Persist whichever case the response actually carried — this # is debugging gold when an operator inspects ``raw`` later. We # normalise the OUTPUT keys to snake_case (Helix convention) # and pull each value via the case-tolerant accessor. raw_to_store: dict[str, Any] = { "event_name": cls._ev_get(event, "eventName", "EventName"), "event_id": cls._ev_get(event, "eventId", "EventId"), "event_time": ev_time_iso, "event_source": cls._ev_get(event, "eventSource", "EventSource"), "user_identity": { "type": cls._ev_get(user_identity, "type", "Type"), "access_key_id": cls._ev_get( user_identity, "accessKeyId", "AccessKeyId", ), "user_name": user_name or None, "principal_id": cls._ev_get( user_identity, "principalId", "PrincipalId", ), "account_id": cls._ev_get( user_identity, "accountId", "AccountId", ), }, } # Drop None-only sub-keys so the JSON stays readable. ui = raw_to_store["user_identity"] raw_to_store["user_identity"] = { k: v for k, v in ui.items() if v is not None } if len(json.dumps(raw_to_store, default=str)) > _RAW_CAP_BYTES: raw_to_store.pop("user_identity", None) return ResourceCreatorInfo( principal_id=str(principal_id), email=email, display_name=display_name, created_at_cloud=created_at_cloud, source="aliyun_actiontrail", raw=raw_to_store, )