"""Tencent Cloud provider. Capabilities: * SERVERS — CVM ``DescribeInstances`` per region * DOMAINS — Tencent Cloud 域名服务 ``DescribeDomainNameList`` * DNS_RECORDS — DNSPod ``DescribeDomainList`` + ``DescribeRecordList`` * LOAD_BALANCERS — CLB ``DescribeLoadBalancers`` per region * CDN — CDN ``DescribeDomainsConfig`` * OBJECT_STORAGE — COS v5 ``ListBuckets`` (via ``cos-python-sdk-v5``) * IAM — CAM ``ListUsers`` + per-user enrichment DNS write methods (create/update/delete) are exposed to the CMDB ``/cmdb/dns-records`` REST path; everything else is read-only inventory. Region list: ``fields['regions']`` comma-separated, default ``ap-guangzhou,ap-shanghai,ap-beijing,ap-chengdu,ap-nanjing, ap-hongkong,ap-singapore,ap-tokyo,ap-seoul,na-siliconvalley,na-ashburn,eu-frankfurt``. The region list is trimmed to the CVM/CLB regions the SDK supports — DNS/CDN/COS are global or self-routed so they ignore it. """ from __future__ import annotations import asyncio import logging from datetime import UTC, date, datetime from typing import Any from app.domains.cmdb.providers.base import ( CloudCdnDomain, CloudCertificate, CloudDnsRecord, CloudDomain, CloudLoadBalancer, CloudProvider, CloudServer, ProviderCapability, ProviderCredentials, register_provider, ) from app.domains.cmdb.providers.exceptions import ( ProviderAuthError, ProviderTransientError, redact_provider_error, ) from app.integrations._tencent_throttle import ( tencent_call_sync as _tc_call, ) logger = logging.getLogger(__name__) # Tencent public regions with CVM availability. Keep this as the default # so a freshly-configured tenant covers every region out of the box; # admins can narrow it via ``fields['regions']`` if they know their # workloads live in one place. # Fallback only. The live region list comes from ``DescribeRegions`` — # see ``TencentProvider._regions``. Kept so a transient API failure # degrades to the previous behaviour rather than syncing nothing. _DEFAULT_REGIONS = ( "ap-guangzhou,ap-shanghai,ap-beijing,ap-chengdu,ap-nanjing," "ap-hongkong,ap-singapore,ap-tokyo,ap-seoul,ap-bangkok,ap-mumbai," "na-siliconvalley,na-ashburn,eu-frankfurt,sa-saopaulo" ) # COS default region — every Tencent tenant has ap-guangzhou enabled. # Bucket listing is global per-SecretId regardless of the region # passed to the SDK, so the value mostly affects endpoint routing. _COS_DEFAULT_REGION = "ap-guangzhou" def _tc_time(value: str | None) -> datetime | None: """Parse a Tencent timestamp ('YYYY-MM-DD HH:MM:SS', Beijing time UTC+8) into a tz-aware UTC datetime. SSL cert times are returned in Beijing time; storing them as UTC keeps expiry math consistent with every other aware-datetime column.""" if not value: return None from datetime import timedelta, timezone beijing = timezone(timedelta(hours=8)) for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d"): try: return datetime.strptime(value, fmt).replace(tzinfo=beijing).astimezone(UTC) except ValueError: continue return None def _make_tc_client(service: str, credentials: ProviderCredentials, region: str = ""): """Instantiate a Tencent Cloud SDK v3 client for the given service.""" from tencentcloud.common import credential as tc_cred from tencentcloud.common.profile.client_profile import ClientProfile from tencentcloud.common.profile.http_profile import HttpProfile cred = tc_cred.Credential( credentials.fields["secret_id"], credentials.fields["secret_key"], ) cp = ClientProfile() cp.httpProfile = HttpProfile() cp.httpProfile.reqTimeout = 30 if service == "cvm": from tencentcloud.cvm.v20170312 import cvm_client return cvm_client.CvmClient(cred, region, cp) if service == "dnspod": from tencentcloud.dnspod.v20210323 import dnspod_client return dnspod_client.DnspodClient(cred, region, cp) if service == "clb": from tencentcloud.clb.v20180317 import clb_client return clb_client.ClbClient(cred, region, cp) if service == "cdn": from tencentcloud.cdn.v20180606 import cdn_client return cdn_client.CdnClient(cred, region, cp) if service == "domain": from tencentcloud.domain.v20180808 import domain_client return domain_client.DomainClient(cred, region, cp) if service == "ssl": from tencentcloud.ssl.v20191205 import ssl_client return ssl_client.SslClient(cred, region, cp) if service == "cam": from tencentcloud.cam.v20190116 import cam_client return cam_client.CamClient(cred, region, cp) if service == "billing": from tencentcloud.billing.v20180709 import billing_client return billing_client.BillingClient(cred, region, cp) raise ValueError(f"Unknown Tencent service: {service!r}") def _translate_sdk_exc(exc: Exception) -> Exception: """Map TencentCloudSDKException codes onto Helix provider exceptions. Keeps the ``list_*`` methods readable and matches other providers — auth errors surface as non-retryable ``ProviderAuthError`` so the sync engine doesn't burn retries on a broken SecretKey. All emitted messages route through ``redact_provider_error`` so SDK exception strings carrying RequestId / echoed request params / SecretId fragments don't land in admin-readable surfaces (sync history, Feishu cards). Code value is preserved for code-based classification but stringified through the redactor too — Tencent occasionally embeds the failing API call's identity into the code string in error envelopes. """ safe_msg = redact_provider_error(exc) code = str(getattr(exc, "code", "") or "") if "AuthFailure" in code or "AuthFailure" in safe_msg or "UnauthorizedOperation" in code: return ProviderAuthError(safe_msg) return ProviderTransientError(safe_msg) def _json_safe(obj: Any) -> Any: """Recursively convert SDK response values to JSON-serializable.""" if isinstance(obj, (datetime, date)): return obj.isoformat() if isinstance(obj, dict): return {k: _json_safe(v) for k, v in obj.items()} if isinstance(obj, (list, tuple)): return [_json_safe(v) for v in obj] if isinstance(obj, (bytes, bytearray)): return obj.decode("utf-8", errors="replace") return obj def _coerce_flag(value: Any) -> dict[str, Any]: """CAM returns ``LoginFlag`` / ``ActionFlag`` as an object whose fields are the individual factors (Phone / Token / Stoken / Wechat / Custom / U2FToken / Passkey), each 0 or 1. Some SDK versions hand back a dict instead, so normalise both shapes.""" if value is None: return {} if isinstance(value, dict): return {str(k): v for k, v in value.items()} out: dict[str, Any] = {} for key in ( "Phone", "Token", "Stoken", "Wechat", "Custom", "Mail", "U2FToken", "Passkey", ): if hasattr(value, key): out[key] = getattr(value, key) return out def _any_factor_on(flag: dict[str, Any]) -> bool: """True when ANY second factor is switched on. Deliberately not just ``Stoken`` (the soft token, which is what this tenant happens to use): a user secured with a hardware key or a passkey is protected, and reporting them as "no MFA" would be the same class of lie this whole change is undoing. """ return any(bool(v) for k, v in flag.items() if k != "Mail") def _parse_tc_datetime(value: Any) -> datetime | None: """Parse Tencent's ``YYYY-MM-DD HH:MM:SS`` and ISO-8601 formats.""" if not value: return None s = str(value).strip() if not s: return None # Format 1: "2025-09-01T10:30:00Z" (ISO) try: return datetime.fromisoformat(s.replace("Z", "+00:00")) except ValueError: pass # Format 2: "2025-09-01 10:30:00" (CVM / CAM) for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S"): try: return datetime.strptime(s, fmt) except ValueError: continue return None @register_provider class TencentProvider(CloudProvider): """Tencent Cloud CMDB data source.""" @classmethod def provider_type(cls) -> str: return "tencent" @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.IAM, }) def __init__(self, credentials: ProviderCredentials) -> None: super().__init__(credentials) configured = (credentials.fields.get("regions") or "").strip() # An explicit list stays an explicit NARROWING — an admin who # typed three regions meant three regions. self._configured_regions = [ r.strip() for r in configured.split(",") if r.strip() ] self._discovered_regions: list[str] | None = None @property def _regions(self) -> list[str]: """Regions to sweep for region-scoped resources (CVM / CLB). Prefers what the provider actually offers over a list baked into Helix. A hardcoded list is wrong in the one direction that matters: a machine created in a region nobody thought to add is invisible forever, and invisible reads exactly like "we have no servers there". Prod Tencent was pinned to three regions (guangzhou / shanghai / beijing) out of the 19 the account can use — the sync had been reporting a clean zero for a decade of possible blind spots. (It really is zero today: all 19 were enumerated by hand on 2026-08-12. That is the point — it was confirmed, not assumed.) Falls back to the static list only if ``DescribeRegions`` fails, so a transient API error degrades to the old behaviour instead of syncing nothing. """ if self._configured_regions: return self._configured_regions if self._discovered_regions is None: self._discovered_regions = self._discover_regions() return self._discovered_regions def _discover_regions(self) -> list[str]: try: from tencentcloud.cvm.v20170312 import models as m client = _make_tc_client("cvm", self._credentials, "ap-guangzhou") resp = self._call( client.DescribeRegions, m.DescribeRegionsRequest(), service="cvm", ) regions = [ str(getattr(r, "Region", "") or "").strip() for r in (getattr(resp, "RegionSet", None) or []) ] regions = [r for r in regions if r] if regions: logger.info( "Tencent DescribeRegions: %d region(s)", len(regions), ) return regions logger.warning( "Tencent DescribeRegions returned no regions — falling back " "to the static list", ) except Exception as exc: # noqa: BLE001 logger.warning( "Tencent DescribeRegions failed (%s) — falling back to the " "static list", redact_provider_error(exc), ) return [r.strip() for r in _DEFAULT_REGIONS.split(",") if r.strip()] # ── Throttle helper ─────────────────────────────────────── # # Funnels every Tencent SDK call through the per-credential token # bucket in ``_tencent_throttle``. Tencent CAM publishes a 5 RPS # ceiling and the per-user grants_sync loop trivially exceeded it # before this shim landed (2026-05-09 incident). CVM / CDN / # DNSPod / 域名注册 endpoints have looser caps but still benefit # from a single chokepoint that retries on RequestLimitExceeded # rather than aborting a sweep mid-page. def _secret_id(self) -> str: return self._credentials.fields.get("secret_id") or "" def _call(self, fn: Any, *args: Any, service: str = "cam", **kwargs: Any) -> Any: return _tc_call( fn, *args, secret_id=self._secret_id(), service=service, **kwargs, ) # ── Servers (CVM) ────────────────────────────────────────────────── def _list_servers_sync(self) -> list[CloudServer]: from tencentcloud.common.exception.tencent_cloud_sdk_exception import ( TencentCloudSDKException, ) from tencentcloud.cvm.v20170312 import models as m page_size = 100 # CVM DescribeInstances max servers: list[CloudServer] = [] for region in self._regions: try: client = _make_tc_client("cvm", self._credentials, region) offset = 0 while True: req = m.DescribeInstancesRequest() req.Limit = page_size req.Offset = offset try: resp = self._call(client.DescribeInstances, req, service="cvm") except TencentCloudSDKException as exc: code = str(getattr(exc, "code", "") or "") if "AuthFailure" in code: raise _translate_sdk_exc(exc) from exc # Any non-auth SDK exception for a single region # (region-unavailable, 5xx InternalError, # RequestLimitExceeded, UnsupportedOperation) — # log and skip THAT region rather than abort # the whole sweep. Matches AWS + Aliyun per- # region tolerance. Auth errors keep propagating # (above) so a broken SecretKey still fails loud. logger.info( "Tencent CVM region %s skipped (%s): %s", region, code or "SDK", exc, ) # Round-22 audit: append partial_failures on # the INNER pagination error too. Previously # only the outer per-region handshake error # recorded a partial_failure; an inner # mid-pagination throttle/5xx would silently # drop pages 2..N's instances from the upsert, # safe-sweep saw them as deletions, and # mark-and-sweep nuked them once the cumulative # drop crossed the 5% floor. self.partial_failures.append({ "leg": f"region:{region}", "capability": "servers", "reason": f"inner_sdk:{code or type(exc).__name__}", "page_offset": offset, }) break batch = resp.InstanceSet or [] for inst in batch: servers.append(_server_from_instance(inst, region)) if len(batch) < page_size: break offset += page_size except ProviderAuthError: raise except TencentCloudSDKException as exc: # Outer auth error (e.g. the very first ``_make_tc_client`` # handshake) — propagate. if "AuthFailure" in str(getattr(exc, "code", "") or ""): raise _translate_sdk_exc(exc) from exc # Any other outer SDK failure: log + keep going. # Pre-launch hardening 2026-05-16: record partial_failure # so a region 5xx'ing on every call doesn't silently # drop its instances and let safe_sweep treat them as # real deletions when crossing the 5% floor. code = str(getattr(exc, "code", "") or "") logger.info( "Tencent CVM region %s skipped (outer): %s", region, exc, ) self.partial_failures.append({ "leg": f"region:{region}", "capability": "servers", "reason": f"outer_sdk:{code or type(exc).__name__}", }) return servers async def list_servers(self) -> list[CloudServer]: return await asyncio.to_thread(self._list_servers_sync) # ── Domains (Tencent Cloud 域名注册) ─────────────────────────────── def _list_domains_sync(self) -> list[CloudDomain]: from tencentcloud.common.exception.tencent_cloud_sdk_exception import ( TencentCloudSDKException, ) from tencentcloud.domain.v20180808 import models as m page_size = 100 # API max: 100 domains: list[CloudDomain] = [] try: client = _make_tc_client("domain", self._credentials) offset = 0 while True: req = m.DescribeDomainNameListRequest() req.Limit = page_size req.Offset = offset try: resp = self._call( client.DescribeDomainNameList, req, service="domain", ) except TencentCloudSDKException as exc: code = str(getattr(exc, "code", "") or "") # ``UnauthorizedOperation.NoPermission`` when the tenant # has never used 域名注册 — the endpoint returns an # unusable 403. Distinguish from an actual bad key so # the admin sees a clean "没有域名" vs "auth failed". if "NoPermission" in code or "ServiceNotOpened" in code: logger.info( "Tencent DescribeDomainNameList not enabled: %s", exc, ) return domains raise _translate_sdk_exc(exc) from exc batch = resp.DomainSet or [] for d in batch: expiry_raw = getattr(d, "ExpirationDate", None) expiry_date = None if expiry_raw: try: expiry_date = datetime.strptime( str(expiry_raw), "%Y-%m-%d", ).date() except ValueError: parsed = _parse_tc_datetime(expiry_raw) expiry_date = parsed.date() if parsed else None domains.append(CloudDomain( domain_name=str(getattr(d, "DomainName", "") or ""), registrar="tencent", expiry_date=expiry_date, auto_renew=bool(getattr(d, "AutoRenew", 0) in (1, True)), icp_filing_no=None, status=( "active" if str(getattr(d, "TransferProhibition", "")).lower() != "true" and not getattr(d, "RealNameAuditStatus", "") else "active" ), raw=_json_safe({ "DomainName": getattr(d, "DomainName", None), "AutoRenew": getattr(d, "AutoRenew", None), "ExpirationDate": expiry_raw, "CreationDate": getattr(d, "CreationDate", None), "DomainId": getattr(d, "DomainId", None), "CodeTld": getattr(d, "CodeTld", None), "Tld": getattr(d, "Tld", None), }), )) offset += page_size # Terminate on the authoritative signal — an empty or short # (last) page — rather than TotalCount: a response that omits # or zeroes TotalCount while returning a full batch would # otherwise stop after page 1 and drop domains on later pages. if not batch or len(batch) < page_size: break except TencentCloudSDKException as exc: raise _translate_sdk_exc(exc) from exc return domains async def list_domains(self) -> list[CloudDomain]: return await asyncio.to_thread(self._list_domains_sync) # ── Certificates (SSL) ──────────────────────────────────────────── def _list_certificates_sync(self) -> list[CloudCertificate]: from tencentcloud.common.exception.tencent_cloud_sdk_exception import ( TencentCloudSDKException, ) from tencentcloud.ssl.v20191205 import models as m certs: list[CloudCertificate] = [] try: # SSL is a global (region-agnostic) service. client = _make_tc_client("ssl", self._credentials) offset, limit = 0, 100 while True: req = m.DescribeCertificatesRequest() req.Offset = offset req.Limit = limit resp = self._call( client.DescribeCertificates, req, service="ssl", ) batch = resp.Certificates or [] for c in batch: certs.append(CloudCertificate( cloud_cert_id=getattr(c, "CertificateId", None) or "", common_name=getattr(c, "Domain", None), sans=list(getattr(c, "SubjectAltName", None) or []), issuer=getattr(c, "Issuer", None), not_before=_tc_time(getattr(c, "CertBeginTime", None)), not_after=_tc_time(getattr(c, "CertEndTime", None)), status=str(getattr(c, "Status", "") or ""), serial_number=None, raw={ k: getattr(c, k, None) for k in ( "CertificateId", "Domain", "Alias", "Status", "CertEndTime", "CertBeginTime", "SubjectAltName", "Issuer", "ProductZhName", ) }, )) offset += limit total = getattr(resp, "TotalCount", None) if not batch or (total is not None and offset >= int(total)): break except TencentCloudSDKException as exc: raise _translate_sdk_exc(exc) from exc return certs async def list_certificates(self) -> list[CloudCertificate]: return await asyncio.to_thread(self._list_certificates_sync) # ── DNS (DNSPod) ────────────────────────────────────────────────── def _list_dns_records_sync(self) -> list[CloudDnsRecord]: from tencentcloud.common.exception.tencent_cloud_sdk_exception import ( TencentCloudSDKException, ) from tencentcloud.dnspod.v20210323 import models as m records: list[CloudDnsRecord] = [] try: client = _make_tc_client("dnspod", self._credentials) domains: list[Any] = [] offset = 0 page_size = 3000 # DNSPod DescribeDomainList max while True: req = m.DescribeDomainListRequest() req.Limit = page_size req.Offset = offset try: resp = self._call( client.DescribeDomainList, req, service="dnspod", ) except TencentCloudSDKException as exc: raise _translate_sdk_exc(exc) from exc batch = resp.DomainList or [] domains.extend(batch) if len(batch) < page_size: break offset += page_size for domain in domains: zone_name = getattr(domain, "Name", None) if not zone_name: continue r_offset = 0 r_page = 1000 # DescribeRecordList max try: while True: rec_req = m.DescribeRecordListRequest() rec_req.Domain = zone_name rec_req.Limit = r_page rec_req.Offset = r_offset rec_resp = self._call( client.DescribeRecordList, rec_req, service="dnspod", ) rec_batch = rec_resp.RecordList or [] for r in rec_batch: # Defensive attribute access — a DNSPod SDK-shape # drift or a record missing a field raises # AttributeError, which is NOT a # TencentCloudSDKException and would escape the # per-zone handler below, failing the ENTIRE DNS # capability sync on one malformed record. Mirror # the getattr discipline used by the other list # paths in this driver. rid = getattr(r, "RecordId", None) if rid is None: continue ttl_raw = getattr(r, "TTL", None) try: ttl_val = int(ttl_raw) if ttl_raw is not None else 0 except (TypeError, ValueError): ttl_val = 0 status_raw = getattr(r, "Status", None) records.append(CloudDnsRecord( zone_name=zone_name, record_id=str(rid), name=getattr(r, "Name", "") or "", record_type=getattr(r, "Type", "") or "", value=getattr(r, "Value", "") or "", ttl=ttl_val, line=getattr(r, "Line", None), status="enable" if status_raw == "ENABLE" else "disable", raw={ "RecordId": rid, "Name": getattr(r, "Name", None), "Type": getattr(r, "Type", None), "Value": getattr(r, "Value", None), "TTL": ttl_raw, "Line": getattr(r, "Line", None), "Status": status_raw, }, )) if len(rec_batch) < r_page: break r_offset += r_page except TencentCloudSDKException as exc: # DNSPod occasionally returns ``ResourceNotFound.NoData`` # for zones that were just created — skip instead of # failing the whole sweep. code = str(getattr(exc, "code", "") or "") if "NoData" in code or "ResourceNotFound" in code: continue raise _translate_sdk_exc(exc) from exc except TencentCloudSDKException as exc: raise _translate_sdk_exc(exc) from exc return records async def list_dns_records(self) -> list[CloudDnsRecord]: return await asyncio.to_thread(self._list_dns_records_sync) def _create_dns_record_sync(self, record) -> str: from tencentcloud.common.exception.tencent_cloud_sdk_exception import ( TencentCloudSDKException, ) from tencentcloud.dnspod.v20210323 import models as m try: client = _make_tc_client("dnspod", self._credentials) req = m.CreateRecordRequest() req.Domain = record.zone_name req.SubDomain = record.name req.RecordType = record.record_type req.RecordLine = record.line or "默认" req.Value = record.value req.TTL = record.ttl resp = self._call(client.CreateRecord, req, service="dnspod") return str(resp.RecordId) except TencentCloudSDKException as exc: raise _translate_sdk_exc(exc) from exc async def create_dns_record(self, record) -> str: return await asyncio.to_thread(self._create_dns_record_sync, record) def _update_dns_record_sync(self, record_id: str, zone_name: str, record) -> None: from tencentcloud.common.exception.tencent_cloud_sdk_exception import ( TencentCloudSDKException, ) from tencentcloud.dnspod.v20210323 import models as m try: client = _make_tc_client("dnspod", self._credentials) req = m.ModifyRecordRequest() req.Domain = zone_name req.RecordId = int(record_id) req.SubDomain = record.name req.RecordType = record.record_type req.RecordLine = record.line or "默认" req.Value = record.value req.TTL = record.ttl self._call(client.ModifyRecord, req, service="dnspod") except TencentCloudSDKException as exc: raise _translate_sdk_exc(exc) from exc async def update_dns_record(self, record_id: str, zone_name: str, record) -> None: await asyncio.to_thread(self._update_dns_record_sync, record_id, zone_name, record) def _delete_dns_record_sync(self, record_id: str, zone_name: str) -> None: from tencentcloud.common.exception.tencent_cloud_sdk_exception import ( TencentCloudSDKException, ) from tencentcloud.dnspod.v20210323 import models as m try: client = _make_tc_client("dnspod", self._credentials) req = m.DeleteRecordRequest() req.Domain = zone_name req.RecordId = int(record_id) self._call(client.DeleteRecord, req, service="dnspod") except TencentCloudSDKException as exc: raise _translate_sdk_exc(exc) from exc async def delete_dns_record(self, record_id: str, zone_name: str) -> None: await asyncio.to_thread(self._delete_dns_record_sync, record_id, zone_name) def _set_dns_record_status_sync(self, record_id: str, zone_name: str, enable: bool) -> None: """Flip DNSPod's per-record ``Status`` flag via the ModifyRecordStatus API. DNSPod accepts ``"ENABLE"`` / ``"DISABLE"`` (uppercase). Toggling a record's status is non-destructive (the record stays in the zone and keeps its RecordId) — distinct from delete + recreate which would mint a fresh RecordId and break our cmdb_dns_records FK. """ from tencentcloud.common.exception.tencent_cloud_sdk_exception import ( TencentCloudSDKException, ) from tencentcloud.dnspod.v20210323 import models as m try: client = _make_tc_client("dnspod", self._credentials) req = m.ModifyRecordStatusRequest() req.Domain = zone_name req.RecordId = int(record_id) req.Status = "ENABLE" if enable else "DISABLE" self._call(client.ModifyRecordStatus, req, service="dnspod") except TencentCloudSDKException as exc: raise _translate_sdk_exc(exc) from exc async def set_dns_record_status(self, record_id: str, zone_name: str, enable: bool) -> None: await asyncio.to_thread(self._set_dns_record_status_sync, record_id, zone_name, enable) # ── Load Balancers (CLB) ────────────────────────────────────────── def _list_load_balancers_sync(self) -> list[CloudLoadBalancer]: from tencentcloud.clb.v20180317 import models as m from tencentcloud.common.exception.tencent_cloud_sdk_exception import ( TencentCloudSDKException, ) page_size = 100 # CLB DescribeLoadBalancers max lbs: list[CloudLoadBalancer] = [] for region in self._regions: try: client = _make_tc_client("clb", self._credentials, region) offset = 0 while True: req = m.DescribeLoadBalancersRequest() req.Limit = page_size req.Offset = offset try: resp = self._call( client.DescribeLoadBalancers, req, service="clb", ) except TencentCloudSDKException as exc: code = str(getattr(exc, "code", "") or "") if "AuthFailure" in code: raise _translate_sdk_exc(exc) from exc # Same per-region tolerance as CVM — log + skip # this region on any non-auth SDK error, don't # abort the whole sweep. Round-22: record # partial_failures so safe_sweep doesn't treat # a region's missing LBs as deletions. logger.info( "Tencent CLB region %s skipped (%s): %s", region, code or "SDK", exc, ) self.partial_failures.append({ "leg": f"region:{region}", "capability": "load_balancers", "reason": f"inner_sdk:{code or type(exc).__name__}", "page_offset": offset, }) break batch = resp.LoadBalancerSet or [] for lb in batch: network_type = "internal" if lb.LoadBalancerType == "INTERNAL" else "public" lb_type = "application" if lb.Forward == 1 else "classic" ip = (lb.LoadBalancerVips or [""])[0] lbs.append(CloudLoadBalancer( lb_id=lb.LoadBalancerId, name=lb.LoadBalancerName, lb_type=lb_type, network_type=network_type, ip_address=ip, region=region, status="active" if lb.Status == 1 else "inactive", raw={ "LoadBalancerId": lb.LoadBalancerId, "LoadBalancerName": lb.LoadBalancerName, "LoadBalancerType": lb.LoadBalancerType, "Status": lb.Status, "Forward": lb.Forward, "VpcId": getattr(lb, "VpcId", None), "CreateTime": str(getattr(lb, "CreateTime", "")), }, )) if len(batch) < page_size: break offset += page_size except ProviderAuthError: raise except TencentCloudSDKException as exc: code = str(getattr(exc, "code", "") or "") if "AuthFailure" in code: raise _translate_sdk_exc(exc) from exc logger.info( "Tencent CLB region %s skipped (outer): %s", region, exc, ) # Round-22 audit: outer handshake failure must record # partial_failures for parity with CVM. self.partial_failures.append({ "leg": f"region:{region}", "capability": "load_balancers", "reason": f"outer_sdk:{code or type(exc).__name__}", }) return lbs async def list_load_balancers(self) -> list[CloudLoadBalancer]: return await asyncio.to_thread(self._list_load_balancers_sync) # ── CDN ────────────────────────────────────────────────────────── def _list_cdn_domains_sync(self) -> list[CloudCdnDomain]: from tencentcloud.cdn.v20180606 import models as m from tencentcloud.common.exception.tencent_cloud_sdk_exception import ( TencentCloudSDKException, ) page_size = 100 # DescribeDomainsConfig max domains: list[CloudCdnDomain] = [] try: client = _make_tc_client("cdn", self._credentials) offset = 0 while True: req = m.DescribeDomainsConfigRequest() req.Offset = offset req.Limit = page_size try: resp = self._call( client.DescribeDomainsConfig, req, service="cdn", ) except TencentCloudSDKException as exc: code = str(getattr(exc, "code", "") or "") if "ResourceNotFound" in code or "ServiceNotOpened" in code: return domains raise _translate_sdk_exc(exc) from exc batch = resp.Domains or [] for d in batch: origin = d.Origin source_type, source_address = "domain", "" if origin and origin.Origins: source_address = origin.Origins[0] origin_type = getattr(origin, "OriginType", "domain") source_type = ( "oss" if origin_type == "cos" else ("ip" if origin_type == "ip" else "domain") ) https_cfg = getattr(d, "Https", None) is_https = bool(https_cfg and getattr(https_cfg, "Switch", "off") == "on") status_map = {"online": "online", "offline": "offline", "processing": "configuring"} status = status_map.get(d.Status, "offline") domains.append(CloudCdnDomain( domain_name=d.Domain, cname=d.Cname or "", status=status, source_type=source_type, source_address=source_address, is_https=is_https, raw={ "Domain": d.Domain, "Cname": d.Cname, "Status": d.Status, "ServiceType": getattr(d, "ServiceType", None), "Area": getattr(d, "Area", None), }, )) if len(batch) < page_size: break offset += page_size except TencentCloudSDKException as exc: raise _translate_sdk_exc(exc) from exc return domains async def list_cdn_domains(self) -> list[CloudCdnDomain]: return await asyncio.to_thread(self._list_cdn_domains_sync) # ── Object Storage (COS v5) ──────────────────────────────────────── def _list_object_storage_sync(self) -> list[dict]: try: from qcloud_cos import CosConfig, CosS3Client except ImportError: logger.warning( "Tencent COS listing skipped — cos-python-sdk-v5 is not " "installed; add it to pyproject.toml", ) return [] try: from qcloud_cos.cos_exception import CosServiceError # type: ignore except ImportError: CosServiceError = Exception # type: ignore[assignment, misc] secret_id = self._credentials.fields.get("secret_id", "") secret_key = self._credentials.fields.get("secret_key", "") # The bucket-list endpoint (service.cos.myqcloud.com) is global, # but the SDK still needs *some* region string for endpoint # signing — the bucket record's ``Location`` carries the real region. config = CosConfig( Region=_COS_DEFAULT_REGION, SecretId=secret_id, SecretKey=secret_key, Scheme="https", ) client = CosS3Client(config) try: resp = client.list_buckets() except CosServiceError as exc: code = str(getattr(exc, "get_error_code", lambda: "")()) if code in ("InvalidAccessKeyId", "SignatureDoesNotMatch"): raise ProviderAuthError(redact_provider_error(exc)) from exc raise ProviderTransientError(redact_provider_error(exc)) from exc except Exception as exc: # noqa: BLE001 — COS SDK raises varied types if "AccessDenied" in str(exc) or "Unauthorized" in str(exc): raise ProviderAuthError(redact_provider_error(exc)) from exc raise ProviderTransientError(redact_provider_error(exc)) from exc buckets = (resp.get("Buckets") or {}).get("Bucket", []) if isinstance(buckets, dict): # single-bucket quirk buckets = [buckets] out = [] for b in buckets: out.append({ "bucket_name": b.get("Name", ""), "region": b.get("Location") or None, "storage_class": None, # per-bucket; needs GetBucketLifecycle "access_type": None, # per-bucket; needs GetBucketACL "status": "active", "size_bytes": None, "object_count": None, "raw": _json_safe(b), }) return out async def list_object_storage(self) -> list[dict]: return await asyncio.to_thread(self._list_object_storage_sync) # ── IAM (CAM users) ────────────────────────────────────────────── def _list_iam_users_sync(self) -> list[dict]: from tencentcloud.cam.v20190116 import models as m from tencentcloud.common.exception.tencent_cloud_sdk_exception import ( TencentCloudSDKException, ) client = _make_tc_client("cam", self._credentials) out: list[dict] = [] # CAM ``ListUsers`` does NOT accept ``Page`` / ``Rp`` parameters # despite the SDK request class exposing them. Setting either # field returns ``UnknownParameter `P``` (Tencent's error parser # mangles unknown field names by dropping leading characters, # same shape as the ``Month`` → ``Onth`` we hit on the billing # API). Verified empirically and in the sibling # ``integrations/drivers/tencent_cloud.py`` driver. The endpoint # returns the full tenant user list in a single response, so a # single call is sufficient. users: list = [] try: req = m.ListUsersRequest() # Defensive: older tencentcloud-sdk-python versions ship a # ``ListUsersRequest`` whose ``__init__`` populated ``_Page`` # / ``_Rp`` defaults (current SDK has an empty body). When # those are non-None ``_serialize`` writes them into the # request body and the API rejects with the misleading # ``UnknownParameter `P``` (Tencent's parser truncates field # names — same shape as the ``Month`` → ``Onth`` billing # bug). Force them to None so they're stripped at serialize # time, regardless of SDK version pinned on prod. for _attr in ("_Page", "_Rp"): if hasattr(req, _attr): setattr(req, _attr, None) resp = self._call(client.ListUsers, req) users = list(getattr(resp, "Data", None) or []) except TencentCloudSDKException as exc: raise _translate_sdk_exc(exc) from exc for u in users: user_name = str(getattr(u, "Name", "") or "") uin = getattr(u, "Uin", None) or getattr(u, "Uid", None) email = str(getattr(u, "Email", "") or "") # Policies + groups (best-effort; tolerate per-user 5xx) attached: list[dict] = [] try: # CAM ``ListAttachedUserAllPolicies`` REQUIRES ``Page`` # (≥1) and ``Rp`` (≤200). The prior code stripped them # based on an incorrect transfer of the ``ListUsers`` # rejection shape; production logged # ``MissingParameter Page`` continuously after that # change shipped (incident 2026-05-16, request id # 93e4e7a1-…). Page through with Rp=200 until either # the page is short or TotalNum is reached so users # with >200 attached policies don't silently truncate # — truncation in the resource-sync path produces an # ``cmdb_iam_users.attached_policies`` snapshot that # grants-sync drift then treats as authoritative. tc_uin = int(uin) if uin is not None else 0 page = 1 rp = 200 collected: list = [] total = 0 while True: pol_req = m.ListAttachedUserAllPoliciesRequest() pol_req.TargetUin = tc_uin pol_req.AttachType = 0 pol_req.Page = page pol_req.Rp = rp pol_resp = self._call( client.ListAttachedUserAllPolicies, pol_req, ) batch = list( getattr(pol_resp, "PolicyList", None) or [] ) collected.extend(batch) total = int(getattr(pol_resp, "TotalNum", 0) or 0) if len(batch) < rp or len(collected) >= total: break page += 1 if page > 10: logger.warning( "Tencent ListAttachedUserAllPolicies hit " "10-page cap for user %s (collected=%d, " "total=%d)", user_name, len(collected), total, ) break for p in collected: attached.append({ "name": str(getattr(p, "PolicyName", "") or ""), "arn": str(getattr(p, "PolicyId", "") or "inline"), }) except TencentCloudSDKException as exc: logger.info("Tencent ListAttachedUserAllPolicies(%s): %s", user_name, exc) groups: list[str] = [] try: # Same Page/Rp defensive stance — if Tencent rejects # them on ListAttachedUserAllPolicies the symmetric # endpoint is suspect too. Default page size returns # the user's groups in one shot for typical tenants. grp_req = m.ListGroupsForUserRequest() grp_req.Uid = int(uin) if uin is not None else 0 grp_resp = self._call(client.ListGroupsForUser, grp_req) for g in getattr(grp_resp, "GroupInfo", None) or []: groups.append(str(getattr(g, "GroupName", "") or "")) except TencentCloudSDKException as exc: logger.info("Tencent ListGroupsForUser(%s): %s", user_name, exc) # ── MFA (per sub-user, really) ──────────────────────── # # The previous comment here claimed CAM exposes per-user MFA # "only through SubAccountKey helpers that are subject to # change", and hard-coded False. That is wrong: # ``DescribeSafeAuthFlagColl`` takes a ``SubUin`` and returns # the sub-user's own ``LoginFlag`` / ``ActionFlag``. Verified # against prod 2026-08-12 — 冯帅 and 李琳 login+action, 左武辉 # login only, ``tiup-server`` none. So all 25 rows were # asserting "no MFA" about a question nobody had asked, and # the compliance view rendered that as fact. # # ``mfa_state`` carries the honest third answer. A probe # failure must NOT collapse to "disabled": "we could not # check" is a different remediation from "it is off". mfa_state = "unknown" mfa_detail: dict[str, Any] = {} if uin is not None: try: flag_req = m.DescribeSafeAuthFlagCollRequest() flag_req.SubUin = int(uin) flag_resp = self._call( client.DescribeSafeAuthFlagColl, flag_req, ) login_flag = _coerce_flag( getattr(flag_resp, "LoginFlag", None), ) action_flag = _coerce_flag( getattr(flag_resp, "ActionFlag", None), ) mfa_state = ( "enabled" if _any_factor_on(login_flag) else "disabled" ) mfa_detail = { "login": login_flag, "action": action_flag, # Sensitive-operation MFA is a separate control; # an admin with login-only MFA is a real finding # that a single bool cannot express. "action_mfa": _any_factor_on(action_flag), } except TencentCloudSDKException as exc: logger.info( "Tencent DescribeSafeAuthFlagColl(%s): %s", user_name, exc, ) mfa_enabled = mfa_state == "enabled" # ── Last console login ─────────────────────────────── # # ``ListUsers`` carries no login timestamp at all (verified # field-by-field against prod: Uin / Name / Uid / Remark / # ConsoleLogin / PhoneNum / CountryCode / Email / CreateTime # / NickName). The old code read ``LastLoginTime`` off this # response, which is why every Tencent row had # ``last_login_at = NULL`` and the inactivity sweep could # never see this tenant. ``GetUser`` is where it lives, as # ``RecentlyLoginTime``. last_login = None try: get_req = m.GetUserRequest() get_req.Name = user_name detail_resp = self._call(client.GetUser, get_req) last_login = _parse_tc_datetime( getattr(detail_resp, "RecentlyLoginTime", None), ) except TencentCloudSDKException as exc: logger.info("Tencent GetUser(%s): %s", user_name, exc) create_date = _parse_tc_datetime(getattr(u, "CreateTime", None)) console_login = getattr(u, "ConsoleLogin", None) out.append({ "user_name": user_name, "arn": f"qcs::cam::uin/{uin}:uin/{uin}" if uin else None, "user_id_external": str(uin) if uin is not None else None, "create_date": create_date, "last_login_at": last_login, "attached_policies": attached, "groups": groups, "mfa_enabled": mfa_enabled, "mfa_state": mfa_state, "tags": {}, "raw": _json_safe({ "Name": user_name, "Uin": uin, "Email": email, "Phone": str(getattr(u, "PhoneNum", "") or ""), # 0 = no console access at all, i.e. a service # account. The authoritative signal for that — far # better than guessing from the name. "ConsoleLogin": console_login, "Remark": str(getattr(u, "Remark", "") or ""), "CountryCode": str(getattr(u, "CountryCode", "") or ""), "MfaDetail": mfa_detail, }), }) return out async def list_iam_users(self) -> list[dict]: return await asyncio.to_thread(self._list_iam_users_sync) # ── Monthly billing via Billing API ───────────────────────── # # Tencent's Billing API exposes ``DescribeBillSummaryByProduct`` # which returns the SUMmed monthly cost broken out by product # (CVM, COS, CDN, …). # # **Currency is hard-coded to CNY, and that is correct today — but # only for a CN-site tenant.** The comment here used to claim we # it captured whatever ``Currency`` the API returns; it does not, and # it cannot: ``BusinessSummaryOverviewItem`` has no ``Currency`` field # at all (verified against the SDK model). The claim was actively # misleading — it reads as "this is already site-aware" (#2179). # # TODO(intl): Tencent Cloud International bills the SAME API in USD. # If this driver is ever pointed at an intl tenant, the hard-coded # CNY label makes the read side divide a USD figure by ~7.2 — a 7× # understatement with no guard anywhere. Gate the label on the # account's site (e.g. an explicit ``site``/``region_domain`` config # field) BEFORE onboarding one. # # Region for the billing client is irrelevant — the API is # global; we use ``ap-guangzhou`` arbitrarily. The SDK requires # a non-empty region string. def _get_monthly_billing_sync(self, month: date) -> dict: from tencentcloud.billing.v20180709 import models as m from tencentcloud.common.exception.tencent_cloud_sdk_exception import ( TencentCloudSDKException, ) month.replace(day=1).strftime("%Y-%m") # Last day of the cycle month — DescribeBillSummaryByProduct # requires the FULL month range, not a single day. Earlier # impl (PR #50) set BeginTime == EndTime + tried to pass an # unsupported ``Month`` parameter, which broke the API # ("UnknownParameter `Onth`") and would have returned only # one day's bill if the param was accepted. from datetime import timedelta as _td first_day = month.replace(day=1) if first_day.month == 12: next_first = first_day.replace(year=first_day.year + 1, month=1) else: next_first = first_day.replace(month=first_day.month + 1) last_day = next_first - _td(days=1) client = _make_tc_client( "billing", self._credentials, "ap-guangzhou", ) cost_by_service: dict[str, float] = {} total = 0.0 currency = "CNY" last_resp: Any = None try: req = m.DescribeBillSummaryByProductRequest() req.BeginTime = f"{first_day.isoformat()} 00:00:00" req.EndTime = f"{last_day.isoformat()} 23:59:59" try: resp = self._call( client.DescribeBillSummaryByProduct, req, service="billing", ) except TencentCloudSDKException as exc: code = str(getattr(exc, "code", "") or "") if "AuthFailure" in code or "Unauthorized" in code: raise _translate_sdk_exc(exc) from exc # Tencent returns FailedOperation.NotPayUser / # NoPayer when the secret-id account isn't the # paying account (financial-permission split). Surface # that as a clear ProviderTransientError so the admin # sees what to fix instead of a generic 500. raise ProviderTransientError( f"Tencent DescribeBillSummaryByProduct {code}: " f"{redact_provider_error(exc)}", ) from exc last_resp = resp # ``SummaryOverview`` is the per-product breakdown; rows # are TencentCloud SDK objects with attributes (not dict). for item in (getattr(resp, "SummaryOverview", []) or []): product = str( getattr(item, "BusinessCodeName", "") or getattr(item, "BusinessCode", "") or "Unknown", ) # ``RealTotalCost`` is the post-discount actual amount # the customer paid; ``TotalCost`` is the LIST price. # Use RealTotalCost for parity with the AWS unblended # cost (also actual-paid). # # Explicit None check, NOT ``a or b`` (#2179): the truthy # chain fell back to the list price whenever # RealTotalCost was absent OR legitimately "0", mixing a # list figure into the invoice measure — the exact # direction ``tencent_cloud.py`` forbids in writing. A # missing invoice figure is missing DATA; report it as # zero-plus-a-warning rather than silently substituting # a number that means something else. real = getattr(item, "RealTotalCost", None) if real is None: logger.warning( "Tencent billing: product %s has no RealTotalCost " "for %s — counting 0 rather than substituting the " "list price (TotalCost=%s)", product, first_day.isoformat(), getattr(item, "TotalCost", None), ) amt = float(str(real if real is not None else "0")) cost_by_service[product] = ( cost_by_service.get(product, 0.0) + amt ) total += amt finally: pass # Tencent SDK responses are class instances, not dicts — # ``_json_safe`` passes them through unchanged and SQLAlchemy # then trips on ``Object of type DescribeBillSummaryByProductResponse # is not JSON serializable``. Convert via the SDK's # ``to_json_string()`` (every TencentCloud SDK response has it), # parse back to dict so ``cmdb_billing_monthly.raw_response`` # gets a clean JSON-friendly snapshot. raw_response: dict[str, Any] = {} if last_resp is not None: try: import json as _json raw_response = _json.loads(last_resp.to_json_string()) except (AttributeError, ValueError): # Fallback: pull whatever attributes are exposed. raw_response = { "RequestId": getattr(last_resp, "RequestId", None), "SummaryOverview": [ { "BusinessCodeName": getattr(it, "BusinessCodeName", None), "RealTotalCost": getattr(it, "RealTotalCost", None), "TotalCost": getattr(it, "TotalCost", None), } for it in (getattr(last_resp, "SummaryOverview", []) or []) ], } return { "month": month.replace(day=1), # Tencent (CN) bills in CNY; this driver runs sync-in-thread and # has no DB session to resolve the FX rate, so it CANNOT convert # to USD itself (unlike the async Aliyun driver). Surface the # native CNY amount + ``currency='CNY'`` so the billing_sync # handler's CNY→USD rebase path (``app/scheduled_jobs/handlers.py`` # — keyed on ``currency=='CNY' and total_cost_native is not # None``) converts it with the platform FX rate. Without # ``total_cost_native`` that rebase never fired and the raw CNY # amount was persisted into ``total_cost_usd`` as if it were USD # — a ~7x overstatement in every multi-cloud USD rollup. # ``total_cost_usd`` stays the native value purely as the # handler's bounded fallback when FX resolution fails (it then # keeps ``currency='CNY'`` so value + label still agree). "total_cost_usd": total, "total_cost_native": total, "cost_by_service": cost_by_service, "currency": currency, "raw_response": raw_response, } async def get_monthly_billing(self, month: date | None = None) -> dict: from datetime import timedelta as _td target = month or (date.today().replace(day=1) - _td(days=1)) return await asyncio.to_thread(self._get_monthly_billing_sync, target) # ── Helpers ────────────────────────────────────────────────────────── def _server_from_instance(inst: Any, region: str) -> CloudServer: """Translate a Tencent CVM ``Instance`` SDK object into a CloudServer.""" status_map = { "RUNNING": "running", "STOPPED": "stopped", "STARTING": "running", "STOPPING": "stopped", "REBOOTING": "running", "SHUTDOWN": "stopped", "TERMINATING": "stopped", "PENDING": "running", "LAUNCH_FAILED": "stopped", } raw_status = str(getattr(inst, "InstanceState", "") or "").upper() status = status_map.get(raw_status, raw_status.lower() or "unknown") public_ips = list(getattr(inst, "PublicIpAddresses", None) or []) private_ips = list(getattr(inst, "PrivateIpAddresses", None) or []) tags = {} for t in getattr(inst, "Tags", None) or []: key = getattr(t, "Key", None) value = getattr(t, "Value", None) if key is not None: tags[str(key)] = str(value) if value is not None else "" os_raw = str(getattr(inst, "OsName", "") or "").lower() os_type = "windows" if "windows" in os_raw else ("linux" if os_raw else None) placement = getattr(inst, "Placement", None) zone = str(getattr(placement, "Zone", "") or "") if placement else None raw = _json_safe({ "InstanceId": getattr(inst, "InstanceId", None), "InstanceName": getattr(inst, "InstanceName", None), "InstanceType": getattr(inst, "InstanceType", None), "InstanceState": getattr(inst, "InstanceState", None), "CPU": getattr(inst, "CPU", None), "Memory": getattr(inst, "Memory", None), "ImageId": getattr(inst, "ImageId", None), "OsName": getattr(inst, "OsName", None), "PublicIpAddresses": public_ips, "PrivateIpAddresses": private_ips, "Zone": zone, "CreatedTime": str(getattr(inst, "CreatedTime", "") or ""), "ExpiredTime": str(getattr(inst, "ExpiredTime", "") or ""), "VirtualPrivateCloud": getattr(inst, "VirtualPrivateCloud", None), }) return CloudServer( instance_id=str(getattr(inst, "InstanceId", "") or ""), name=str(getattr(inst, "InstanceName", "") or getattr(inst, "InstanceId", "") or ""), status=status, region=region, zone=zone, vcpu=int(getattr(inst, "CPU", 0) or 0) or None, memory_mb=(int(getattr(inst, "Memory", 0) or 0) * 1024) or None, public_ips=public_ips, private_ips=private_ips, image_id=str(getattr(inst, "ImageId", "") or "") or None, os_type=os_type, tags=tags, raw=raw, )