"""Aliyun (Alibaba Cloud) provider — ECS servers + Domain registration."""

from __future__ import annotations

import asyncio
import base64
import hashlib
import hmac
import json
import logging
import re
import urllib.parse
import uuid
import xml.etree.ElementTree as ET
from datetime import UTC, date, datetime, timedelta
from email.utils import formatdate

import httpx

# defusedxml hardens the standard parser against billion-laughs / external
# entity attacks. We still expose ``ET.Element`` for type hints, but every
# call that parses untrusted bytes (e.g. an OSS XML response that comes
# from an attacker-controllable network path) MUST go through
# ``defusedxml.ElementTree.fromstring`` rather than ``ET.fromstring``.
from defusedxml.ElementTree import fromstring as _safe_fromstring

from app.core.exceptions import format_exception_for_log
from app.domains.cmdb.providers.base import (
    CloudCdnDomain,
    CloudCertificate,
    CloudDnsRecord,
    CloudDomain,
    CloudLoadBalancer,
    CloudProvider,
    CloudServer,
    ProviderCapability,
    ProviderCredentials,
    ServerMetricPoint,
    ServerMetricTarget,
    register_provider,
)
from app.domains.cmdb.providers.exceptions import (
    CapabilityNotSupported,
    ProviderAuthError,
    ProviderRateLimitError,
    ProviderTransientError,
)

logger = logging.getLogger(__name__)

# Safety ceiling on paginated list loops. Each loop normally terminates on an
# empty batch, but if an Aliyun API version silently ignores the page
# parameter (a documented cross-version quirk) it would return a full page
# forever and accumulate unbounded until the sync timeout / worker OOM. With
# PageSize 50–500 this cap bounds a single list at ~0.5–2M items — far beyond
# any real tenant — while stopping a runaway loop. Hitting it is logged.
_MAX_LIST_PAGES = 5000

# Aliyun error ``Code`` values that mean "this credential is technically
# valid but the requested service is not subscribed / activated for the
# account" — i.e. the credential itself is fine; there's just nothing
# to enumerate. The gateway often returns these as HTTP 403 (because
# the AK has no permission against an unsubscribed service), which the
# old codepath misclassified as ``ProviderAuthError`` and turned into a
# red ``cmdb.resource_sync ... auth error: ...`` line in the scheduled-
# jobs history (user report 2026-05-05, AWS SG / Aliyun CDN).
#
# Mapping these to ``CapabilityNotSupported`` makes the per-capability
# sync result land in the ``skipped`` bucket — historical rows survive
# untouched, no false alert, no red status. Each entry is the verbatim
# Aliyun ``Code`` string returned in the JSON body. Keep this list
# narrow and explicit (not a regex) so a real auth fault never gets
# silently swallowed as "service not subscribed".
#
# Trust model: classification reads the ``Code`` field out of the
# response body. Helix only ever talks to Aliyun over TLS pinned at
# Aliyun's PKI, so an active MITM able to forge a ``CdnServiceNotFound``
# body — turning a real auth failure into a "skipped" capability — is
# out of scope for our threat model. If TLS to the Aliyun gateway can
# be tampered with, far worse than capability-skipping is on the table.
# Documented for completeness in cloud-lifecycle-ops.md.
#
# Adding new codes: each new entry MUST come from a verbatim Code
# observed in a real production failure (not from Aliyun docs / SDK
# enums / blog posts). The shorter this list is, the less surface a
# real auth fault has to be silently swallowed. Currently only the
# CDN family has hit prod — start there.
ALIYUN_SERVICE_NOT_SUBSCRIBED_CODES: frozenset[str] = frozenset({
    # CDN — DescribeUserDomains on an account that never opened CDN.
    # Reported 2026-05-05 by AWS SG / Aliyun CDN scheduled-job audit.
    "CdnServiceNotFound",
    # CDN's older variant in some regions; same root cause as the
    # above — kept because the gateway returns either form across
    # cn-* and intl regions.
    "CdnService.NotFound",
})


def _extract_aliyun_error_code(resp: httpx.Response) -> str | None:
    """Pull the ``Code`` field out of an Aliyun error response.

    Aliyun's REST APIs return JSON like
    ``{"RequestId": "...", "HostId": "...", "Code": "<error_code>",
    "Message": "..."}`` on failure (some services use 4xx, some 200
    + body). Returns ``None`` if the body isn't JSON or has no Code.
    Never raises — error-classification must not itself fault.
    """
    try:
        body = resp.json()
    except ValueError:
        # ``json.JSONDecodeError`` is a ValueError subclass; any
        # non-JSON body (HTML 502 from a misrouted gateway, plain-text
        # rate-limit page) lands here. Return None so the caller falls
        # through to the standard ProviderAuthError / raise_for_status
        # paths.
        return None
    if not isinstance(body, dict):
        return None
    code = body.get("Code")
    return code if isinstance(code, str) else None

# Aliyun ECS regions to sync (extend as needed)
ECS_REGIONS = [
    "cn-hangzhou", "cn-shanghai", "cn-beijing", "cn-shenzhen",
    "cn-hongkong", "ap-southeast-1", "ap-northeast-1", "us-east-1",
]

# Fixed FX rates used to convert native currency to USD for the
# headline ``total_cost_usd`` field consumed by the billing handler.
# Native amount + currency are still surfaced via ``total_cost_native``
# / ``currency``. Rates are static; rotate in code if they drift >5%.
_FX_TO_USD = {
    "USD": 1.0,
    "CNY": 0.14,
    "EUR": 1.08,
}


def _aliyun_sign(params: dict, secret: str) -> str:
    """Compute Aliyun RPC HMAC-SHA1 signature."""
    sorted_params = sorted(params.items())
    query = urllib.parse.urlencode(
        [(k, v) for k, v in sorted_params],
        quote_via=urllib.parse.quote,
    )
    string_to_sign = "GET&%2F&" + urllib.parse.quote(query, safe="")
    key = (secret + "&").encode()
    sig = hmac.new(key, string_to_sign.encode(), hashlib.sha1).digest()
    return base64.b64encode(sig).decode()


def _redact_aliyun_error(text: str | None) -> str:
    """Strip ``AccessKeyId`` / ``Signature`` query params + truncate.

    Aliyun error responses sometimes include the failed request URL —
    which carries the signed query string with ``AccessKeyId=…`` and
    ``Signature=…``. Stack traces and ``logger.warning("...: %s", exc)``
    callsites in sync.py would otherwise leak the AK in plaintext into
    log aggregation. AK alone isn't a full credential but it's PII-class
    data and an attacker with log access can correlate it with a leaked
    SK from elsewhere.

    OSS-specific: when a sub-account ``list_object_storage`` call fails
    with ``SignatureDoesNotMatch``, OSS echoes the server-computed
    ``<StringToSign>`` element in the XML body, which (under STS
    credentials) includes ``x-oss-security-token:<full-token>\\n``. The
    token is the bearer credential — leaking it into a sync_jobs row /
    Feishu alert is a full-credential exposure for the AssumeRole
    DurationSeconds window (default 1h, max 12h per Aliyun STS limits).
    """
    if not text:
        return ""
    import re as _re  # noqa: PLC0415
    # Aliyun RPC query-style: ``AccessKeyId=AKID&Signature=SIG&SecurityToken=TOK``
    redacted = _re.sub(
        r"(AccessKeyId|Signature|SecurityToken)=[^&\s\"']+",
        r"\1=<redacted>",
        text,
    )
    # OSS error-body StringToSign echo includes canonicalized headers
    # ``x-oss-security-token:<token>`` followed by ``\n`` (newline) or
    # XML element boundary. Strip the value up to the next whitespace,
    # ``<``, or ``&`` (XML / query-string boundary chars) — these are
    # the anchoring chars that terminate a canonical header line in
    # either the raw echo, XML-escaped (``&#10;``), or query-style
    # (``&Signature=…``) forms. ``\s`` already covers newline/CR/tab/
    # space — no need to enumerate. Including ``&`` here means a body
    # like ``...token:FOO&Signature=BAR`` cleanly redacts both halves
    # in this single pass + the SecurityToken= pass above.
    redacted = _re.sub(
        r"(x-oss-security-token:)[^\s<&]+",
        r"\1<redacted>",
        redacted,
        flags=_re.IGNORECASE,
    )
    return redacted[:300]


