"""UCloud cloud driver — full lifecycle. CMDB discovery via :class:`UCloudProvider` (servers / DNS / ULB / UCDN / UFile). Domain registration is NOT covered — UCloud exposes no public "list all registered domains" API (see the UCloud provider module docstring). Account lifecycle: **IAM mode**. UCloud's IAM product (``iam-api``) uses the ``User`` / ``Policy`` / ``AccessKey`` terminology; "SubAccount" is the UI name but the API calls are ``CreateUser`` / ``ListUsers`` / ``DeleteUser`` / ``ListPolicies`` / ``AttachPoliciesToUser`` / ``DetachPoliciesFromUser`` / ``ListAccessKeys`` /``DeleteAccessKey``. A grant = a UCloud IAM User (``CreateUser``) + a policy attachment (``AttachPoliciesToUser``). Helix uses the tenant as the single ``standalone_iam`` scope (consistent with Tencent). Role key convention: ``role_key = UCloud PolicyName`` (matches the UCloud console). Grant resolves name → PolicyURN internally via ``ListPolicies``. ``owner_role_keys`` contains the built-in ``AdminPolicy`` name. Daily billing: ``ListUBillDetail`` with ``BillingCycle=YYYY-MM`` aggregated client-side by day. UCloud has no true per-day API, so we return rows anchored to the monthly bill's breakdown (each resource's PayTime is day-granular → sum per (day, BusinessName)). """ from __future__ import annotations import asyncio import logging import math from datetime import UTC, date, datetime from typing import Any import httpx from app.domains.cmdb.providers.exceptions import redact_provider_error from app.domains.cmdb.providers.ucloud import ( _UNSET, _UNSET_TYPE, UCLOUD_PRIVATE_KEY_FIELD, UCLOUD_PUBLIC_KEY_FIELD, UCloudProvider, _first_present, _is_retryable_retcode, _sign_params, ) from app.integrations.driver_base import ( CloseScopeResult, CloudAccessContext, CloudAccountRevokeContext, CloudDailyCost, CloudGrantResult, CloudPrincipalNotFoundError, CloudRoleInfo, CloudScopeInfo, CloudScopeRef, CostFact, DiscoveredGrant, ProvisioningResult, ) from app.integrations.drivers._cloud_shared import ( _CloudDriverBase, account_id_candidates, ) from app.integrations.registry import register_driver from app.integrations.schemas import ( IntegrationFieldInfo, IntegrationTestResult, ) logger = logging.getLogger(__name__) _API_ENDPOINT = "https://api.ucloud.cn" async def _signed_call( client: httpx.AsyncClient, public_key: str, private_key: str, action: str, params: dict[str, Any] | None = None, *, project_id: str | None = None, retryable: bool = False, ) -> dict[str, Any]: """Shared async wrapper around UCloud API calls with v1 signing. Separate from :class:`UCloudProvider._call` so lifecycle paths (IAM, billing) can fire without instantiating a full provider. Error translation mirrors the provider's (RetCode 100/230/231 = auth; other non-zero = transient). ``retryable`` (default ``False``): when ``True`` the call is retried in-place — 3 attempts, exponential backoff — on transport errors, HTTP 5xx, and body-level transient RetCodes (see :func:`app.domains.cmdb.providers.ucloud._is_retryable_retcode`, e.g. 8498 "Call third api service timeout"). This mirrors the CMDB provider's ``_call`` retry budget so a single UCloud gateway timeout doesn't fail the cost_facts cron or an IAM read. MUST stay ``False`` for NON-idempotent mutations (``CreateUser`` / ``DeleteUser`` / ``AttachPoliciesToUser`` / ``DetachPoliciesFromUser`` / ``DeleteAccessKey``): a retry after a timeout that actually succeeded on the cloud side would surface a spurious "already exists" / "not found" and corrupt the provisioning / offboarding verdict. Those flows are made resilient at the JOB level instead (idempotent re-run on the next sweep). Read actions (``GetRegion`` / ``List*`` / ``ListUBillDetail``) are idempotent and pass ``retryable=True``. """ if not public_key or not private_key: raise RuntimeError("UCloud public_key / private_key not configured") full_params: dict[str, Any] = { "Action": action, "PublicKey": public_key, } if project_id: full_params["ProjectId"] = project_id if params: full_params.update( {k: str(v) for k, v in params.items() if v is not None}, ) # UCloud's v1 signature is NOT time-bound (``_sign_params`` only # sorts + concatenates — no Timestamp / nonce), so the same signed # payload is safe to re-POST on retry without re-signing. full_params["Signature"] = _sign_params(full_params, private_key) attempts = 3 if retryable else 1 last_exc: Exception | None = None for attempt in range(attempts): try: resp = await client.post(_API_ENDPOINT, data=full_params) except httpx.HTTPError as exc: if not retryable: # Preserve the historical raw-propagation contract for # mutations — a timed-out mutation must NOT silently # re-fire. raise last_exc = RuntimeError( f"UCloud {action} transport error: " f"{redact_provider_error(exc, max_len=400)}", ) if attempt < attempts - 1: await asyncio.sleep(2 ** attempt) continue raise last_exc from exc if resp.status_code in (401, 403): raise RuntimeError(f"UCloud auth error: HTTP {resp.status_code}") if retryable and resp.status_code >= 500 and attempt < attempts - 1: # Drain + release the connection before the backoff sleep # (parity with the provider ``_call`` 5xx path). await resp.aclose() await asyncio.sleep(2 ** attempt) continue resp.raise_for_status() try: data: dict[str, Any] = resp.json() except ValueError as exc: # The non-JSON body can be an upstream edge-proxy / WAF page # that echoes the inbound request including signed query # params (Signature, public_key). Scrub through # redact_provider_error so those parameters don't land in the # failure-notifier card or in ``last_test_message``. raise RuntimeError( f"UCloud non-JSON response: {redact_provider_error(resp.text, max_len=400)}", ) from exc ret = int(data.get("RetCode") or 0) if ret == 0: return data msg = str(data.get("Message") or action) if ( retryable and _is_retryable_retcode(ret, msg) and attempt < attempts - 1 ): await asyncio.sleep(2 ** attempt) continue raise RuntimeError(f"UCloud {action} RetCode={ret}: {msg}") # Defensive — every in-loop path returns or raises. if last_exc is not None: raise last_exc raise RuntimeError(f"UCloud {action}: retry budget exhausted") # Admin-editable fallback when platform_settings empty. UCloud's built-in # policies (``AdminPolicy``, ``ReadOnlyPolicy``, ``BillPolicy``, etc.) # are namespace-wide; the list below is the intersection of what every # mainstream UCloud tenant has out of the box. _UCLOUD_FALLBACK_POLICIES: list[dict[str, Any]] = [ { "role_key": "AdminPolicy", "display_name": "AdminPolicy", "description": "管理员权限 — 所有资源读写 + IAM 管理。慎重授予。", }, { "role_key": "ReadOnlyPolicy", "display_name": "ReadOnlyPolicy", "description": "所有资源的只读权限。", }, { "role_key": "BillPolicy", "display_name": "BillPolicy", "description": "账单和账户余额只读。", }, { "role_key": "UHostFullAccess", "display_name": "UHostFullAccess", "description": "UHost 云主机的全权管理。", }, { "role_key": "UHostReadOnly", "display_name": "UHostReadOnly", "description": "UHost 云主机的只读访问。", }, { "role_key": "UFileReadOnly", "display_name": "UFileReadOnly", "description": "UFile 对象存储的只读访问。", }, { "role_key": "UDNRFullAccess", "display_name": "UDNRFullAccess", "description": "UCloud DNS 记录的全权管理。", }, ] @register_driver class UCloudDriver(_CloudDriverBase): provider = "ucloud" display_name = "UCloud" description = ( "UCloud(UHost / UDNR / ULB / UCDN / UFile / 域名 + SubAccount 子账号 + 账单)。" "IAM 模式 — 单租户下建 SubAccount + 挂策略 + 离职删除账号。" ) provider_cls = UCloudProvider supported_capabilities = frozenset({ "servers", "dns_records", "load_balancers", "cdn", "object_storage", "user_revoke", }) capabilities = [ "servers", "dns_records", "load_balancers", "cdn", "object_storage", "user_revoke", ] account_mode = "iam" # UCloud SubAccount grants are tenant-wide (one sub-account has one # set of attached policies across all projects it belongs to). grants_scope_wide = False owner_role_keys = frozenset({"AdminPolicy"}) # UCloud bills natively in CNY (``AmountReal`` / ``Amount`` on # ``ListUBillDetail`` are CNY decimals). Mirror Aliyun + Tencent — # the cost-ingestion service resolves ``cny_per_usd`` via # ``app.core.fx_rate.resolve_cny_per_usd`` and passes it in; # ``list_cost_facts`` raises if the rate is missing so a half- # configured platform can't land ``amount_usd=0`` rows that would # silently understate UCloud spend in every USD aggregate. requires_cny_fx = True supports_test_connection = True config_schema = [ IntegrationFieldInfo( key=UCLOUD_PUBLIC_KEY_FIELD, label="Public Key", is_secret=True, required=True, placeholder="long base-like string from UCloud console", ), IntegrationFieldInfo( key=UCLOUD_PRIVATE_KEY_FIELD, label="Private Key", is_secret=True, required=True, ), IntegrationFieldInfo( key="project_ids", label="Project IDs(多项目,逗号分隔)", is_secret=False, required=False, placeholder="org-xxxxxx,org-yyyyyy", hint=( "UCloud 的 UHost/ULB/UFile/UCDN 列表接口按单个 ProjectId " "过滤,主账号下的每个项目必须逐个声明才会被同步扫描。" "填多个项目 ID(逗号分隔)覆盖全部项目;留空则只同步默认" "项目——其余项目里的资源对 Helix 不可见(2026-07 DLP-test " "漂移事故根因)。填写后优先于下方单个 Project ID。" ), ), IntegrationFieldInfo( key="project_id", label="Project ID(单项目,旧字段;留空=默认项目)", is_secret=False, required=False, placeholder="org-xxxxxx", hint="向后兼容的单项目字段;上方 Project IDs 非空时此字段被忽略。", ), IntegrationFieldInfo( key="regions", label="Regions(逗号分隔)", is_secret=False, required=False, default=( "cn-bj2,cn-sh2,cn-gd,cn-qz,cn-north-07,cn-wlcb,hk," "us-ca,us-ws,ge-fra,sg,kr-seoul,th-bkk,jpn-tky" ), placeholder="cn-bj2,cn-sh2", ), # UCloud UDNR has no "list all zones" API. Admin declares which # zones the DNS sync should walk; empty → DNS sync surfaces # "skipped — no zones configured" rather than an empty sync # that would sweep historical records. IntegrationFieldInfo( key="dns_zones", label="DNS Zones(逗号分隔,留空则跳过 DNS 同步)", is_secret=False, required=False, placeholder="example.com,example2.cn", ), ] # ── helpers ──────────────────────────────────────────────── def _keys(self) -> tuple[str, str]: public_key = str(self.config.get(UCLOUD_PUBLIC_KEY_FIELD) or "").strip() private_key = str(self.config.get(UCLOUD_PRIVATE_KEY_FIELD) or "").strip() return public_key, private_key def _project_id(self) -> str | None: pid = str(self.config.get("project_id") or "").strip() return pid or None async def _call( self, action: str, params: dict[str, Any] | None = None, *, retryable: bool = False, project_id: str | None | _UNSET_TYPE = _UNSET, ) -> dict[str, Any]: public_key, private_key = self._keys() # ``project_id`` override (default ``_UNSET`` → integration's # configured singular ``project_id``, preserving existing IAM / # discovery behavior). Billing reads pass ``project_id=None`` # explicitly to force an ACCOUNT-WIDE fetch: UCloud billing is # account-level, so ``ListUBillDetail`` with no ProjectId returns # every project's line items in one call. Threading it through # the legacy singular field would silently filter the bill to one # project (the ``project_ids`` multi-project field is for resource # discovery, not billing) — so billing deliberately ignores both. effective_pid = ( self._project_id() if isinstance(project_id, _UNSET_TYPE) else project_id ) # Per-call AsyncClient matches the Aliyun / Azure driver pattern # and avoids ``ResourceWarning: unclosed `` # when the framework drops the driver without calling close() # (no uniform close() contract across lifecycle paths today). # HTTP keep-alive at the OS level still makes paginated loops # cheap; the overhead is one socket per request, not a full # TLS handshake. # ``retryable`` is forwarded to ``_signed_call`` — pass True ONLY # for idempotent reads (``GetRegion`` / ``List*`` / # ``ListUBillDetail``); mutations keep the default (no retry) so a # timed-out write never silently re-fires. See ``_signed_call``. async with httpx.AsyncClient(timeout=30.0, trust_env=False) as client: return await _signed_call( client, public_key, private_key, action, params, project_id=effective_pid, retryable=retryable, ) # ── Health probe ──────────────────────────────────────── async def test_connection(self) -> IntegrationTestResult: public_key, private_key = self._keys() if not public_key or not private_key: return IntegrationTestResult( ok=False, message="UCloud public_key / private_key not configured", ) # ``GetRegion`` is the lightest possible auth probe — lists # available regions for the AK/SK, returns immediately. try: data = await self._call("GetRegion", retryable=True) except Exception as exc: # noqa: BLE001 return IntegrationTestResult( ok=False, message=f"{type(exc).__name__}: {exc}", ) region_count = len(data.get("Regions") or []) return IntegrationTestResult( ok=True, message=f"OK · {region_count} region(s) visible", detail={"regions_enabled": region_count}, ) # ── Scope sync ────────────────────────────────────────── async def list_scopes(self) -> list[CloudScopeInfo]: """UCloud has a single scope per tenant (standalone_iam). Projects (``GetProjectList``) exist but they're billing / quota partitions, not security boundaries — SubAccount policies attach at the tenant level. Treating the tenant as the one scope keeps the lifecycle service's branching consistent with Tencent. ``external_id`` is the **default project's** ``ProjectId`` — UCloud doesn't expose a tenant-level ``Uin`` equivalent, but every account ships with a default project whose id is stable for the tenancy and unique across UCloud. Falling back to the Helix integration UUID (the previous behaviour) was the same family of bug as the Tencent / AWS / Aliyun ghost-scope incident: rebuilding the integration row regenerated a different ``external_id`` and stranded every downstream grant. Resolution failures raise loudly so the admin sees the credential / API issue instead of silently re-rooting. """ try: data = await self._call("GetProjectList", retryable=True) except Exception as exc: # noqa: BLE001 raise RuntimeError( f"UCloud GetProjectList failed (cannot resolve canonical " f"ProjectId for scope): {exc}", ) from exc projects = data.get("ProjectSet") or data.get("Projects") or [] # Prefer the default project (IsDefault=Yes / true / "True"); if # none is flagged default, fall back to the first project in # creation order — UCloud's API returns them oldest-first. default_pid: str | None = None for proj in projects: flag = str(proj.get("IsDefault") or "").lower() if flag in ("yes", "true", "1"): default_pid = str(proj.get("ProjectId") or "") break if not default_pid and projects: default_pid = str(projects[0].get("ProjectId") or "") if not default_pid: raise RuntimeError( "UCloud GetProjectList returned no projects; refusing " "to fall back to integration UUID (would create a " "ghost scope on every integration rebuild).", ) alias = str(self.config.get("_alias") or f"ucloud-{default_pid}") return [CloudScopeInfo( scope_type="standalone_iam", external_id=default_pid, display_name=alias, status="active", raw={"mode": "iam", "default_project_id": default_pid}, )] # ── Role catalog ──────────────────────────────────────── async def list_scope_roles( self, scope: CloudScopeRef, ) -> list[CloudRoleInfo]: from app.integrations.drivers._cloud_defaults import load_default_roles defaults = await load_default_roles( "cloud.ucloud.default_policies", _UCLOUD_FALLBACK_POLICIES, ) out: list[CloudRoleInfo] = [] # Admin-curated built-in policies (platform_setting) for entry in defaults: key = str(entry.get("role_key") or "").strip() if not key: continue out.append(CloudRoleInfo( role_key=key, display_name=str(entry.get("display_name") or key), description=entry.get("description"), is_custom=False, source="ucloud_builtin", scope=scope, is_promoted=True, )) # Live tenant-custom policies via ``ListPolicies`` — ``Owner=User`` # returns admin-created custom policies; we omit ``Owner=UCloud`` # here because those are already covered by the ``defaults`` # block above (platform-settings-curated built-ins). offset = 0 limit = 100 while True: try: data = await self._call( "ListPolicies", {"Owner": "User", "Offset": offset, "Limit": limit}, retryable=True, ) except Exception as exc: # noqa: BLE001 # UCloud IAM is a paid add-on on some tenants. Missing # permission → surface the built-in defaults only and log. logger.info("UCloud ListPolicies(Owner=User) failed: %s", exc) break items = data.get("Policies") or data.get("DataSet") or [] for p in items: name = str(p.get("PolicyName") or "") if not name: continue if any(r.role_key == name for r in out): continue # dedupe against builtins out.append(CloudRoleInfo( role_key=name, display_name=name, description=str(p.get("Description") or "") or None, is_custom=True, source="ucloud_custom", scope=scope, raw={ "policy_urn": p.get("PolicyURN"), "policy_type": p.get("PolicyType"), }, is_promoted=True, )) total = int(data.get("TotalCount") or 0) offset += limit if offset >= total or not items: break return out # ── Grant / revoke ────────────────────────────────────── def _user_name_for(self, email: str) -> str: """Derive a deterministic UCloud IAM UserName from an email. UCloud IAM UserName allows letters, digits, hyphen, underscore up to 64 chars. Keep the local-part (lowercased) + sanitize. Falls back to ``helix-`` when the local-part is entirely non-alphanumeric (e.g. CJK-only addresses) — same pattern as AWS/Aliyun/Tencent drivers. Without the hash the fallback was the literal string ``"helix-user"``, causing two employees with non-ASCII local-parts to share one IAM user and collide on grant/revoke. """ import hashlib local = email.split("@", 1)[0].lower() sanitized = "".join( c for c in local if c.isascii() and (c.isalnum() or c in "-_") ) if not sanitized: # Compatibility identity, not secret protection: changing the # digest would orphan existing UCloud IAM revoke lookups. digest = hashlib.sha1( email.encode("utf-8"), usedforsecurity=False, ).hexdigest()[:12] return f"helix-{digest}" return sanitized[:64] async def grant_access( self, ctx: CloudAccessContext, ) -> CloudGrantResult: user_name = self._user_name_for(ctx.principal_email) policy_name = ctx.role_key # 1. Ensure IAM User exists (idempotent via ``AlreadyExist``) created_user = False try: resp = await self._call("CreateUser", { "UserName": user_name, "Email": ctx.principal_email, "DisplayName": ctx.principal_display_name or user_name, "Description": f"helix managed — {ctx.principal_email}", # ``Active`` = console login enabled; mirrors the UCloud # console default for programmatic grants. "LoginProfileStatus": "Active", "AccessKeyStatus": "Inactive", }) created_user = True sub_uid = str(resp.get("UserId") or "") except Exception as exc: # noqa: BLE001 msg = str(exc) if "AlreadyExist" in msg or "Existed" in msg or "UserExist" in msg: sub_uid = await self._resolve_user_id(user_name) if not sub_uid: return CloudGrantResult( ok=False, error_message=( "UCloud CreateUser reports existing user but " f"ListUsers can't find {user_name!r}" ), ) elif "InvalidEmail" in msg or ("Invalid" in msg and "Email" in msg): raise CloudPrincipalNotFoundError( f"UCloud rejected IAM user email {ctx.principal_email!r}: {msg}" ) from exc else: return CloudGrantResult(ok=False, error_message=msg) # 2. Resolve PolicyName → PolicyURN (AttachPoliciesToUser wants URN) policy_urn = await self._resolve_policy_urn(policy_name) if not policy_urn: return CloudGrantResult( ok=False, error_message=( f"UCloud: policy {policy_name!r} not found in this tenant" ), ) # 3. Attach policy (idempotent). UCloud IAM ``AttachPoliciesToUser`` # requires ``Scope`` with valid values ``"All"`` (tenant-wide) or # ``"Specified"`` (project-scoped, requires ``ProjectID``). Helix's # single standalone_iam scope model maps to tenant-wide → ``"All"``. # Earlier code passed ``"Account"`` — UCloud's current API rejects # that with RetCode=220 / 230 (verified 2026-05-14 prod). See # https://docs.ucloud.cn/api/iam-api/attach_policies_to_user. try: await self._call("AttachPoliciesToUser", { "UserName": user_name, "PolicyURNs.0": policy_urn, "Scope": "All", }) except Exception as exc: # noqa: BLE001 msg = str(exc) if "AlreadyAttached" in msg or "AlreadyExist" in msg or "Existed" in msg: return CloudGrantResult( ok=True, already_exists=True, external_ref=f"{user_name}::{policy_urn}::{policy_name}", detail={ "user_name": user_name, "user_id": sub_uid, "policy_urn": policy_urn, "policy_name": policy_name, "created": created_user, }, ) return CloudGrantResult(ok=False, error_message=msg) return CloudGrantResult( ok=True, external_ref=f"{user_name}::{policy_urn}::{policy_name}", detail={ "user_name": user_name, "user_id": sub_uid, "policy_urn": policy_urn, "policy_name": policy_name, "created": created_user, }, ) async def revoke_access( self, ctx: CloudAccessContext, ) -> CloudGrantResult: # external_ref shape produced by ``grant_access`` is # ``{user_name}::{policy_urn}::{policy_name}``. ``policy_urn`` # itself contains ``::`` (e.g. ``ucs:iam::ucs:policy/UBillFullAccess``), # so a naive ``split("::")[:3]`` truncates the URN to ``ucs:iam`` # and we send a malformed PolicyURN to UCloud — which surfaces # as RetCode=220 "Missing params [UserName or PolicyURNs or # Scope]" because UCloud's URN validator rejects the truncated # value and reports the input as absent. Reconstruct the URN # by taking everything between ``parts[1:-1]`` and rejoining. user_name: str policy_urn: str | None = None policy_name: str | None = None ref = ctx.existing_external_ref or "" if ref and "::" in ref: parts = ref.split("::") user_name = parts[0] if len(parts) >= 3: policy_name = parts[-1] or None urn = "::".join(parts[1:-1]).strip() if urn: policy_urn = urn elif len(parts) == 2 and parts[1]: # Legacy 2-part rows (pre-{user::urn::name} convention) — # parts[1] is the policy_name; URN resolved below. policy_name = parts[1] else: user_name = self._user_name_for(ctx.principal_email) if not policy_urn: policy_urn = await self._resolve_policy_urn( policy_name or ctx.role_key, ) if not policy_urn: # Policy gone from the tenant — the attachment is effectively # absent. Treat as success. return CloudGrantResult(ok=True, already_absent=True) try: # UCloud ``DetachPoliciesFromUser`` requires ``Scope`` with # valid values ``"All"`` (tenant-wide) or ``"Specified"`` # (project-scoped, requires ``ProjectID``). Per the docs: # https://docs.ucloud.cn/api/iam-api/detach_policies_from_user. # # Investigation log (Round-23 / P0092): # * Pre-fix sent malformed ``PolicyURN=ucs:iam`` (URN parse # bug above) with no Scope → RetCode=220 "Missing params # [UserName or PolicyURNs or Scope]". The error listed # three fields generically; only PolicyURNs was bad. # * Mid-day fix added ``Scope="Account"`` → RetCode=230 # "Params [Invalid Scope] not available" (verified prod). # Wrong value — the API expects ``"All"`` / ``"Specified"``, # not ``"Account"``. Earlier mistake assumed Attach's # value would match (legacy code used ``"Account"`` on # Attach too — also wrong per current API). # * Current fix: pass ``"All"`` consistent with the # standalone_iam tenant-wide scope model. Same value used # on Attach (corrected in this PR). await self._call("DetachPoliciesFromUser", { "UserName": user_name, "PolicyURNs.0": policy_urn, "Scope": "All", }) except Exception as exc: # noqa: BLE001 msg = str(exc) # UCloud surfaces "user already gone" through several error # codes — every one means the detach is a no-op and the # caller should treat as idempotent success. RetCode=11021 # "Account not exist" added 2026-06-01 after a prod incident # (chujie@ leaver, 3 UCloud grants sweep-retried every hour # because the IAM user had been admin-deleted out-of-band — # driver returned ok=False and ``sweep_pending_manual_cloud_ # revokes`` never cleared ``requires_manual_cloud_revoke``). if ( "NotExist" in msg or "NotAttached" in msg or "NoSuchPolicy" in msg or "UserNotExist" in msg or "RetCode=11021" in msg or "Account not exist" in msg ): return CloudGrantResult(ok=True, already_absent=True) return CloudGrantResult(ok=False, error_message=msg) return CloudGrantResult(ok=True, external_ref=ctx.existing_external_ref) async def _resolve_policy_urn(self, policy_name: str) -> str | None: """Look up PolicyURN for a given PolicyName. UCloud IAM ``ListPolicies`` requires ``Owner`` to disambiguate built-in (``UCloud``) vs tenant-custom (``User``) policies; we walk both. """ if not policy_name: return None for owner in ("UCloud", "User"): offset = 0 limit = 100 while True: try: data = await self._call("ListPolicies", { "Owner": owner, "Offset": offset, "Limit": limit, }, retryable=True) except Exception as exc: # noqa: BLE001 logger.warning( "UCloud ListPolicies(owner=%s) failed: %s", owner, exc, ) break items = data.get("Policies") or data.get("DataSet") or [] for p in items: if str(p.get("PolicyName") or "") == policy_name: urn = p.get("PolicyURN") if urn: return str(urn) total = int(data.get("TotalCount") or 0) offset += limit if offset >= total or not items: break return None async def _resolve_user_id(self, user_name: str) -> str: """Find the IAM user's UserId by UserName. Round-42 audit: a transient ``ListUsers`` failure used to return ``""``, which the caller interpreted as "user does not exist" and (in ``grant_access`` line 435-443) surfaced as an opaque "CreateUser reports existing but ListUsers can't find it" admin error. Raise instead so the caller gets the real cause (5xx/throttle) and can retry the whole flow safely. """ offset = 0 limit = 100 while True: try: data = await self._call("ListUsers", { "Offset": offset, "Limit": limit, }, retryable=True) except Exception as exc: # noqa: BLE001 logger.warning( "UCloud _resolve_user_id ListUsers failed — " "raising rather than mis-classifying as 'user " "missing': %s", exc, ) raise items = data.get("Users") or data.get("DataSet") or [] for u in items: if str(u.get("UserName") or "") == user_name: return str(u.get("UserId") or "") total = int(data.get("TotalCount") or 0) offset += limit if offset >= total or not items: return "" # ── Last login (ListUsers → LastLoginTime) ────────────── async def get_last_login( self, ctx: CloudAccessContext, ) -> datetime | None: user_name = self._user_name_for(ctx.principal_email) offset = 0 limit = 100 while True: try: data = await self._call("ListUsers", { "Offset": offset, "Limit": limit, }, retryable=True) except Exception as exc: # noqa: BLE001 # Last-login is a soft signal; the inactivity sweep # treats ``None`` as "unknown" and does NOT revoke # without a definite no-login result, so swallowing # here is safe (and preferable to aborting the whole # sweep on one user's transient throttle). Warn so a # persistent outage is operator-visible. logger.warning( "UCloud ListUsers(get_last_login) failed; " "returning None (sweep will skip, not revoke): %s", exc, ) return None items = data.get("Users") or data.get("DataSet") or [] for u in items: if str(u.get("UserName") or "") == user_name: ts = u.get("LastLoginTime") or u.get("LoginTime") if not ts: return None try: # tz-aware UTC: the inactivity sweep compares # against ``datetime.now(UTC)``; a naive return # would raise TypeError on the comparison and # crash the sweep (or, if silently coerced # downstream, drift by local-tz offset). return datetime.fromtimestamp(int(ts), tz=UTC) except (TypeError, ValueError, OSError): return None total = int(data.get("TotalCount") or 0) offset += limit if offset >= total or not items: return None # ── Reverse grants discovery ─────────────────────────── async def list_scope_grants( self, scope: CloudScopeRef, ) -> list[DiscoveredGrant]: """Enumerate every IAM user × attached policy in the tenant. Mass-revoke safety (Round-42 audit): a transient ``ListUsers`` / ``ListPoliciesForUser`` failure used to be swallowed (``return out`` / ``continue``). The grants-sync sweep treats the returned snapshot as authoritative — a partial result on a transient 5xx/throttle would interpret as "cloud removed every grant" and revoke every Helix-tracked UCloud grant. Raise instead so the sweep aborts with ``status=failed`` rather than committing a corrupting empty snapshot. """ out: list[DiscoveredGrant] = [] offset = 0 limit = 100 # Page ceiling — 500 pages × 100 = 50k users, above any realistic # tenant, while bounding a degenerate loop. _MAX_USER_PAGES = 500 users: list[dict[str, Any]] = [] pages_walked = 0 while True: pages_walked += 1 if pages_walked > _MAX_USER_PAGES: raise RuntimeError( "UCloud list_scope_grants: ListUsers hit " f"{_MAX_USER_PAGES}-page ceiling at offset={offset} — " "refusing to loop or write a partial snapshot.", ) try: data = await self._call("ListUsers", { "Offset": offset, "Limit": limit, }, retryable=True) except Exception as exc: # noqa: BLE001 logger.warning( "UCloud list_scope_grants: ListUsers failed at " "offset=%d — refusing to write partial snapshot " "(would mass-revoke against drift): %s", offset, exc, ) raise items = data.get("Users") or data.get("DataSet") or [] users.extend(items) total = int(data.get("TotalCount") or 0) offset += limit # Terminate on a SHORT page (fewer than a full ``limit`` returned) # rather than on ``offset >= total``. A degenerate/absent # ``TotalCount`` of 0 with a FULL first page would previously break # after page 1, truncating the roster — the grants-sweep then reads # the tail as "cloud removed those grants" and mass-revokes. Only # trust ``offset >= total`` when the API actually reported a # positive total. if not items or len(items) < limit: break if total > 0 and offset >= total: break for u in users: user_name = str(u.get("UserName") or "") email = str(u.get("Email") or u.get("UserEmail") or "").strip().lower() if not email and "@" in user_name: email = user_name.lower() if not email or not user_name: continue uid = str(u.get("UserId") or "") # List attached policies for this user. UCloud's # ``ListPoliciesForUser`` IS paginated (``Offset`` / ``Limit`` / # ``TotalCount``) — calling it once without paging silently # truncates the tail when a user holds more policies than the # default page size. The grants-sync sweep then treats the # truncated tail as "policies removed cloud-side" and the # next drift sweep mass-revokes them. # Same defensive shape as ``list_scope_roles`` / # ``_resolve_policy_urn`` / ``_resolve_user_id`` above. A # transient mid-pagination failure raises so the whole # snapshot is discarded rather than written partial. policies: list[dict[str, Any]] = [] p_offset = 0 p_limit = 100 _MAX_POLICY_PAGES = 200 p_pages_walked = 0 while True: p_pages_walked += 1 if p_pages_walked > _MAX_POLICY_PAGES: raise RuntimeError( f"UCloud ListPoliciesForUser({user_name}) hit " f"{_MAX_POLICY_PAGES}-page ceiling at offset=" f"{p_offset} — refusing to loop or truncate.", ) try: p_data = await self._call("ListPoliciesForUser", { "UserName": user_name, "Offset": p_offset, "Limit": p_limit, }, retryable=True) except Exception as exc: # noqa: BLE001 logger.warning( "UCloud ListPoliciesForUser(%s) failed at " "offset=%d — refusing to drop this user from " "the snapshot (would mass-revoke their grants): %s", user_name, p_offset, exc, ) raise page = p_data.get("Policies") or p_data.get("DataSet") or [] policies.extend(page) p_total = int(p_data.get("TotalCount") or 0) p_offset += p_limit # Same short-page termination as the user loop: a full page # with an absent/zero TotalCount must keep paging, else the # policy tail is silently truncated and drift-swept. if not page or len(page) < p_limit: break if p_total > 0 and p_offset >= p_total: break for p in policies: pname = str(p.get("PolicyName") or "") if not pname: continue purn = str(p.get("PolicyURN") or "") out.append(DiscoveredGrant( principal_email=email, principal_type="user", role_key=pname, role_display_name=pname, external_ref=f"{user_name}::{purn}::{pname}", scope=scope, granted_at=None, raw={ "user_name": user_name, "user_id": uid, "policy_urn": purn, "policy_name": pname, }, )) return out # ── Account revocation ───────────────────────────────── async def revoke_account( self, ctx: CloudAccountRevokeContext, ) -> ProvisioningResult: """Remove the UCloud IAM user cleanly. Deletion order: 1. Detach every policy from the user 2. Delete all access keys (``DeleteAccessKey`` keyed by ``AccessKeyID``) 3. Delete the user (``DeleteUser``) Each step swallows NotExist so a partially-cleaned user still succeeds on retry. """ raw_id = (ctx.account_id or "").strip() if not raw_id: return ProvisioningResult( ok=False, error_message="UCloud revoke_account requires account_id (IAM UserName or email)", ) # Literal handle first — see ``account_id_candidates``. Deriving # the local part unconditionally turns this whole chain into a # silent no-op whenever the IAM user is genuinely named after an # address (prod Tencent: 17 of 22 users, every revoke reporting # ok=True while the account kept AdministratorAccess). candidates = account_id_candidates(raw_id, self._user_name_for) user_name = "" p_data: dict[str, Any] = {} probe_error: str | None = None for candidate in candidates: try: p_data = await self._call("ListPoliciesForUser", { "UserName": candidate, }, retryable=True) user_name = candidate break except Exception as exc: # noqa: BLE001 msg = str(exc) if "UserNotExist" in msg or "NotExist" in msg: continue probe_error = msg 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(label: str, action: str, params: dict[str, Any]) -> bool: """Run a step of the deletion chain. Returns True on success, False otherwise. Non-NSE errors are logged but NOT re-raised — that way a single transient failure in step N doesn't prevent step N+1 from executing (each step is idempotent, so the retry sweep can finish whatever this run missed). Mirrors Tencent's ``_revoke_account_sync`` resilience pattern. """ try: await self._call(action, params) steps.append(label) return True except Exception as exc: # noqa: BLE001 msg = str(exc) if ( "NotExist" in msg or "NotAttached" in msg or "NoSuchPolicy" in msg or "UserNotExist" in msg ): return False logger.warning( "UCloud revoke_account step %s failed (will continue): %s", label, exc, ) return False # 1. Detach policies — ``p_data`` came from the resolution probe # above, so the user is known to exist and the list is already in # hand (one call, not two). for p in (p_data.get("Policies") or p_data.get("DataSet") or []): purn = str(p.get("PolicyURN") or "") if not purn: continue # ``Scope="All"`` is mandatory and the only legal value for # the tenant-wide attachments Helix manages. See # ``revoke_access`` above for full rationale. await _swallow( f"detach_policy:{p.get('PolicyName') or purn}", "DetachPoliciesFromUser", { "UserName": user_name, "PolicyURNs.0": purn, "Scope": "All", }, ) # 2. Delete access keys (ListAccessKeys + DeleteAccessKey) try: k_data = await self._call("ListAccessKeys", {"UserName": user_name}, retryable=True) except Exception as exc: # noqa: BLE001 logger.info("UCloud ListAccessKeys(%s): %s", user_name, exc) k_data = {} for k in (k_data.get("AccessKeys") or k_data.get("DataSet") or []): ak_id = str(k.get("AccessKeyID") or k.get("AccessKeyId") or "") if not ak_id: continue # UCloud IAM DeleteAccessKey requires BOTH UserName and # AccessKeyID — without UserName the API rejects with # ``InvalidArgument``. await _swallow( f"delete_access_key:{ak_id[:12]}", "DeleteAccessKey", {"UserName": user_name, "AccessKeyID": ak_id}, ) # 3. Delete the IAM user try: await self._call("DeleteUser", {"UserName": user_name}) steps.append("delete_user") except Exception as exc: # noqa: BLE001 msg = str(exc) if "UserNotExist" in msg or "NotExist" in msg: return ProvisioningResult( ok=True, external_ref=user_name, raw_response={"note": "user already absent", "steps": steps}, ) return ProvisioningResult( ok=False, error_message=f"UCloud 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. UCloud is single-tenant IAM (no multi-account organization API). For ``standalone_iam`` "real closure" means **evacuate every Helix-tracked IAM user on the scope** by running the same multi-step revoke chain that offboarding uses. The cloud account itself is not closed (UCloud requires the account owner to do that via the billing console). Anything else raises ``NotImplementedError``. """ if scope.scope_type != "standalone_iam": raise NotImplementedError( f"UCloud driver cannot close scope_type={scope.scope_type!r}", ) return await self._evacuate_iam_users(iam_account_ids or []) async def _evacuate_iam_users( self, iam_account_ids: list[str], ) -> CloseScopeResult: """Run the per-user IAM 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 IAM 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: 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"UCloud evacuation partial: {len(failures)} of " f"{len(iam_account_ids)} IAM 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}, ) # ── Daily billing (ListUBillDetail per-month aggregation) ─── 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]: """Daily cost breakdown via ``ListUBillDetail`` aggregation. UCloud exposes per-resource itemized bills through ``ListUBillDetail`` keyed by ``BillingCycle=YYYY-MM``. Each row carries a ``PayTime`` (day-granular) and ``TotalPayPrice`` in CNY. We fetch each month that the date range intersects, then aggregate client-side by ``(day, BusinessName)``. Currency is CNY (same decision as Aliyun + Tencent — the ``CloudDailyCost.total_cost_usd`` field name is a legacy misnomer; ``currency`` is authoritative). """ del bq_billing # AWS-driver-specific kwarg; accept and ignore # Build the set of YYYY-MM cycles to fetch cycles: list[str] = [] anchor = start_date.replace(day=1) while anchor <= end_date: cycles.append(anchor.strftime("%Y-%m")) if anchor.month == 12: anchor = anchor.replace(year=anchor.year + 1, month=1) else: anchor = anchor.replace(month=anchor.month + 1) # ``by_day[day] = (total, {BusinessName: cost})`` by_day: dict[date, tuple[float, dict[str, float]]] = {} for cycle in cycles: offset = 0 limit = 100 # ListUBillDetail max # 500 pages × 100 rows = 50,000 rows/cycle. Real tenants # we probed sit around 5–8k rows/month, so this has ~6× # headroom. Hit = logged warning (see below). max_pages = 500 hit_cap = True for _ in range(max_pages): try: data = await self._call("ListUBillDetail", { "BillingCycle": cycle, "Offset": offset, "Limit": limit, }, retryable=True, project_id=None) # account-wide bill except Exception as exc: # noqa: BLE001 msg = str(exc) # Split fail-open list: # * ``UnknownAction`` / ``NotOpen`` are billing-API- # not-enabled signals and are safely fail-open # ("the tenant has no daily-billing data because # they didn't enable the feature"). # * ``NoAuth`` / ``Permission`` are real auth / # permission failures: the credential is broken # or the role lost the billing scope. Fail-open # here silently advances ``last_seen_billing_daily_at`` # on the scope, which feeds stale 7-day baselines # into spike-detection and silences budget alerts # while the credential is dead. Mirror Aliyun's # ``_acs3_call`` posture and re-raise so the # lifecycle service surfaces the per-scope failure. if any(x in msg for x in ("UnknownAction", "NotOpen")): logger.info( "UCloud ListUBillDetail not enabled: %s", exc, ) return self._emit_rows(by_day, start_date, end_date, scope) # Mid-pagination 5xx / throttle / transient / auth: # re-raise so the lifecycle service records the # per-scope failure and the next cron tick re-ingests # the whole window — partial data is a harder bug # class to spot than an explicit failure. raise RuntimeError( f"UCloud ListUBillDetail cycle={cycle} " f"offset={offset} failed: {exc}", ) from exc # Response key is ``Items`` (confirmed against live API # 2026-04-24); fall back to ``UBillDataSet`` / ``DataSet`` # in case UCloud renames it in a future version. items = ( data.get("Items") or data.get("UBillDataSet") or data.get("DataSet") or [] ) for row in items: # ``StartTime`` is Unix-epoch seconds of the usage # start. ``CreateTime`` fires when the bill row was # materialized (typically same day but not always); # prefer StartTime so charges land on the day the # resource actually ran. ts = row.get("StartTime") or row.get("CreateTime") or row.get("PayTime") if not ts: continue try: if isinstance(ts, (int, float)) or str(ts).isdigit(): # UCloud bills in CST. ``.date()`` against a # naive UTC-converted timestamp shifts every # CST 00:00–08:00 row into the previous day; # daily totals on the billing chart drift by # 8h of charges. Anchor to Asia/Shanghai so # the day bucket matches the billing TZ. from zoneinfo import ZoneInfo # noqa: PLC0415 day = datetime.fromtimestamp( int(ts), tz=ZoneInfo("Asia/Shanghai"), ).date() else: day = datetime.strptime( str(ts)[:10], "%Y-%m-%d", ).date() except (ValueError, OSError): continue if day < start_date or day > end_date: continue # ``Amount`` is the gross CNY cost; ``AmountReal`` is # post-discount. We use AmountReal when available so # the daily rollup matches what UCloud actually bills # (coupons already netted out). Presence-based via # ``_first_present`` — a numeric ``AmountReal=0`` # (fully-couponed line) must stay 0, not fall # through to the gross ``Amount``. amt_s = _first_present( row, ("AmountReal", "Amount", "TotalPayPrice"), ) try: amt = float(amt_s) if amt_s is not None else 0.0 except (TypeError, ValueError): continue # Service name: prefer the human-readable # ``ItemDetails[0].ProductName`` when present (e.g. # "快照服务套餐1"), fall back to the resource-type # code (e.g. "USNAPSHOT"). item_details = row.get("ItemDetails") or [] svc = "" if isinstance(item_details, list) and item_details: first = item_details[0] if isinstance(first, dict): svc = str(first.get("ProductName") or "") if not svc: # Fallback string must match # ``list_cost_facts`` (``"Other"``) so the # same upstream row produces the same # service-bucket on both billing_daily and # cost_facts. Round-41 align. svc = str( row.get("ResourceType") or row.get("ProductName") or row.get("BusinessName") or "Other", ) total, services = by_day.get(day, (0.0, {})) total += amt services[svc] = services.get(svc, 0.0) + amt by_day[day] = (total, services) if len(items) < limit: hit_cap = False break offset += limit if hit_cap: # Raise instead of warn: a partial daily snapshot has no # self-correction path — the cron records status=ok with # wrong numbers, silently undercounting billing totals. # Matches list_cost_facts behaviour (raises on cap hit). raise RuntimeError( f"UCloud ListUBillDetail hit {max_pages}-page cap at " f"offset={offset} for cycle={cycle} — billing data is " f"partial. Raise max_pages in the driver config if the " f"account has more than {max_pages * limit} line items " f"per billing cycle.", ) return self._emit_rows(by_day, start_date, end_date, scope) @staticmethod def _emit_rows( by_day: dict[date, tuple[float, dict[str, float]]], start_date: date, end_date: date, scope: CloudScopeRef | None, ) -> list[CloudDailyCost]: out: list[CloudDailyCost] = [] for day, (total, services) in sorted(by_day.items()): if day < start_date or day > end_date: continue out.append(CloudDailyCost( date=day, scope=scope, total_cost_usd=total, # legacy field name; CNY, see currency currency="CNY", cost_by_service=services, )) return out # ── Line-item cost facts (ListUBillDetail per-row) ────────── 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]: """Per-resource line items via ``ListUBillDetail`` aggregation. Same API call as :meth:`list_billing_daily` (UCloud doesn't have a separate line-item endpoint — ``ListUBillDetail`` IS the line item feed) but emits ``CostFact`` rows preserving the per-resource dimensions (resource_id, resource_type, region, service) instead of aggregating to (day, BusinessName). **Aggregation within (day × resource_id × usage_type)** is done client-side BEFORE returning to the ingestion service. The cost-facts upsert is keyed on ``sha256(date|integration|scope|resource_id|usage_type)`` with ``ON DUPLICATE KEY UPDATE`` — colliding rows would OVERWRITE rather than SUM, silently dropping money. UCloud's ``ListUBillDetail`` can return multiple rows for the same (resource, day, charge type) when add-on items are billed separately, so we pre-sum at the natural-key granularity. **Reconciliation with ``list_billing_daily``**: both use the same ``AmountReal`` → ``Amount`` → ``TotalPayPrice`` priority and the same Asia/Shanghai day bucketing, so daily rollup CNY totals and ``cost_facts`` CNY aggregates reconcile exactly for the same window. ``amount_usd`` is computed by the driver per the ``CostFact`` contract (drivers must convert; the ingestor does not). ``principal_tag_keys`` is AWS-specific (IAM principal tags on Cost Explorer); UCloud has no equivalent dimension. Accept and ignore. ``bq_billing`` likewise AWS-driver-specific. """ del principal_tag_keys, bq_billing # FX guard — mirrors Tencent / Aliyun (CNY-native peers). The # ingestion service must resolve a rate before calling because # the driver is DB-free; ``None`` / ≤0 here is a service-layer # bug, not a transient — raise hard rather than silently land # ``amount_usd=0`` rows that would understate UCloud spend in # every downstream USD aggregate until an admin noticed. if cny_per_usd is None or cny_per_usd <= 0: raise RuntimeError( f"UCloud 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.", ) # Build the set of YYYY-MM cycles intersecting the window — # same cycle walk as ``list_billing_daily`` so the two paths # cover identical periods. cycles: list[str] = [] anchor = start_date.replace(day=1) while anchor <= end_date: cycles.append(anchor.strftime("%Y-%m")) if anchor.month == 12: anchor = anchor.replace(year=anchor.year + 1, month=1) else: anchor = anchor.replace(month=anchor.month + 1) # Aggregate to the cost_facts natural key BEFORE emitting. # bucket key = (day, resource_id, usage_type) # bucket value = dict carrying the summed amount + the # first-seen non-money dimensions (service / region / etc.). # Same-key rows are pre-summed; once aggregated, each bucket # emits exactly one CostFact and ``ON DUPLICATE KEY UPDATE`` # in the ingestor becomes a no-op refresh, not an overwrite- # drops-money collision. buckets: dict[ tuple[date, str | None, str | None], dict[str, Any], ] = {} for cycle in cycles: offset = 0 limit = 100 # ListUBillDetail max # Same headroom as ``list_billing_daily``. Hitting the cap # in cost_facts is treated as HARDER than there — a # silently-truncated fact ingest would land in the DB # tagged as a "complete" day's bill and every per-resource # chargeback would understate the truncated tenant's # spend. Mirrors ``Tencent._list_cost_facts_sync``. max_pages = 500 hit_cap = True for _ in range(max_pages): try: data = await self._call("ListUBillDetail", { "BillingCycle": cycle, "Offset": offset, "Limit": limit, }, retryable=True, project_id=None) # account-wide bill except Exception as exc: # noqa: BLE001 msg = str(exc) # Same fail-open split as ``list_billing_daily``: # ``UnknownAction`` / ``NotOpen`` mean the billing # add-on isn't enabled on this tenant → return # what we've accumulated so far (likely empty). # The cron records ``status='ok', facts=0``; the # tenant genuinely has no line items to ingest. if any(x in msg for x in ("UnknownAction", "NotOpen")): logger.info( "UCloud ListUBillDetail not enabled " "(cost_facts path): %s", redact_provider_error(exc), ) return self._emit_cost_facts( buckets, start_date, end_date, scope, float(cny_per_usd), ) # Mid-pagination 5xx / throttle / auth: re-raise so # the lifecycle service records the per-scope # failure and the next cron tick re-ingests the # whole window. Partial data is a harder bug class # to spot than an explicit failure. The exception # text is run through ``redact_provider_error`` to # strip any UCloud AK / signature / proxy-WAF # echo before it lands in ``cmdb_sync_jobs. # capability_results['error']`` (DB-readable by # every cloud_admin), the failure-notifier card, # and the 30+ day log retention. raise RuntimeError( f"UCloud ListUBillDetail (cost_facts) " f"cycle={cycle} offset={offset} failed: " f"{redact_provider_error(exc)}", ) from exc # Response key parity with ``list_billing_daily``. items = ( data.get("Items") or data.get("UBillDataSet") or data.get("DataSet") or [] ) for row in items: # Day bucket — Asia/Shanghai TZ to match the # billing-side calendar (UCloud bills in CST). # Same logic as ``list_billing_daily``. ts = ( row.get("StartTime") or row.get("CreateTime") or row.get("PayTime") ) if not ts: continue try: if isinstance(ts, (int, float)) or str(ts).isdigit(): from zoneinfo import ZoneInfo # noqa: PLC0415 day = datetime.fromtimestamp( int(ts), tz=ZoneInfo("Asia/Shanghai"), ).date() else: day = datetime.strptime( str(ts)[:10], "%Y-%m-%d", ).date() except (ValueError, OSError): continue if day < start_date or day > end_date: continue # ``AmountReal`` (post-discount) when present so # the per-resource USD figure aligns with what # UCloud actually billed. Identical priority AND # presence semantics (``_first_present``) as # ``list_billing_daily`` — both paths must sum to # the same CNY total for the window, and a numeric # ``AmountReal=0`` must not fall through to the # gross ``Amount``. amt_s = _first_present( row, ("AmountReal", "Amount", "TotalPayPrice"), ) try: native = float(amt_s) if amt_s is not None else 0.0 except (TypeError, ValueError): continue # Defend against poisoned upstream — ``float("nan")`` # / ``float("inf")`` parse without raising, but # propagate NaN/Inf into ``amount_usd`` which then # corrupts every downstream ``SUM()`` aggregate # (NaN + anything = NaN). UCloud is a trusted # upstream behind TLS + signature, but the guard # is cheap and the failure mode is silent so # belt-and-braces it. if not math.isfinite(native): logger.warning( "UCloud ListUBillDetail (cost_facts): " "dropping row with non-finite amount %r " "for resource %r in cycle %s", amt_s, row.get("ResourceId"), cycle, ) continue if native == 0: # Zero-amount rows are noise (placeholder / # 0-yuan refunds), and they'd pollute the # bucket with empty dimensions for no money. # Mirrors Tencent ``_list_cost_facts_sync``. continue # Service name: prefer human-readable # ``ItemDetails[0].ProductName`` then the resource # type code, matching ``list_billing_daily`` so # the service breakdown reconciles across the two # tables. item_details = row.get("ItemDetails") or [] svc = "" if isinstance(item_details, list) and item_details: first = item_details[0] if isinstance(first, dict): svc = str(first.get("ProductName") or "") if not svc: svc = str( row.get("ResourceType") or row.get("ProductName") or row.get("BusinessName") or "Other", ) resource_id = ( str(row.get("ResourceId") or "") or None ) resource_type = ( str(row.get("ResourceType") or "") or None ) # UCloud's bill rows carry NO top-level ``Region`` # (always null); the region lives in ``AzGroupCName`` # as a CN display name ("华北（北京2）", "香港"). It's # the only region signal the bill exposes, so use it # (better a human-readable AZ-group than a NULL # "cost by region" dimension). Prefer a real # ``Region`` if a future API revision ever adds one. region = ( str(row.get("Region") or row.get("AzGroupCName") or "") or None ) # ``ChargeType`` (年付/月付/按需/试用/...) is UCloud's # SKU-equivalent dimension. Using it as usage_type # keeps prepaid vs pay-as-you-go on the same # resource separate (different fact rows) while # allowing same-charge-type rows to aggregate. usage_type = ( str(row.get("ChargeType") or "") or None ) resource_group = ( str(row.get("ProjectName") or "") or None ) # Friendly resource name: bill rows leave # ``ResourceName`` null but stash the console name in # ``ResourceExtendInfo`` as ``{KeyId:"name", Value}``. # Recover it so the per-resource "小账" shows names, # not bare IDs; fall back to id when absent. ext_name = "" ext = row.get("ResourceExtendInfo") or [] if isinstance(ext, list): for kv in ext: if isinstance(kv, dict) and kv.get("KeyId") == "name": ext_name = str(kv.get("Value") or "") break key = (day, resource_id, usage_type) bucket = buckets.get(key) if bucket is None: buckets[key] = { "native": native, "service": svc, "resource_type": resource_type, "region": region, "resource_group": resource_group, "resource_name": ( str(row.get("ResourceName") or "") or ext_name or resource_id ), "raw": { "BillingCycle": cycle, "ResourceType": str( row.get("ResourceType") or "", ), "BusinessName": str( row.get("BusinessName") or "", ), "ChargeType": str( row.get("ChargeType") or "", ), "PayTime": str(row.get("PayTime") or ""), }, } else: bucket["native"] += native if len(items) < limit: hit_cap = False break offset += limit if hit_cap: # Stricter than ``list_billing_daily`` (which only # warns): a truncated cost_facts ingest plants a # plausible-but-incomplete fact set in the DB, and # per-resource chargeback would silently understate # the truncated tenant's spend until an operator # noticed. Same posture as Tencent. ``cycle`` and # ``offset`` are driver-generated (no untrusted # input) so this message is safe without redaction, # but kept the same length/format as other surfaces # for uniform parsing downstream. raise RuntimeError( f"UCloud ListUBillDetail (cost_facts) hit " f"{max_pages}-page cap at offset={offset} for " f"cycle={cycle} — ingestion would be incomplete. " "Split the window or raise max_pages.", ) return self._emit_cost_facts( buckets, start_date, end_date, scope, float(cny_per_usd), ) def _emit_cost_facts( self, buckets: dict[tuple[date, str | None, str | None], dict[str, Any]], start_date: date, end_date: date, scope: CloudScopeRef | None, fx_rate: float, ) -> list[CostFact]: out: list[CostFact] = [] integration_id = self.integration_id for (day, resource_id, usage_type), bucket in sorted( buckets.items(), key=lambda kv: (kv[0][0], kv[0][1] or "", kv[0][2] or ""), ): if day < start_date or day > end_date: continue native = float(bucket["native"]) if native == 0: continue out.append(CostFact( date=day, integration_id=integration_id, service=str(bucket["service"]) or "Other", amount_usd=native / fx_rate, amount_native=native, currency="CNY", scope=scope, resource_group=bucket.get("resource_group"), resource_id=resource_id, resource_name=bucket.get("resource_name") or resource_id, resource_type=bucket.get("resource_type"), region=bucket.get("region"), usage_type=usage_type, tags={}, # UCloud bill has no per-row tag dimension raw=dict(bucket.get("raw") or {}), )) return out # Silence unused-import warnings for optional / re-exported symbols. _ = (CloudPrincipalNotFoundError,)