def _cas_date(value: str | None) -> datetime | None:
    """Parse an Aliyun CAS cert date into a tz-aware UTC datetime.
    CAS returns ``StartDate`` / ``EndDate`` as ``YYYY-MM-DD`` (some
    tenants return a full timestamp); be tolerant of both."""
    if not value:
        return None
    s = str(value).strip()
    if not s:
        return None
    # Some CAS responses return epoch milliseconds (13 digits) or seconds
    # (10 digits) as a string/number instead of a formatted date.
    if s.isdigit():
        ts = int(s)
        if ts > 10_000_000_000:  # ms
            ts //= 1000
        try:
            return datetime.fromtimestamp(ts, tz=UTC)
        except (ValueError, OverflowError, OSError):
            return None
    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d"):
        try:
            return datetime.strptime(s, fmt).replace(tzinfo=UTC)
        except ValueError:
            continue
    return None


def _common_params(action: str, ak: str) -> dict:
    return {
        "Format": "JSON",
        "Version": "2014-05-26",
        "AccessKeyId": ak,
        "SignatureMethod": "HMAC-SHA1",
        "Timestamp": datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ"),
        "SignatureVersion": "1.0",
        "SignatureNonce": uuid.uuid4().hex,
        "Action": action,
    }


# Transport-layer exceptions httpx raises BEFORE a status code exists.
# OSS list_buckets ran without these guards pre-fix (the RPC ``_call``
# path on line ~286 already wrapped them) — a single ReadTimeout or
# ConnectError mid-pagination would propagate as bare ``httpx.HTTPError``
# and the entire OSS leg of a sync tick was marked failed.
_OSS_TRANSPORT_RETRY_EXCEPTIONS: tuple[type[BaseException], ...] = (
    httpx.ConnectError,
    httpx.ConnectTimeout,
    httpx.ReadError,
    httpx.ReadTimeout,
    httpx.WriteError,
    httpx.WriteTimeout,
    httpx.PoolTimeout,
    httpx.RemoteProtocolError,
)


async def _oss_get_with_transport_retry(
    client: httpx.AsyncClient,
    url: str,
    *,
    endpoint_label: str,
    max_attempts: int = 5,
    headers: dict[str, str] | None = None,
    params: dict[str, str] | None = None,
) -> httpx.Response:
    """OSS GET with transport-layer retry. Status-code handling is the
    caller's responsibility — this helper ONLY retries on the
    transport exceptions httpx raises before a status code arrives.

    Matches the retry budget + backoff used by the Azure CMDB driver's
    ``_request_with_transport_retry`` (providers/azure.py) for parity:
    5 attempts, exponential backoff ``min(2 ** attempt, 60)`` seconds.

    Retries are safe for OSS GetService because the request is
    idempotent and the signature stays valid for the full 15-minute
    ``Date`` header window — we don't need to re-sign per attempt.
    """
    attempt = 0
    while True:
        try:
            return await client.get(url, headers=headers, params=params)
        except _OSS_TRANSPORT_RETRY_EXCEPTIONS as exc:
            if attempt >= max_attempts - 1:
                raise
            delay = min(2 ** attempt, 60)
            logger.warning(
                "Aliyun OSS transport error %s attempt=%d/%d "
                "delay=%ds: %s",
                endpoint_label, attempt + 1, max_attempts, delay,
                format_exception_for_log(exc, max_len=200),
            )
            await asyncio.sleep(delay)
            attempt += 1


@register_provider
class AliyunProvider(CloudProvider):

    @classmethod
    def provider_type(cls) -> str:
        return "aliyun"

    @classmethod
    def supported_capabilities(cls) -> frozenset[ProviderCapability]:
        return frozenset({
            ProviderCapability.SERVERS,
            ProviderCapability.DOMAINS,
            ProviderCapability.DNS_RECORDS,
            ProviderCapability.CERTIFICATES,
            ProviderCapability.LOAD_BALANCERS,
            ProviderCapability.CDN,
            ProviderCapability.OBJECT_STORAGE,
            ProviderCapability.METRICS,
        })

    def __init__(self, credentials: ProviderCredentials) -> None:
        super().__init__(credentials)
        self._ak = credentials.fields["access_key_id"]
        self._sk = credentials.fields["access_key_secret"]
        # Optional STS session token. Empty when using long-lived AK/SK
        # (the master / single-account case); populated when a subclass
        # (AlicloudIntlProvider) temporarily swaps in AssumeRole'd
        # credentials for a sub-account fetch. Aliyun signing requires
        # the token to ride alongside the temp AK as a query parameter,
        # which ``_common_params`` injects when set.
        self._security_token: str | None = None
        self._client = httpx.AsyncClient(timeout=30.0, trust_env=False)

    def _regions(self) -> list[str]:
        """Region list to iterate for region-scoped calls (ECS, SLB).

        Honors ``regions`` in config (comma-separated) first; falls back
        to the class-level default. Subclasses (AliCloud 海外) override
        the default. ``DNS`` / ``domain`` / ``OSS`` / ``CDN`` endpoints
        are global — they intentionally don't use this list.
        """
        configured = (self._credentials.fields.get("regions") or "").strip()
        if configured:
            return [r.strip() for r in configured.split(",") if r.strip()]
        return list(ECS_REGIONS)

    async def _call(self, endpoint: str, params: dict) -> dict:
        # When running with assumed-role credentials, ``SecurityToken``
        # must be in the signed query — Aliyun rejects ``InvalidSecurityToken``
        # otherwise. ``_common_params`` set the temp ``AccessKeyId``;
        # we inject the matching token here so every codepath benefits
        # without each ``list_*`` call having to remember.
        if self._security_token:
            params.setdefault("SecurityToken", self._security_token)
        # Bounded retry on 429 / 5xx: exponential backoff (1s, 2s, 4s).
        # 4xx other than 429 are not retried — they are deterministic
        # client errors. Auth (401/403) propagates immediately, EXCEPT
        # when the body's ``Code`` is one of the well-known
        # service-not-subscribed values — in that case the credential
        # is fine, the service simply isn't activated, and we surface
        # ``CapabilityNotSupported`` so the scheduled-job marks the
        # capability ``skipped`` instead of red ``failed (auth error)``.
        # All error messages flow through ``_redact_aliyun_error`` so
        # any echoed AccessKeyId / Signature / SecurityToken in the
        # response body is stripped before it lands in logs.
        last_exc: Exception | None = None
        for attempt in range(3):
            # Refresh Timestamp + SignatureNonce per attempt. Aliyun
            # rejects requests with Timestamp skew >15min (InvalidTime
            # Stamp.Expired). If the worker has been blocked for
            # >15min before the retry fires (asyncio backpressure /
            # stalled prior leg) every attempt fails on a frozen
            # Timestamp. Regenerating the nonce also avoids replay
            # collisions if any retry eventually succeeds.
            #
            # CRITICAL: drop the prior attempt's ``Signature`` before
            # re-signing. ``_aliyun_sign`` sorts ALL params into the
            # signing string; if the stale Signature value lingers it
            # gets included in the new string_to_sign and Aliyun
            # rejects with ``SignatureDoesNotMatch`` because the
            # server recomputes the signature WITHOUT the Signature
            # param. Without this pop, retries are guaranteed to fail.
            params.pop("Signature", None)
            params["Timestamp"] = datetime.now(UTC).strftime(
                "%Y-%m-%dT%H:%M:%SZ"
            )
            params["SignatureNonce"] = uuid.uuid4().hex
            params["Signature"] = _aliyun_sign(params, self._sk)
            try:
                resp = await self._client.get(endpoint, params=params)
            except httpx.HTTPError as exc:
                # Round-22 audit: httpx connect/read timeouts + remote
                # protocol errors are NOT subclasses of HTTP-status
                # codes; previously they propagated raw out of _call
                # so the sync engine couldn't classify them as
                # transient and the retry semantics diverged from
                # every other provider. Wrap as ProviderTransientError
                # — bounded backoff loop already handles it; on the
                # third attempt the caller sees a uniform transient.
                last_exc = ProviderTransientError(
                    _redact_aliyun_error(str(exc) or type(exc).__name__),
                )
                if attempt < 2:
                    await asyncio.sleep(2 ** attempt)
                    continue
                raise last_exc from exc
            if resp.status_code in (401, 403):
                code = _extract_aliyun_error_code(resp)
                if code in ALIYUN_SERVICE_NOT_SUBSCRIBED_CODES:
                    raise CapabilityNotSupported(
                        f"Aliyun service not activated for this account "
                        f"(Code={code!r}). Skipping — credential is valid; "
                        f"subscribe the service in the Aliyun console to "
                        f"enable this capability."
                    )
                raise ProviderAuthError(_redact_aliyun_error(resp.text))
            if resp.status_code == 429 or resp.status_code >= 500:
                last_exc = (
                    ProviderRateLimitError()
                    if resp.status_code == 429
                    else ProviderTransientError(_redact_aliyun_error(resp.text))
                )
                if attempt < 2:
                    await asyncio.sleep(2 ** attempt)
                    continue
                raise last_exc
            # Some Aliyun services return 200 OK with an error JSON
            # body containing ``Code``+``Message`` instead of HTTP 4xx
            # — surface SNF cases here too. resp.raise_for_status() is
            # a no-op for 2xx so the body needs explicit inspection.
            if 200 <= resp.status_code < 300:
                code_in_body = _extract_aliyun_error_code(resp)
                if code_in_body in ALIYUN_SERVICE_NOT_SUBSCRIBED_CODES:
                    raise CapabilityNotSupported(
                        f"Aliyun service not activated for this account "
                        f"(Code={code_in_body!r}). Skipping — credential "
                        f"is valid; subscribe the service in the Aliyun "
                        f"console to enable this capability."
                    )
            resp.raise_for_status()
            return resp.json()
        # Defensive fallback — loop should always either return or raise.
        if last_exc:
            raise last_exc
        raise ProviderTransientError("Aliyun _call exhausted retries without response")

    # ── Utilization metrics (CloudMonitor / Cms) ──────────────────────
    #
    # CloudMonitor's ``DescribeMetricList`` (namespace ``acs_ecs_dashboard``)
    # is account-global — one query per metric returns datapoints for every
    # monitored instance (each datapoint carries ``instanceId``), so we
    # query once per metric and filter to the target set in-memory rather
    # than fanning out per region. CPU + network rates land agentless;
    # memory / disk usage need the CloudMonitor agent and are absent (→
    # NULL) otherwise. Network is reported as a RATE (bit/s) and converted
    # to bytes-per-hour to match the cross-cloud ``net_*_bytes`` contract.
    _CMS_VERSION = "2019-01-01"
    _CMS_NAMESPACE = "acs_ecs_dashboard"
    # (field, MetricName, kind) — kind: "pct" | "rate_in" | "rate_out"
    _CMS_METRIC_SPECS = (
        ("cpu", "CPUUtilization", "pct"),
        ("mem", "memory_usedutilization", "pct"),
        ("disk", "diskusage_utilization", "pct"),
        ("net_in_pub", "InternetInRate", "rate"),
        ("net_out_pub", "InternetOutRate", "rate"),
        ("net_in_pri", "IntranetInRate", "rate"),
        ("net_out_pri", "IntranetOutRate", "rate"),
    )

    def _cms_endpoint(self) -> str:
        """CloudMonitor (Cms) API endpoint. China (mainland) accounts use
        the central cn-hangzhou host; the intl subclass overrides."""
        return "https://metrics.cn-hangzhou.aliyuncs.com/"

    async def list_server_metrics(
        self,
        targets: list[ServerMetricTarget],
        *,
        lookback_days: int = 14,
        period_seconds: int = 3600,
    ) -> list[ServerMetricPoint]:
        want = {t.instance_id for t in targets if t.instance_id}
        if not want:
            return []
        endpoint = self._cms_endpoint()
        end_ms = int(datetime.now(UTC).timestamp() * 1000)
        start_ms = end_ms - lookback_days * 86400 * 1000

        # (instance_id, ts_ms) -> field dict of raw metric values
        merged: dict[tuple[str, int], dict[str, float]] = {}

        for field_key, metric_name, _kind in self._CMS_METRIC_SPECS:
            next_token: str | None = None
            pages = 0
            while True:
                pages += 1
                if pages > _MAX_LIST_PAGES:
                    break
                params = _common_params("DescribeMetricList", self._ak)
                params["Version"] = self._CMS_VERSION
                params.update({
                    "Namespace": self._CMS_NAMESPACE,
                    "MetricName": metric_name,
                    "Period": str(period_seconds),
                    "StartTime": str(start_ms),
                    "EndTime": str(end_ms),
                    "Length": "1000",
                })
                if next_token:
                    params["NextToken"] = next_token
                try:
                    data = await self._call(endpoint, params)
                except CapabilityNotSupported:
                    # CloudMonitor not activated → no metrics at all.
                    return []
                except (ProviderTransientError, ProviderRateLimitError) as exc:
                    self.partial_failures.append({
                        "leg": f"metric:{metric_name}",
                        "capability": "metrics",
                        "reason": "transient",
                        "error_message": _redact_aliyun_error(str(exc)),
                    })
                    break
                except ProviderAuthError:
                    # CPU is agentless — a 403 on the PRIMARY cpu metric means
                    # the credential itself is broken / CloudMonitor not
                    # authorized, which must surface as a hard failure (else a
                    # dead AK/SK looks like a green run producing no_metrics
                    # for every server). Agent-only metrics (memory/disk) 403
                    # when the CloudMonitor agent isn't installed — that IS
                    # benign, skip quietly.
                    if field_key == "cpu":
                        raise
                    self.partial_failures.append({
                        "leg": f"metric:{metric_name}",
                        "capability": "metrics",
                        "reason": "unavailable_or_denied",
                    })
                    break

                raw_points = data.get("Datapoints")
                if raw_points:
                    try:
                        parsed = json.loads(raw_points)
                    except (ValueError, TypeError):
                        parsed = []
                    for dp in parsed:
                        instance_id = dp.get("instanceId")
                        ts = dp.get("timestamp")
                        avg = dp.get("Average")
                        if instance_id not in want or ts is None or avg is None:
                            continue
                        merged.setdefault((instance_id, int(ts)), {})[field_key] = float(avg)
                        maxv = dp.get("Maximum")
                        if field_key == "cpu" and maxv is not None:
                            merged.setdefault((instance_id, int(ts)), {})["cpu_max"] = float(maxv)

                next_token = data.get("NextToken")
                if not next_token:
                    break

        return self._cms_merge_to_points(merged, period_seconds)

    @staticmethod
    def _cms_merge_to_points(
        merged: dict[tuple[str, int], dict[str, float]],
        period_seconds: int,
    ) -> list[ServerMetricPoint]:
        points: list[ServerMetricPoint] = []
        for (instance_id, ts_ms), f in merged.items():
            # Rate (bit/s) → total bytes over the bucket: bps / 8 * period.
            def _bytes(*keys: str) -> float | None:
                vals = [f[k] for k in keys if k in f]
                if not vals:
                    return None
                return sum(vals) / 8.0 * period_seconds

            points.append(ServerMetricPoint(
                instance_id=instance_id,
                timestamp=datetime.fromtimestamp(ts_ms / 1000, tz=UTC),
                cpu_pct=f.get("cpu"),
                cpu_pct_max=f.get("cpu_max"),
                mem_pct=f.get("mem"),
                disk_used_pct=f.get("disk"),
                net_in_bytes=_bytes("net_in_pub", "net_in_pri"),
                net_out_bytes=_bytes("net_out_pub", "net_out_pri"),
            ))
        return points

    async def list_servers(self) -> list[CloudServer]:
        servers: list[CloudServer] = []
        for region in self._regions():
            page = 1
            while True:
                params = _common_params("DescribeInstances", self._ak)
                params.update({
                    "RegionId": region,
                    "PageSize": "100",
                    "PageNumber": str(page),
                })
                try:
                    data = await self._call(
                        "https://ecs.aliyuncs.com/", params,
                    )
                except (ProviderTransientError, ProviderRateLimitError) as exc:
                    # Round-22 audit: parity with list_load_balancers —
                    # apply per-region tolerance for transient/throttle
                    # errors. Previously list_servers would propagate
                    # and abort the whole sweep, losing the other
                    # regions' instances even though only one region
                    # was unhealthy. Auth errors still propagate
                    # (raised as ProviderAuthError from _call).
                    self.partial_failures.append({
                        "leg": f"region:{region}",
                        "capability": "servers",
                        "reason": (
                            "rate_limit"
                            if isinstance(exc, ProviderRateLimitError)
                            else "transient"
                        ),
                        "page": page,
                    })
                    break
                instances = (data.get("Instances") or {}).get("Instance", [])
                if not instances:
                    break
                for inst in instances:
                    instance_id = inst.get("InstanceId")
                    if not instance_id:
                        # Defensive: skip rows missing the primary key rather
                        # than raising KeyError mid-sweep.
                        continue
                    public_ips = (
                        (inst.get("PublicIpAddress") or {}).get("IpAddress", [])
                        + [
                            eip["IpAddress"]
                            for eip in (inst.get("EipAddress") or {}).get("EipAddress", [])
                            if eip.get("IpAddress")
                        ]
                    )
                    private_ips = (inst.get("InnerIpAddress") or {}).get("IpAddress", [])
                    # VPC private IPs
                    for nic in (inst.get("NetworkInterfaces") or {}).get("NetworkInterface", []):
                        for pip in (nic.get("PrivateIpSets") or {}).get("PrivateIpSet", []):
                            ip = pip.get("PrivateIpAddress")
                            if ip and ip not in private_ips:
                                private_ips.append(ip)
                    tags = {
                        t.get("TagKey", ""): t.get("TagValue", "")
                        for t in (inst.get("Tags") or {}).get("Tag", [])
                        if t.get("TagKey")
                    }
                    raw_status = inst.get("Status", "")
                    status = {
                        "Running": "running",
                        "Stopped": "stopped",
                        "Deleted": "terminated",
                    }.get(raw_status, raw_status.lower())
                    servers.append(CloudServer(
                        instance_id=instance_id,
                        name=inst.get("InstanceName", ""),
                        status=status,
                        region=region,
                        zone=inst.get("ZoneId"),
                        vcpu=inst.get("Cpu"),
                        memory_mb=inst.get("Memory"),
                        public_ips=public_ips,
                        private_ips=private_ips,
                        image_id=inst.get("ImageId"),
                        os_type="windows" if "win" in (inst.get("OSType") or "").lower() else "linux",
                        tags=tags,
                        raw=inst,
                    ))
                page += 1
        return servers

    # ── Server lifecycle (ECS actions) ────────────────────────
    #
    # Aliyun ECS exposes StartInstance / StopInstance / RebootInstance
    # / DeleteInstance on the same ``ecs.aliyuncs.com`` endpoint the
    # list call uses — same v2014-05-26 API version, same HMAC-SHA1
    # signing path via ``_call``.
    #
    # Idempotency: Aliyun's error codes ``IncorrectInstanceStatus`` /
    # ``InvalidInstanceStatus.Stopped`` etc. fire when the instance is
    # already in the target state. We surface those as success so
    # double-clicks don't 5xx.

    _ALIYUN_IDEMPOTENT_CODES: frozenset[str] = frozenset({
        # Generic "you asked to start an already-running instance" / etc.
        # NOTE: must be EXACT Code-field values from the Aliyun JSON
        # body, not substring matches. ``IncorrectInstanceStatus``
        # (bare) is also emitted by some legacy endpoints; keep it
        # alongside the dotted variants.
        "IncorrectInstanceStatus",
        "IncorrectInstanceStatus.Initializing",
        "IncorrectInstanceStatus.Running",
        "IncorrectInstanceStatus.Stopped",
        # ECS "operation conflicts with current status" — same idea.
        "OperationConflict",
        # Termination of an already-deleted/terminated instance.
        "InvalidInstanceId.NotFound",
    })

    # Aliyun error messages have shape ``... "Code":"<code_value>" ...``
    # in the JSON body. Extract via anchored regex so a real auth error
    # whose Message text happens to mention ``IncorrectInstanceStatus``
    # doesn't false-positive into "idempotent success".
    _ALIYUN_CODE_FIELD_RE = re.compile(r'"Code"\s*:\s*"([^"]+)"')

    @classmethod
    def _is_idempotent_aliyun_error(cls, exc_msg: str) -> bool:
        """Structured check: parse the ``Code`` field out of the redacted
        Aliyun error body, then check exact membership. Falls back to
        False on parse miss (safer than substring fuzziness)."""
        if not exc_msg:
            return False
        m = cls._ALIYUN_CODE_FIELD_RE.search(exc_msg)
        if not m:
            return False
        return m.group(1) in cls._ALIYUN_IDEMPOTENT_CODES

    async def _lifecycle_call(
        self,
        action: str,
        *,
        instance_id: str,
        region: str | None,
        extra: dict | None = None,
    ) -> None:
        if not region:
            raise ValueError(
                f"Aliyun {action} requires region — server row has no region",
            )
        params = _common_params(action, self._ak)
        params.update({
            "RegionId": region,
            "InstanceId": instance_id,
        })
        if extra:
            params.update(extra)
        try:
            await self._call("https://ecs.aliyuncs.com/", params)
        except ProviderAuthError as exc:
            # Aliyun returns 403 with ``IncorrectInstanceStatus`` (etc.)
            # in the body — our ``_call`` classifies the HTTP 403 as
            # auth before reading the ``Code`` field. Re-classify when
            # the body's ``Code`` matches one of the idempotent codes.
            msg = str(exc)
            if self._is_idempotent_aliyun_error(msg):
                logger.info(
                    "Aliyun %s on %s already in target state — "
                    "treating as idempotent success: %s",
                    action, instance_id, msg,
                )
                return
            raise
        except ProviderTransientError as exc:
            msg = str(exc)
            if self._is_idempotent_aliyun_error(msg):
                logger.info(
                    "Aliyun %s on %s already in target state — "
                    "treating as idempotent success: %s",
                    action, instance_id, msg,
                )
                return
            raise

    async def power_on(
        self,
        *,
        instance_id: str,
        region: str | None = None,
        project_id: str | None = None,
    ) -> None:
        await self._lifecycle_call(
            "StartInstance", instance_id=instance_id, region=region,
        )

    async def power_off(
        self,
        *,
        instance_id: str,
        region: str | None = None,
        project_id: str | None = None,
        force: bool = True,
    ) -> None:
        # ``ForceStop=true`` default matches legacy CMDB behavior. A
        # graceful shutdown can hang indefinitely when the guest OS is
        # unresponsive — for an admin-driven Stop button the
        # legacy-equivalent "stop now" semantic is what users expect.
        # Pass ``force=False`` explicitly when a future caller (e.g. a
        # graceful-maintenance workflow) needs OS-level shutdown.
        await self._lifecycle_call(
            "StopInstance",
            instance_id=instance_id,
            region=region,
            extra={"ForceStop": "true" if force else "false"},
        )

    async def restart_server(
        self,
        *,
        instance_id: str,
        region: str | None = None,
        project_id: str | None = None,
    ) -> None:
        await self._lifecycle_call(
            "RebootInstance", instance_id=instance_id, region=region,
        )

    async def destroy_server(
        self,
        *,
        instance_id: str,
        region: str | None = None,
        project_id: str | None = None,
    ) -> None:
        # ``Force=true`` lets Aliyun terminate even when the instance is
        # currently running (otherwise it must be stopped first and the
        # admin would have to two-step every deletion). Match legacy
        # CMDB's TerminateInstances-with-force shape.
        await self._lifecycle_call(
            "DeleteInstance",
            instance_id=instance_id,
            region=region,
            extra={"Force": "true"},
        )

    # Subclasses override when the same RPC shape lives on a
    # different domain — e.g. AlicloudIntlProvider uses
    # ``domain-intl.aliyuncs.com``. Domestic Aliyun account keys
    # cannot call the intl domain service and vice-versa, so the
    # endpoint split is essential (200 OK vs 400 Bad Request observed
    # in prod 2026-04-24 07:00).
    _DOMAIN_API_HOST = "https://domain.aliyuncs.com/"
    # Each host pins its own API version independently. Probed
    # 2026-04-24 against the real endpoints:
    #   domain.aliyuncs.com       → 2018-01-29
    #   domain-intl.aliyuncs.com  → 2017-12-18 (returns InvalidVersion
    #                               on 2018-01-29)
    # Domestic silently accepted the ECS default (2014-05-26) for years
    # so the mismatch went unnoticed until the intl split.
    _DOMAIN_API_VERSION = "2018-01-29"

    async def list_domains(self) -> list[CloudDomain]:
        domains: list[CloudDomain] = []
        page = 1
        while True:
            if page > _MAX_LIST_PAGES:
                logger.warning("aliyun list_domains hit page cap %d; truncating", _MAX_LIST_PAGES)
                break
            params = _common_params("QueryDomainList", self._ak)
            params["Version"] = self._DOMAIN_API_VERSION
            params.update({
                "PageSize": "100",
                "PageNum": str(page),
            })
            data = await self._call(self._DOMAIN_API_HOST, params)
            items = (data.get("Data") or {}).get("Domain", [])
            if not items:
                break
            for d in items:
                expiry_date = None
                if d.get("ExpirationDate"):
                    try:
                        expiry_date = datetime.strptime(
                            d["ExpirationDate"], "%Y-%m-%d %H:%M:%S"
                        ).date()
                    except ValueError:
                        pass
                domains.append(CloudDomain(
                    domain_name=d.get("DomainName", ""),
                    registrar="aliyun",
                    expiry_date=expiry_date,
                    auto_renew=d.get("AutoRenew", False),
                    icp_filing_no=None,  # separate ICP API
                    status="active" if d.get("DomainStatus") == "3" else "inactive",
                    raw=d,
                ))
            page += 1
        return domains

    # Aliyun SSL 证书服务 (CAS). DescribeUserCertificateList is the
    # classic list endpoint (v2018-07-13) returning the user's uploaded +
    # purchased certs with CN / SANs / issuer / start+end dates.
    _CAS_API_HOST = "https://cas.aliyuncs.com/"
    _CAS_API_VERSION = "2018-07-13"

    async def list_certificates(self) -> list[CloudCertificate]:
        certs: list[CloudCertificate] = []
        page = 1
        while True:
            if page > _MAX_LIST_PAGES:
                logger.warning("aliyun list_certificates hit page cap %d; truncating", _MAX_LIST_PAGES)
                break
            params = _common_params("DescribeUserCertificateList", self._ak)
            params["Version"] = self._CAS_API_VERSION
            params.update({"ShowSize": "50", "CurrentPage": str(page)})
            data = await self._call(self._CAS_API_HOST, params)
            items = data.get("CertificateList") or []
            if not items:
                break
            for c in items:
                sans_raw = c.get("Sans") or ""
                sans = [s.strip() for s in str(sans_raw).split(",") if s.strip()]
                # CAS ``Fingerprint`` format varies (often SHA1, colon-
                # delimited). Normalize to lowercase hex; only treat it as
                # a SHA-256 fingerprint (the shape the probed rows use for
                # cross-source dedup) when it's exactly 64 hex chars —
                # otherwise leave it out so we don't false-match on a SHA1.
                fp_raw = str(c.get("Fingerprint") or "").replace(":", "").lower()
                fp256 = fp_raw if len(fp_raw) == 64 and all(
                    ch in "0123456789abcdef" for ch in fp_raw
                ) else None
                certs.append(CloudCertificate(
                    cloud_cert_id=str(c.get("Id", "") or ""),
                    common_name=c.get("Common") or c.get("Name"),
                    sans=sans,
                    issuer=c.get("Issuer"),
                    not_before=_cas_date(c.get("StartDate")),
                    not_after=_cas_date(c.get("EndDate")),
                    status="expired" if c.get("Expired") else "issued",
                    serial_number=c.get("SerialNo") or c.get("Serial"),
                    fingerprint_sha256=fp256,
                    raw=c,
                ))
            total = data.get("TotalCount")
            if total is not None and page * 50 >= int(total):
                break
            page += 1
        return certs

    async def list_dns_records(self) -> list[CloudDnsRecord]:
        records: list[CloudDnsRecord] = []
        # Step 1: discover all DNS zones managed by Aliyun DNS
        zone_names: list[str] = []
        page = 1
        while True:
            if page > _MAX_LIST_PAGES:
                logger.warning("aliyun list_dns_records zone-scan hit page cap %d; truncating", _MAX_LIST_PAGES)
                break
            params = _common_params("DescribeDomains", self._ak)
            params["Version"] = "2015-01-09"
            params.update({"PageSize": "100", "PageNumber": str(page)})
            data = await self._call("https://alidns.aliyuncs.com/", params)
            items = (data.get("Domains") or {}).get("Domain", [])
            if not items:
                break
            zone_names.extend(d.get("DomainName", "") for d in items if d.get("DomainName"))
            page += 1

        # Step 2: fetch records per zone
        for zone in zone_names:
            rpage = 1
            while True:
                if rpage > _MAX_LIST_PAGES:
                    logger.warning("aliyun list_dns_records hit page cap %d for zone %s; truncating", _MAX_LIST_PAGES, zone)
                    break
                params = _common_params("DescribeDomainRecords", self._ak)
                params["Version"] = "2015-01-09"
                params.update({
                    "DomainName": zone,
                    "PageSize": "500",
                    "PageNumber": str(rpage),
                })
                try:
                    data = await self._call("https://alidns.aliyuncs.com/", params)
                except (ProviderAuthError, ProviderRateLimitError, ProviderTransientError):
                    # Auth / rate limit / 5xx must propagate — swallowing them
                    # would let mark-and-sweep nuke previously-seen records.
                    raise
                except httpx.HTTPStatusError as exc:
                    # Per-zone "not found" only — narrow not-found code check.
                    code = ""
                    try:
                        code = str((exc.response.json() or {}).get("Code") or "")
                    except (ValueError, AttributeError):
                        pass
                    if code in ("InvalidDomainName.NoExist", "DomainRecordNotBelongToUser"):
                        break
                    raise
                items = (data.get("DomainRecords") or {}).get("Record", [])
                if not items:
                    break
                for r in items:
                    records.append(CloudDnsRecord(
                        zone_name=zone,
                        record_id=str(r.get("RecordId", "")),
                        name=r.get("RR", "@"),
                        record_type=r.get("Type", "A"),
                        value=r.get("Value", ""),
                        ttl=int(r.get("TTL", 600)),
                        line=r.get("Line"),
                        status="enable" if r.get("Status") == "ENABLE" else "disable",
                        raw=r,
                    ))
                rpage += 1
        return records

    async def list_load_balancers(self) -> list[CloudLoadBalancer]:
        lbs: list[CloudLoadBalancer] = []
        page_size = 100  # Aliyun SLB max page size
        for region in self._regions():
            all_lbs: list[dict] = []
            page = 1
            region_skipped = False
            while True:
                params = _common_params("DescribeLoadBalancers", self._ak)
                params["Version"] = "2014-05-15"
                params["RegionId"] = region
                params["PageNumber"] = str(page)
                params["PageSize"] = str(page_size)
                try:
                    data = await self._call("https://slb.aliyuncs.com/", params)
                except ProviderAuthError:
                    # Auth must propagate — broken AK/SK, not a per-region issue.
                    raise
                except (ProviderRateLimitError, ProviderTransientError) as exc:
                    # Per-region tolerance — log and continue, matches AWS / Tencent.
                    logger.info(
                        "Aliyun SLB region %s skipped: %s", region, str(exc)[:200],
                    )
                    region_skipped = True
                    break
                batch = (data.get("LoadBalancers") or {}).get("LoadBalancer", [])
                if not batch:
                    break
                all_lbs.extend(batch)
                page += 1
            if region_skipped:
                continue
            for lb in all_lbs:
                network_type = "internal" if lb.get("AddressType") == "intranet" else "public"
                lbs.append(CloudLoadBalancer(
                    lb_id=lb.get("LoadBalancerId", ""),
                    name=lb.get("LoadBalancerName", ""),
                    lb_type="classic",
                    network_type=network_type,
                    ip_address=lb.get("Address", ""),
                    region=region,
                    status=lb.get("LoadBalancerStatus", "unknown"),
                    raw=lb,
                ))
        return lbs

    async def list_cdn_domains(self) -> list[CloudCdnDomain]:
        domains: list[CloudCdnDomain] = []
        page = 1
        while True:
            if page > _MAX_LIST_PAGES:
                logger.warning("aliyun list_cdn_domains hit page cap %d; truncating", _MAX_LIST_PAGES)
                break
            params = _common_params("DescribeUserDomains", self._ak)
            params["Version"] = "2018-05-10"
            params.update({"PageSize": "50", "PageNumber": str(page)})
            data = await self._call("https://cdn.aliyuncs.com/", params)
            items = (data.get("Domains") or {}).get("PageData", [])
            if not items:
                break
            for d in items:
                sources = (d.get("Sources") or {}).get("Source", [])
                source_type, source_address = "domain", ""
                if sources:
                    s = sources[0]
                    source_type = s.get("Type", "domain")
                    source_address = s.get("Content", "")
                status_map = {
                    "online": "online", "offline": "offline",
                    "configuring": "configuring", "checking": "configuring",
                }
                status = status_map.get(d.get("DomainStatus", "offline"), "offline")
                domains.append(CloudCdnDomain(
                    domain_name=d.get("DomainName", ""),
                    cname=d.get("Cname", ""),
                    status=status,
                    source_type=source_type,
                    source_address=source_address,
                    is_https=False,  # requires separate API; default False
                    raw=d,
                ))
            page += 1
        return domains

    async def create_dns_record(self, record) -> str:
        params = _common_params("AddDomainRecord", self._ak)
        params["Version"] = "2015-01-09"
        params["DomainName"] = record.zone_name
        params["RR"] = record.name
        params["Type"] = record.record_type
        params["Value"] = record.value
        params["TTL"] = str(record.ttl)
        if record.line:
            params["Line"] = record.line
        data = await self._call("https://alidns.aliyuncs.com/", params)
        return str(data["RecordId"])

    async def update_dns_record(self, record_id: str, zone_name: str, record) -> None:
        params = _common_params("UpdateDomainRecord", self._ak)
        params["Version"] = "2015-01-09"
        params["RecordId"] = record_id
        params["RR"] = record.name
        params["Type"] = record.record_type
        params["Value"] = record.value
        params["TTL"] = str(record.ttl)
        if record.line:
            params["Line"] = record.line
        await self._call("https://alidns.aliyuncs.com/", params)

    async def delete_dns_record(self, record_id: str, zone_name: str) -> None:
        params = _common_params("DeleteDomainRecord", self._ak)
        params["Version"] = "2015-01-09"
        params["RecordId"] = record_id
        await self._call("https://alidns.aliyuncs.com/", params)

    async def set_dns_record_status(self, record_id: str, zone_name: str, enable: bool) -> None:
        """Flip the record's enable / disable flag via Aliyun DNS API.

        Endpoint: ``SetDomainRecordStatus`` (DNS 2015-01-09). Note the
        capitalization difference from DNSPod: Aliyun accepts
        ``"Enable"`` / ``"Disable"`` (PascalCase), while DNSPod wants
        uppercase ``"ENABLE"`` / ``"DISABLE"``. We keep the provider-
        local string here so callers can stay portable on the bool.
        """
        params = _common_params("SetDomainRecordStatus", self._ak)
        params["Version"] = "2015-01-09"
        params["RecordId"] = record_id
        params["Status"] = "Enable" if enable else "Disable"
        await self._call("https://alidns.aliyuncs.com/", params)

    async def list_object_storage(self) -> list[dict]:
        """List OSS buckets via the OSS GetService (ListBuckets) REST API.

        Signing note — OSS Auth v1 (HMAC-SHA1) limitation:
        Auth v1 uses only the Date header (UTC, RFC 2822 via formatdate(usegmt=True))
        for replay prevention. There is no per-request nonce. OSS accepts requests
        whose Date is within ±15 minutes of server time, meaning a captured request
        could be replayed within that window.

        This is an inherent limitation of OSS Auth v1, not a Helix bug.  The
        mitigation is to rotate Access Key credentials regularly and restrict the
        key's RAM policy to read-only ListBuckets.  Upgrade to OSS Auth v4 (signed
        headers + x-oss-content-sha256) if stricter replay protection is required.

        STS temp-credential handling: when ``self._security_token`` is set
        (sub-account fanout via AssumeRole — see :class:`AlicloudIntlProvider`),
        the token MUST ride in the ``x-oss-security-token`` header AND be
        included in the ``CanonicalizedOSSHeaders`` block of StringToSign.
        Forgetting either half yields a 403 ``InvalidSecurityToken.NotFound``
        on every sub-account leg regardless of the assumed role's policy —
        same protocol contract as the ``SecurityToken=`` query parameter in
        the RPC ``_call`` path, just expressed via headers because OSS uses
        REST-style auth instead of Aliyun's RPC query signing.
        """
        # formatdate(usegmt=True) always produces a UTC date string (e.g.
        # "Tue, 15 Apr 2026 08:00:00 GMT") regardless of the host's local
        # timezone — safe to use in non-UTC environments.
        url = "https://oss.aliyuncs.com/"
        _OSS_NS = "http://doc.oss-cn-hangzhou.aliyuncs.com"
        result: list[dict] = []
        marker = ""
        async with httpx.AsyncClient(timeout=30.0, trust_env=False) as client:
            while True:
                date_str = formatdate(usegmt=True)  # nosec B324 — Auth v1, see docstring
                # CanonicalizedOSSHeaders per OSS Auth v1 spec: every
                # request header starting with ``x-oss-`` (lowercase),
                # sorted by name, joined as ``name:value\n``. When no
                # such header is present the block is empty (legacy
                # behavior preserved). We only emit one such header —
                # ``x-oss-security-token`` — so sorting is trivial.
                # Re-read ``self._security_token`` per iteration so a
                # token rotation mid-pagination (e.g. STS refresh fired
                # by ``_impersonate``) takes effect on the next page.
                token = self._security_token
                # Reject control characters in the token defensively. A
                # newline/CR in the value would (a) break the OSS Auth v1
                # canonical-headers parser yielding ``SignatureDoesNotMatch``
                # with an opaque cause, and (b) be rejected by httpx's
                # ``h11.LocalProtocolError`` as an "illegal header value"
                # — propagating out as an uncategorized HTTPError that
                # skips this loop's 401/403/429/5xx classifier. Real STS
                # tokens never contain these chars; this guard converts
                # the rare malformed-token scenario into a clean
                # ``ProviderAuthError`` so the sync engine's retry/skip
                # paths react uniformly.
                if token and ("\n" in token or "\r" in token):
                    raise ProviderAuthError(
                        "Aliyun OSS: STS SecurityToken contains control "
                        "character — likely upstream STS refresh bug",
                    )
                canonicalized_oss_headers = (
                    f"x-oss-security-token:{token}\n" if token else ""
                )
                string_to_sign = (
                    f"GET\n\n\n{date_str}\n{canonicalized_oss_headers}/"
                )
                signature = base64.b64encode(
                    hmac.new(  # nosec B324 — OSS Auth v1 mandates HMAC-SHA1
                        self._sk.encode("utf-8"),
                        string_to_sign.encode("utf-8"),
                        hashlib.sha1,
                    ).digest()
                ).decode("utf-8")
                headers = {
                    "Date": date_str,
                    "Authorization": f"OSS {self._ak}:{signature}",
                }
                if token:
                    headers["x-oss-security-token"] = token
                # OSS GetService caps at 100 buckets per call; walk via marker.
                params = {"max-keys": "100"}
                if marker:
                    params["marker"] = marker
                # Transport-layer retry — pre-fix this single GET had NO
                # try/except wrap, so a ReadTimeout / ConnectError mid-
                # pagination raised raw to ``_fan_out_per_subscription``'s
                # outer handler and the entire OSS leg was marked failed
                # for the tick. The RPC ``_call`` path (ECS / RDS / etc.)
                # already had this wrap (see ``_call`` ~L286) — parity
                # fix so OSS behaves the same way. Re-signing per attempt
                # is unnecessary here because the same signature is valid
                # for the duration of the Date header's 15-minute window.
                resp = await _oss_get_with_transport_retry(
                    client, url,
                    endpoint_label=f"list_buckets marker={marker!r}",
                    headers=headers,
                    params=params,
                )

                if resp.status_code in (401, 403):
                    # OSS SignatureDoesNotMatch error XML embeds the
                    # ``<StringToSign>`` element which carries the
                    # request URL — and OSS Auth v1 puts AccessKeyId
                    # in the signed string. Redact before surfacing
                    # so the AK fragment doesn't land in sync history
                    # / Feishu cards.
                    raise ProviderAuthError(
                        f"Aliyun OSS list_buckets {resp.status_code}: "
                        f"{_redact_aliyun_error(resp.text)}",
                    )
                if resp.status_code == 429:
                    raise ProviderRateLimitError()
                if resp.status_code >= 500:
                    raise ProviderTransientError(
                        f"Aliyun OSS list_buckets {resp.status_code}: "
                        f"{_redact_aliyun_error(resp.text)}",
                    )
                if resp.status_code != 200:
                    raise ProviderTransientError(
                        f"Aliyun OSS list_buckets {resp.status_code}: "
                        f"{_redact_aliyun_error(resp.text)}",
                    )

                root = _safe_fromstring(resp.text)
                buckets = root.findall(".//Bucket") or root.findall(f".//{{{_OSS_NS}}}Bucket")
                for b in buckets:
                    def _t(tag: str, _b: ET.Element = b, _ns: str = _OSS_NS) -> str | None:
                        return _b.findtext(tag) or _b.findtext(f"{{{_ns}}}{tag}")

                    result.append({
                        "bucket_name": _t("Name") or "",
                        "region": _t("Location") or "",
                        "storage_class": _t("StorageClass"),
                        "access_type": None,
                        "status": "active",
                        "size_bytes": None,
                        "object_count": None,
                    })

                # Walk IsTruncated / NextMarker.
                truncated_text = (
                    root.findtext("IsTruncated")
                    or root.findtext(f"{{{_OSS_NS}}}IsTruncated")
                    or "false"
                )
                if str(truncated_text).strip().lower() != "true":
                    break
                next_marker = (
                    root.findtext("NextMarker")
                    or root.findtext(f"{{{_OSS_NS}}}NextMarker")
                    or ""
                )
                if not next_marker:
                    # Fall back to the last seen bucket name as marker.
                    if buckets:
                        last = buckets[-1]
                        next_marker = (
                            last.findtext("Name")
                            or last.findtext(f"{{{_OSS_NS}}}Name")
                            or ""
                        )
                if not next_marker:
                    break
                marker = next_marker
        return result

    # ── Monthly billing via BSS QueryAccountBill ──────────────────
    #
    # BSS endpoint host varies by tenant region:
    #   * 国内阿里云 → ``business.aliyuncs.com`` (default).
    #   * AliCloud 海外 → ``business.ap-southeast-1.aliyuncs.com``
    #     (subclass overrides ``_BSS_HOST``).
    # Using the wrong host returns ``NotApplicable: caller site doesn't
    # match API region`` — see PR #49 for the AliCloud Intl daily fix.
    # Page ceiling for the monthly BSS walk (#2222 — the daily walks in
    # ``integrations/drivers/aliyun_cloud.py`` grew this in #2153; these
    # two monthly ones were left with a bare ``while True`` on NextToken).
    # A degenerate BSS response that keeps returning a token loops for
    # ever: every call succeeds fast, so no per-request timeout fires and
    # the worker watchdog cannot see a hung loop — the job just pins a
    # scheduled-jobs concurrency slot and hammers BSS. 500 × 300 rows is
    # far above any real monthly bill. Inherited by the intl subclass.
    _MAX_BILL_PAGES = 500
    _BSS_HOST = "business.aliyuncs.com"
    _BSS_VERSION = "2017-12-14"

    async def get_monthly_billing(self, month: date | None = None) -> dict:
        """Fetch one month's pretax cost via BSS ``QueryAccountBill``.

        Default ``month``: the previous full month, anchored to the
        scheduled-job timezone (handler computes Asia/Shanghai). Aliyun
        bills natively in CNY — we keep it as CNY in the response and
        flag ``currency='CNY'`` so the multi-cloud aggregator doesn't
        silently sum CNY into a USD total at a stale fixed rate.

        Pagination: BSS ``MaxResults`` defaults to 20; on busy accounts
        a single month spans more products than that. Pin to 300 (the
        documented max) and follow ``NextToken`` until no cursor
        returns. Single API call per page; no per-day fan-out at this
        granularity.
        """
        target = (month or (date.today().replace(day=1) - timedelta(days=1))).replace(day=1)
        cycle = target.strftime("%Y-%m")
        endpoint = f"https://{self._BSS_HOST}/"
        cost_by_service: dict[str, float] = {}
        total_native = 0.0
        currency = "CNY"
        next_token: str | None = None
        last_page: dict = {}
        pages_walked = 0
        while True:
            params = _common_params("QueryAccountBill", self._ak)
            pages_walked += 1
            if pages_walked > self._MAX_BILL_PAGES:
                raise RuntimeError(
                    f"Aliyun QueryAccountBill {cycle} hit the "
                    f"{self._MAX_BILL_PAGES}-page ceiling — refusing to loop; "
                    "surfacing as a transient failure rather than pinning "
                    "a scheduled-jobs slot indefinitely (#2222).",
                )
            # BSS uses a different default Version than ECS. Override
            # AFTER ``_common_params`` populates the rest so the rest of
            # the signing surface stays uniform.
            params["Version"] = self._BSS_VERSION
            params["BillingCycle"] = cycle
            params["Granularity"] = "MONTHLY"
            params["IsGroupByProduct"] = "true"
            params["MaxResults"] = "300"
            if next_token:
                params["NextToken"] = next_token
            data = await self._call(endpoint, params)
            last_page = data
            payload = data.get("Data") or {}
            items = (payload.get("Items") or {}).get("Item") or []
            for item in items:
                amt = float(item.get("PretaxAmount") or 0)
                svc = str(item.get("ProductName") or "Unknown")
                cost_by_service[svc] = cost_by_service.get(svc, 0.0) + amt
                total_native += amt
                cur = item.get("Currency")
                if cur:
                    currency = str(cur)
            prev_token = next_token
            next_token = payload.get("NextToken") or None
            if not next_token:
                break
            if next_token == prev_token:
                raise RuntimeError(
                    f"Aliyun QueryAccountBill {cycle} returned a "
                    "non-advancing NextToken — refusing to loop (#2222).",
                )
        # Carry the NATIVE amount in ``total_cost_usd`` for any KNOWN
        # currency and let the billing handler resolve FX with the
        # platform's dynamic, weekly-refreshed CNY/USD rate — the same
        # contract Tencent / UCloud follow (see their drivers' return
        # comments). The handler keys its CNY→USD rebase on
        # ``currency=='CNY' and total_cost_native is not None`` and only
        # falls back to ``total_cost_usd`` when FX resolution fails — at
        # which point it keeps ``currency='CNY'`` so value + label still
        # agree and the billing-overview reader converts correctly.
        #
        # Round-666 fix: this driver previously pre-multiplied by a
        # STALE hard-coded ``_FX_TO_USD['CNY']=0.14`` while still
        # labelling the row ``currency='CNY'``. On the handler's
        # FX-resolve-fails fallback that persisted an already-USD value
        # tagged CNY, which the overview reader then divided by ~7.2 a
        # SECOND time → ~7× understatement of Aliyun spend. Surfacing
        # the native value (label + value both native) removes that
        # double-conversion and the stale-rate drift entirely.
        #
        # Unknown currency still surfaces ``total_cost_usd=None`` so the
        # handler / budget alert skip USD-denominated alerts rather than
        # reporting e.g. JPY 1,000,000 as $1,000,000.
        fx = _FX_TO_USD.get(currency)
        if fx is None:
            logger.warning(
                "Aliyun billing returned unknown currency %r; "
                "surfacing total_cost_usd=None (add the rate to "
                "_FX_TO_USD if this currency is real). total_native=%s",
                currency, total_native,
            )
            total_usd = None
        else:
            total_usd = total_native
        return {
            "month": target,
            "total_cost_usd": total_usd,
            "total_cost_native": total_native,
            "cost_by_service": cost_by_service,
            "currency": currency,
            "raw_response": last_page,
        }

    async def close(self) -> None:
        await self._client.aclose()