"""Azure provider — Virtual Machines, Azure DNS, Storage Accounts, Load Balancers. One provider instance per Azure AD tenant (``integrations.provider='azure'``). All calls are OAuth2 client-credentials + ARM REST via ``httpx`` — we deliberately don't pull in the ``azure-mgmt-*`` SDKs (each one is tens of MB and drags in ``msrestazure``, ``cryptography`` pins, etc.) to keep the backend image lean. Capabilities: * SERVERS — ``Microsoft.Compute/virtualMachines`` + a best-effort instanceView lookup for power state + a machineType→vCPU map. * DNS_RECORDS — ``Microsoft.Network/dnsZones`` + recordSets per zone. Private DNS zones (``Microsoft.Network/privateDnsZones``) are also enumerated; most customers register public zones so the split is preserved in ``raw.zone_scope`` = ``public`` | ``private``. * OBJECT_STORAGE — ``Microsoft.Storage/storageAccounts`` (one row per storage account, not per container — blob containers live under the account and would 10× the cardinality without adding actionable info). * LOAD_BALANCERS — ``Microsoft.Network/loadBalancers`` + Application Gateway (``Microsoft.Network/applicationGateways``). We collapse them into one logical feed since Helix's cmdb_load_balancers table doesn't distinguish ARM sub-flavours. The driver fans out over every subscription the SP can see (via ``/subscriptions?api-version=2022-12-01``) with bounded concurrency; per-subscription 403s are logged and skipped rather than nuking the whole sync — same policy as GCP's per-project fan-out. Public IPs are not embedded on VM records — ARM returns a NIC-ref, and the NIC carries a public-IP-ref. We resolve them in a best-effort second pass (bounded parallelism) so the UI gets public IPs without forcing every sync to do 3 round trips per VM. On error we store the VM with an empty public IP list and log once — the inventory is still useful. """ from __future__ import annotations import asyncio import hashlib import logging from datetime import UTC, date, datetime, timedelta from typing import Any import httpx from app.core.exceptions import format_exception_for_log from app.domains.cmdb.providers.base import ( CloudDnsRecord, CloudLoadBalancer, CloudProvider, CloudResourceInventoryItem, CloudServer, ProviderCapability, ProviderCredentials, ServerMetricPoint, ServerMetricTarget, classify_resource_category, register_provider, ) from app.domains.cmdb.providers.exceptions import ( CapabilityNotSupported, ProviderAuthError, ProviderRateLimitError, ProviderTransientError, redact_provider_error, ) logger = logging.getLogger(__name__) _LOGIN_BASE = "https://login.microsoftonline.com" _ARM_BASE = "https://management.azure.com" # ARM API versions. These are pinned because ARM routes are versioned # and "latest" isn't a safe default — a new api-version can silently # change response shapes. _API_SUBSCRIPTIONS = "2022-12-01" _API_COMPUTE = "2023-09-01" # virtualMachines + instanceView _API_NETWORK = "2023-09-01" # NICs, public IPs, load balancers _API_COST_MANAGEMENT = "2023-11-01" # Cost Management query API _API_DNS = "2018-05-01" # dnsZones + recordSets (stable GA) _API_PRIVATE_DNS = "2020-06-01" # privateDnsZones _API_STORAGE = "2023-01-01" # storageAccounts _API_APPGW = "2023-09-01" # applicationGateways # Fan-out concurrency limits — friendly to ARM's per-subscription quota. _MAX_CONCURRENT_SUBSCRIPTIONS = 8 _MAX_CONCURRENT_VM_ENRICH = 16 # 5xx retry budget for ARM GETs. 3 total attempts (initial + 2 retries) # with exponential backoff 1s, 2s — keeps a single transient ARM 5xx # (the production-observed pm-validation-priv-vm101_z1 NIC fetch # pattern, 2026-05-24) from flipping the resource_sync cron to red. # Cron interval is 1h so a 3-second worst-case retry budget is # negligible relative to the cycle; longer-than-3-attempt outages # are genuine ARM service incidents that SHOULD surface as # partial_failure for admin attention. _DEFAULT_ARM_5XX_RETRIES = 2 @register_provider class AzureProvider(CloudProvider): """Azure CMDB data source.""" @classmethod def provider_type(cls) -> str: return "azure" @classmethod def supported_capabilities(cls) -> frozenset[ProviderCapability]: return frozenset({ ProviderCapability.SERVERS, ProviderCapability.DNS_RECORDS, ProviderCapability.OBJECT_STORAGE, ProviderCapability.LOAD_BALANCERS, ProviderCapability.METRICS, }) def __init__(self, credentials: ProviderCredentials) -> None: super().__init__(credentials) self._tenant_id = str(credentials.fields.get("tenant_id") or "").strip() self._client_id = str(credentials.fields.get("client_id") or "").strip() self._client_secret = str( credentials.fields.get("client_secret") or "", ).strip() if not self._tenant_id or not self._client_id or not self._client_secret: raise ProviderAuthError( "Azure provider missing tenant_id / client_id / client_secret", ) self._token_cache: dict[str, str] = {} self._client: httpx.AsyncClient | None = None self._resource_inventory_enabled = str( credentials.fields.get("resource_inventory_enabled") or "", ).strip().lower() in ("1", "true", "yes", "on") def _http(self) -> httpx.AsyncClient: # Reuse one client across a sync — each sync is a short-lived # call path, and keeping the client warm saves ~50ms / request # on TLS re-handshake. Closed via ``close()``. if self._client is None: self._client = httpx.AsyncClient(timeout=30.0, trust_env=False) return self._client async def close(self) -> None: if self._client is not None: try: await self._client.aclose() finally: self._client = None # ── Auth helpers ────────────────────────────────────────────── async def _token(self, resource: str) -> str: """OAuth2 bearer for the given resource base URL. ``resource`` is one of ``_ARM_BASE`` or ``https://graph.microsoft.com``. Tokens are cached per resource for the lifetime of this provider instance (at most one sync), which is safely inside the token TTL. """ if resource in self._token_cache: return self._token_cache[resource] url = f"{_LOGIN_BASE}/{self._tenant_id}/oauth2/v2.0/token" data = { "grant_type": "client_credentials", "client_id": self._client_id, "client_secret": self._client_secret, "scope": f"{resource}/.default", } client = self._http() # AAD token endpoint is the prerequisite for every ARM/Graph # call in this driver — a single transport flake here cascades # to every capability in the same sync. Wrap with the shared # transport-retry helper so ReadTimeout / ConnectError / # PoolTimeout get 5 attempts before surfacing. Symmetric with # the cost driver's ``_post_with_retry`` (fc08c02e). # 2026-05-24: ALSO retry 5xx (Microsoft's login.microsoftonline.com # very rarely 5xx's but when it does, the failure is correlated # across every retry-in-a-tight-loop call in the same sync — # bounded 5xx retry with exponential backoff costs ≤3s extra # when AAD is healthy and saves the entire sync's worth of # capabilities when AAD has a transient hiccup). resp: httpx.Response | None = None for attempt in range(_DEFAULT_ARM_5XX_RETRIES + 1): label = ( f"_token resource={resource}" if attempt == 0 else f"_token resource={resource} (5xx-retry {attempt}/{_DEFAULT_ARM_5XX_RETRIES})" ) resp = await _request_with_transport_retry( "POST", client, url, endpoint_label=label, data=data, headers={"Content-Type": "application/x-www-form-urlencoded"}, ) if resp.status_code < 500 or attempt >= _DEFAULT_ARM_5XX_RETRIES: break delay = min(2 ** attempt, 30) logger.warning( "Azure AAD token 5xx resource=%s attempt=%d/%d " "delay=%ds status=%d", resource, attempt + 1, _DEFAULT_ARM_5XX_RETRIES + 1, delay, resp.status_code, ) await asyncio.sleep(delay) assert resp is not None # noqa: S101 — loop guarantees assignment if resp.status_code in (400, 401, 403): # Pre-launch hardening 2026-05-16: ``resp.text`` from Azure # AAD on 400/401/403 frequently echoes the failing request # including ``client_id`` and ``client_secret`` form values # when the SP supplied a malformed secret — strip those # before letting the error string land in audit logs. raise ProviderAuthError( f"Azure token endpoint {resp.status_code}: " f"{redact_provider_error(resp.text)}", ) if resp.status_code != 200: raise ProviderTransientError( f"Azure token endpoint {resp.status_code} (after " f"{_DEFAULT_ARM_5XX_RETRIES + 1} attempts): " f"{redact_provider_error(resp.text)}", ) token = str(resp.json().get("access_token") or "") if not token: raise ProviderAuthError("Azure token response missing access_token") self._token_cache[resource] = token return token async def _arm_request( self, url: str, *, endpoint_label: str, params: dict[str, str] | None = None, token: str | None = None, ) -> httpx.Response: """Single source of truth for ARM HTTP GETs. Layered retry budget: * Transport-layer (5 attempts on ReadTimeout / ConnectError / PoolTimeout / RemoteProtocolError) via :func:`_request_with_transport_retry`. * 429: honour ``Retry-After`` once, capped at 60s. * 5xx: bounded retry with exponential backoff (``_DEFAULT_ARM_5XX_RETRIES`` attempts, ``2 ** attempt`` s up to 30 s). Pre-launch hardening 2026-05-24: prior to this consolidation ``_arm_get`` had only 429 retry and ``_arm_list_all``'s ``nextLink`` hop had ONLY transport retry. A single transient 5xx on a NIC GET (production incident: ARM returned 5xx on ``pm-validation-priv-vm101_z1``) surfaced as ``ProviderTransientError`` → ``partial_failure`` → cron run flipped red. Next sync self- heals but the hourly noise drowns real failures. This helper retries 5xx symmetric with how the cost driver and Aliyun provider handle transient server errors. Returns the final ``httpx.Response`` — does NOT raise on 4xx/5xx. The caller's status-code branch (4xx routing, 5xx → ProviderTransientError) keeps the same shape so per-capability ``skipped_legs`` / ``partial_failures`` classification is unchanged on budget exhaustion. """ if token is None: token = await self._token(_ARM_BASE) client = self._http() for attempt in range(_DEFAULT_ARM_5XX_RETRIES + 1): label = ( endpoint_label if attempt == 0 else f"{endpoint_label} (5xx-retry {attempt}/{_DEFAULT_ARM_5XX_RETRIES})" ) resp = await _request_with_transport_retry( "GET", client, url, endpoint_label=label, params=params, headers={"Authorization": f"Bearer {token}"}, ) if resp.status_code == 429: retry_after = int(resp.headers.get("Retry-After") or 1) # Release the 429 response's connection back to the pool # before sleeping — httpx Response holds the underlying # socket until ``.aread()`` / ``.aclose()`` runs; a long # Retry-After (up to 60s) blocking a held connection # would silently drain the pool under load. await resp.aclose() # ARM commonly returns Retry-After: 30/60 under throttling. # Capping at 10s and retrying immediately just gets re- # throttled; honour the server-stated cooldown up to 60s. await asyncio.sleep(min(retry_after, 60)) resp = await _request_with_transport_retry( "GET", client, url, endpoint_label=f"{endpoint_label} (post-429)", params=params, headers={"Authorization": f"Bearer {token}"}, ) if resp.status_code == 429: raise ProviderRateLimitError(retry_after=retry_after) # Round-30 code-review fix (2026-05-24): if the response is # 401 mid-retry-loop, the cached token has rotated/expired. # Without invalidating the cache here, every subsequent # `_arm_request` in this sync (typically dozens of NIC GETs # via `_resolve_nics`) reuses the stale token and hits 401 # in lockstep — the entire capability flips red even though # a single token refresh would recover it. Drop the cache # entry; the next `_token(_ARM_BASE)` call (either inside # this method via the optional `token=` param being None on # the next caller, or in `_arm_get` re-entry) fetches a # fresh one. We do NOT retry within `_arm_request` itself # on 401 — the caller's auth branch surfaces it as # ProviderAuthError, which propagates correctly. if resp.status_code == 401: self._token_cache.pop(_ARM_BASE, None) if resp.status_code >= 500 and attempt < _DEFAULT_ARM_5XX_RETRIES: delay = min(2 ** attempt, 30) logger.warning( "Azure ARM 5xx %s attempt=%d/%d delay=%ds status=%d", endpoint_label, attempt + 1, _DEFAULT_ARM_5XX_RETRIES + 1, delay, resp.status_code, ) # Release the intermediate 5xx response's connection # before the backoff sleep — same pool-hygiene rationale # as the 429 branch above. Status code was already read; # the discarded body is not needed for an attempt that # will be retried. (The FINAL 5xx response on budget # exhaustion is returned to the caller un-closed so it # can read ``.text`` for the error message.) await resp.aclose() await asyncio.sleep(delay) continue return resp # Loop is range(N+1) and the only continue branch is the # intermediate-5xx path above, which fires when # ``attempt < N`` — the final iteration where # ``attempt == N`` always falls through to ``return resp``. # This line is therefore unreachable, but ``mypy`` insists on a # terminal statement. raise AssertionError("_arm_request loop fell through") # pragma: no cover async def _arm_get( self, path: str, api_version: str, *, extra_params: dict[str, str] | None = None, ) -> dict[str, Any]: """GET against ARM with the standard api-version query + retries. Transport / 429 / 5xx retry budget all live in :meth:`_arm_request`. 4xx (other than 401/429) bubbles up via ``_raise_http`` so the per-capability handler can decide whether to skip (per-sub 403) or propagate (wrong resource ID on an enrich call). """ url = f"{_ARM_BASE}{path}" params = {"api-version": api_version} if extra_params: params.update(extra_params) resp = await self._arm_request( url, endpoint_label=f"_arm_get {path}", params=params, ) if resp.status_code == 401: raise ProviderAuthError( f"Azure ARM 401: {redact_provider_error(resp.text)}", ) if resp.status_code >= 500: raise ProviderTransientError( f"Azure ARM {resp.status_code} (after " f"{_DEFAULT_ARM_5XX_RETRIES + 1} attempts): " f"{redact_provider_error(resp.text)}", ) if resp.status_code >= 400: # 403 / 404 — bubble up so per-capability handlers can # decide whether to skip (per-sub 403) or propagate (wrong # resource ID on an enrich call). _raise_http(resp) return resp.json() async def _arm_list_all( self, path: str, api_version: str, *, extra_params: dict[str, str] | None = None, ) -> list[dict[str, Any]]: """List a paged ARM resource (follows ``nextLink``). Returns the flattened ``value`` arrays. Safe on empty responses. Pagination hops share the same transport + 429 + 5xx retry budget as the initial page via :meth:`_arm_request` — pre-fix a transient 5xx mid-pagination silently truncated the page boundary (transport retry was added in PR #477, but a successful TCP roundtrip returning a 5xx body still aborted). """ items: list[dict[str, Any]] = [] page = await self._arm_get(path, api_version, extra_params=extra_params) items.extend(page.get("value") or []) next_link = page.get("nextLink") while next_link: resp = await self._arm_request( next_link, endpoint_label="_arm_list_all nextLink", ) if resp.status_code >= 500: raise ProviderTransientError( f"Azure ARM {resp.status_code} on nextLink (after " f"{_DEFAULT_ARM_5XX_RETRIES + 1} attempts): " f"{redact_provider_error(resp.text)}", ) if resp.status_code >= 400: _raise_http(resp) page = resp.json() items.extend(page.get("value") or []) next_link = page.get("nextLink") return items # ── Subscription discovery + fan-out ───────────────────────── async def _list_subscription_ids(self) -> list[str]: items = await self._arm_list_all("/subscriptions", _API_SUBSCRIPTIONS) out: list[str] = [] for sub in items: state = str(sub.get("state", "Enabled")) if state.lower() != "enabled": continue sub_id = str(sub.get("subscriptionId") or "") if sub_id: out.append(sub_id) return out async def list_resource_inventory(self) -> list[CloudResourceInventoryItem]: if not self._resource_inventory_enabled: raise CapabilityNotSupported( "Azure tenant-wide resource inventory is disabled; enable it after " "the service principal has Reader access on intended subscriptions", ) async def per_subscription(sub_id: str) -> list[CloudResourceInventoryItem]: rows = await self._arm_list_all( f"/subscriptions/{sub_id}/resources", "2021-04-01", ) out: list[CloudResourceInventoryItem] = [] for row in rows: resource_id = str(row.get("id") or "").strip() resource_type = str(row.get("type") or "unknown").strip() if not resource_id: continue service = resource_type.split("/", 1)[0].removeprefix("Microsoft.").casefold() category, sensitivity = classify_resource_category(resource_type, service) parts = resource_id.split("/") resource_group = parts[parts.index("resourceGroups") + 1] if "resourceGroups" in parts else None properties: dict[str, Any] = ( row["properties"] if isinstance(row.get("properties"), dict) else {} ) out.append(CloudResourceInventoryItem( provider_resource_id=resource_id, resource_type=resource_type, service=service, category=category, name=str(row.get("name") or resource_id.rsplit("/", 1)[-1]), scope_external_id=sub_id, region=str(row.get("location") or "") or None, status=str(properties.get("provisioningState") or "") or None, tags=row.get("tags") if isinstance(row.get("tags"), dict) else None, metadata={ "resource_group": resource_group, "kind": row.get("kind"), "managed_by": row.get("managedBy"), "sku": row.get("sku"), "properties": properties, }, sensitivity=sensitivity, )) return out return await self._fan_out_per_subscription( per_subscription, capability="resource_inventory", ) async def _fan_out_per_subscription( self, fn: Any, # async callable: (sub_id) -> list[...] *, capability: str | None = None, ) -> list[Any]: """Concurrent per-subscription fan-out. ``capability`` (optional) is the capability label attached to every emitted ``partial_failures`` / ``skipped_legs`` leg entry so admin-readable surfaces can render which capability the leg belongs to. The NIC-level leg (see ``_resolve_nics``) already carries ``"capability": "servers"``; this parameter brings the subscription-level legs into the same shape — without it a servers-cron 403 and a dns_records-cron 403 surfaced as identical ``subscription:sub-xyz`` legs in the failure-notifier signature, collapsing two different failures into one alert. """ sub_ids = await self._list_subscription_ids() if not sub_ids: return [] sem = asyncio.Semaphore(_MAX_CONCURRENT_SUBSCRIPTIONS) def _attach_capability(entry: dict[str, Any]) -> dict[str, Any]: if capability and "capability" not in entry: entry["capability"] = capability return entry async def _one(sub_id: str) -> list[Any]: async with sem: try: return await fn(sub_id) except ProviderAuthError: # Global auth broken — nuke the whole sync so the # admin sees a clear error instead of empty results. raise except _HttpError as exc: # Per-subscription 4xx = SP can list the sub but # lacks resource-level access (common with ARM's # inherited-but-not-granted behaviour — e.g. 400 # ``subscription does not exist`` for subs the SP # sees through org visibility but can't enter). # Skip the sub and keep going rather than nuking # the whole fan-out. if 400 <= exc.status < 500: logger.info( "Azure per-subscription call skipped (sub=%s, " "status=%s): %s", sub_id, exc.status, redact_provider_error(exc.body, max_len=200), ) # Round-26 (2026-05-18): same Round-25 GCP fix # but for Azure — the ARM ``MissingSubscriptionRegistration`` # error code means the resource provider (e.g. # Microsoft.Compute) was never registered on # this subscription, so every ``virtualMachines`` # list call returns 4xx by design. Pre-fix the # driver bucketed this with "real RBAC gap" 403s # in ``partial_failures`` and a tenant with N # subs of M not-yet-registered providers paged # IT admin every cron tick. Route the known # admin-config-state errors to ``skipped_legs`` # (capability stays green; surfaced for # visibility) and leave real auth gaps in # ``partial_failures`` so the failure-notifier # still pages. skip_category = _classify_azure_4xx_admin_skip( exc.body, ) if skip_category is not None: category, detail = skip_category # Round-26 review P1-3: parallel diagnostic # info-log so an admin grepping worker logs # can find Azure skips by (category, detail) # the same way GCP skips can. The raw-body # log above this branch fires for ALL Azure # 4xx; this finer-grained one fires only # for skipped legs and is grep-friendly. logger.info( "Azure per-subscription call routed to " "skipped_legs (sub=%s, %s: %s)", sub_id, category, detail or "unknown", ) self.skipped_legs.append(_attach_capability({ "leg": f"subscription:{sub_id}", "reason": ( f"{category}:{detail or 'unknown'}" ), })) return [] leg_entry: dict[str, Any] = { "leg": f"subscription:{sub_id}", "reason": f"http_{exc.status}", } msg = _azure_4xx_error_message(exc.body) if msg: leg_entry["error_message"] = msg # Pre-launch hardening 2026-05-16: record the # skipped subscription so the sync engine # surfaces ``partial_failures`` in # ``capability_results``. Without this, a sync # that hits 403 on 3 of 20 subs reports # ``success`` even though 15% of resources # silently vanished from the upsert + sweep. self.partial_failures.append( _attach_capability(leg_entry), ) return [] raise ProviderTransientError(redact_provider_error(exc)) from exc except Exception as exc: # noqa: BLE001 logger.warning( "Azure per-subscription call failed (sub=%s): %s", sub_id, redact_provider_error(exc), ) self.partial_failures.append(_attach_capability({ "leg": f"subscription:{sub_id}", "reason": type(exc).__name__, })) return [] results = await asyncio.gather(*[_one(s) for s in sub_ids]) flat: list[Any] = [] for bundle in results: flat.extend(bundle) return flat # ── Servers (Compute VMs) ──────────────────────────────────── async def list_servers(self) -> list[CloudServer]: vm_sku_table = await self._build_vm_sku_table() return await self._fan_out_per_subscription( lambda sub_id: self._list_servers_for_subscription(sub_id, vm_sku_table), capability="servers", ) # ── Utilization metrics (Azure Monitor) ─────────────────────────── # # Azure Monitor's data-plane metrics API is per-resource (no batch), # so we call it once per VM (bounded concurrency) using the VM's ARM # resourceId — carried in ``target.raw["id"]`` because # ``cmdb_servers.instance_id`` for Azure is the ``vmId`` GUID, not the # ARM path. "Percentage CPU" + "Network In/Out Total" are emitted with # no in-guest agent. Memory would need "Available Memory Bytes" ÷ the # SKU's total to become a percentage and isn't universally present, so # it's left NULL here (agent-only, like the other clouds). _METRICS_API_VERSION = "2018-01-01" _METRICS_CONCURRENCY = 8 async def list_server_metrics( self, targets: list[ServerMetricTarget], *, lookback_days: int = 14, period_seconds: int = 3600, ) -> list[ServerMetricPoint]: arm_targets: list[tuple[str, str]] = [] for t in targets: arm_id = str((t.raw or {}).get("id") or "").strip() if arm_id and t.instance_id: arm_targets.append((arm_id, t.instance_id)) if not arm_targets: return [] end = datetime.now(UTC) start = end - timedelta(days=lookback_days) timespan = ( f"{start.strftime('%Y-%m-%dT%H:%M:%SZ')}/" f"{end.strftime('%Y-%m-%dT%H:%M:%SZ')}" ) interval = f"PT{max(1, period_seconds // 3600)}H" params = { "api-version": self._METRICS_API_VERSION, "metricnames": "Percentage CPU,Network In Total,Network Out Total", "aggregation": "Average,Maximum,Total", "timespan": timespan, "interval": interval, } sem = asyncio.Semaphore(self._METRICS_CONCURRENCY) points: list[ServerMetricPoint] = [] async def _one(arm_id: str, instance_id: str) -> None: async with sem: url = f"{_ARM_BASE}{arm_id}/providers/microsoft.insights/metrics" try: resp = await self._arm_request( url, endpoint_label="metrics", params=params, ) except ProviderRateLimitError: self.partial_failures.append({ "leg": f"vm:{instance_id}", "capability": "metrics", "reason": "rate_limited", }) return except Exception as exc: # noqa: BLE001 self.partial_failures.append({ "leg": f"vm:{instance_id}", "capability": "metrics", "reason": "transport", # Credential-safe: this leg lands in the admin-visible # scheduled-job result, so redact at source like the # inventory path (never format_exception_for_log raw). "error_message": redact_provider_error(exc, max_len=160), }) return if resp.status_code in (403, 404): self.partial_failures.append({ "leg": f"vm:{instance_id}", "capability": "metrics", "reason": str(resp.status_code), }) return if resp.status_code != 200: self.partial_failures.append({ "leg": f"vm:{instance_id}", "capability": "metrics", "reason": f"http_{resp.status_code}", }) return try: body = resp.json() except ValueError: return points.extend(_parse_azure_metrics(body, instance_id)) await asyncio.gather( *(_one(arm_id, iid) for arm_id, iid in arm_targets), ) return points async def _build_vm_sku_table(self) -> dict[str, tuple[int | None, int | None]]: """Fetch ``Microsoft.Compute/skus`` once per sync; map VM size name → (vCPU, memoryMB). Size names are tenant-global so one subscription's SKU table covers all subscriptions. Returns an empty dict if the call fails — VMs still sync, they just get ``vcpu=None`` / ``memory_mb=None``. """ try: sub_ids = await self._list_subscription_ids() except Exception: # noqa: BLE001 return {} if not sub_ids: return {} try: items = await self._arm_list_all( f"/subscriptions/{sub_ids[0]}/providers/Microsoft.Compute/skus", _API_COMPUTE, extra_params={"$filter": "resourceType eq 'virtualMachines'"}, ) except Exception as exc: # noqa: BLE001 logger.info("Azure compute SKU table lookup failed: %s", exc) return {} table: dict[str, tuple[int | None, int | None]] = {} for sku in items: name = str(sku.get("name") or "") if not name or name in table: continue caps = {c.get("name"): c.get("value") for c in sku.get("capabilities") or []} vcpu_raw = caps.get("vCPUs") or caps.get("vCPUsAvailable") mem_raw = caps.get("MemoryGB") vcpu = _safe_int(vcpu_raw) memory_mb = None mem_gb = _safe_float(mem_raw) if mem_gb is not None: memory_mb = int(mem_gb * 1024) table[name] = (vcpu, memory_mb) return table async def _list_servers_for_subscription( self, sub_id: str, sku_table: dict[str, tuple[int | None, int | None]], ) -> list[CloudServer]: vms = await self._arm_list_all( f"/subscriptions/{sub_id}/providers/Microsoft.Compute/virtualMachines", _API_COMPUTE, extra_params={"statusOnly": "false"}, ) if not vms: return [] nic_refs: set[str] = set() for vm in vms: for nic in ((vm.get("properties") or {}).get("networkProfile") or {}).get( "networkInterfaces", ) or []: ref = nic.get("id") if ref: nic_refs.add(str(ref)) nic_lookup = await self._resolve_nics(list(nic_refs)) power_state_lookup = await self._resolve_power_states(sub_id, vms) out: list[CloudServer] = [] for vm in vms: server = _vm_to_server(vm, sub_id, nic_lookup, power_state_lookup, sku_table) # ``None`` signals that one of the VM's NICs failed to # resolve — skip the upsert this sync so we don't overwrite # the existing row's IPs with empties. safe_sweep's prev- # success baseline + 95% floor preserve the DB row. if server is not None: out.append(server) return out async def _resolve_nics( self, nic_refs: list[str], ) -> dict[str, tuple[list[str], list[str]] | None]: """Map each NIC resource-id → (private_ips, public_ips), or ``None`` if the fetch failed. ARM doesn't return IPs inline on the VM record; we have to GET each NIC and, for NICs with a public IP ref, GET that too. To keep concurrency bounded we cap in-flight requests — VM fleets of 1k+ would otherwise flood ARM. Pre-launch hardening 2026-05-16: previously a failed NIC GET stored ``([], [])`` which silently overwrote the VM's existing IPs on upsert (data corruption). Now we store ``None`` so ``_vm_to_server`` can skip the VM entirely — combined with safe-sweep's prev-success baseline + 95% floor, the existing DB row's IPs survive until a clean sync re-reads them. """ if not nic_refs: return {} sem = asyncio.Semaphore(_MAX_CONCURRENT_VM_ENRICH) results: dict[str, tuple[list[str], list[str]] | None] = {} async def _one(nic_ref: str) -> None: async with sem: try: nic = await self._arm_get(nic_ref, _API_NETWORK) except Exception as exc: # noqa: BLE001 logger.info( "Azure NIC fetch failed (%s): %s", nic_ref, exc, ) results[nic_ref] = None self.partial_failures.append({ "leg": f"nic:{nic_ref}", "capability": "servers", "reason": f"nic_fetch_failed:{type(exc).__name__}", }) return private_ips: list[str] = [] public_ip_refs: list[str] = [] for cfg in (nic.get("properties") or {}).get( "ipConfigurations", ) or []: props = cfg.get("properties") or {} priv = props.get("privateIPAddress") if priv: private_ips.append(str(priv)) pip = props.get("publicIPAddress") or {} pip_id = pip.get("id") if pip_id: public_ip_refs.append(str(pip_id)) public_ips: list[str] = [] for pip_ref in public_ip_refs: try: pip_resp = await self._arm_get(pip_ref, _API_NETWORK) except Exception as exc: # noqa: BLE001 logger.info( "Azure public IP fetch failed (%s): %s", pip_ref, exc, ) continue addr = (pip_resp.get("properties") or {}).get("ipAddress") if addr: public_ips.append(str(addr)) results[nic_ref] = (private_ips, public_ips) await asyncio.gather(*[_one(r) for r in nic_refs]) return results async def _resolve_power_states( self, sub_id: str, vms: list[dict[str, Any]], ) -> dict[str, str]: """Best-effort per-VM power state via a single statuses call. Uses ``virtualMachines?$expand=instanceView`` when available; falls back to per-VM instanceView if the expanded list 403s. Returns a ``{vm_resource_id: power_state}`` map. Missing entries get the VM's ``provisioningState`` as a fallback. """ try: expanded = await self._arm_list_all( f"/subscriptions/{sub_id}/providers/Microsoft.Compute/virtualMachines", _API_COMPUTE, extra_params={"$expand": "instanceView", "statusOnly": "false"}, ) except Exception as exc: # noqa: BLE001 logger.info( "Azure VM instanceView fan-out failed (sub=%s): %s", sub_id, exc, ) return {} out: dict[str, str] = {} for vm in expanded: vm_id = str(vm.get("id") or "") iv = (vm.get("properties") or {}).get("instanceView") or {} for status in iv.get("statuses") or []: code = str(status.get("code") or "") if code.startswith("PowerState/"): out[vm_id] = code.split("/", 1)[1] break return out # ── Azure DNS + Private DNS ────────────────────────────────── async def list_dns_records(self) -> list[CloudDnsRecord]: return await self._fan_out_per_subscription( self._list_dns_records_for_subscription, capability="dns_records", ) async def _list_dns_records_for_subscription( self, sub_id: str, ) -> list[CloudDnsRecord]: out: list[CloudDnsRecord] = [] # Public DNS zones. try: public_zones = await self._arm_list_all( f"/subscriptions/{sub_id}/providers/Microsoft.Network/dnszones", _API_DNS, ) except _HttpError as exc: if exc.status in (403, 404): public_zones = [] else: raise for zone in public_zones: zone_name = str(zone.get("name") or "") zone_id = str(zone.get("id") or "") if not zone_id: continue try: records = await self._arm_list_all( f"{zone_id}/recordsets", _API_DNS, ) except _HttpError as exc: if exc.status in (403, 404): continue raise for rr in records: out.extend( _expand_dns_recordset( rr, zone_name, sub_id, zone_scope="public", ) ) # Private DNS zones — optional, 403s if SP doesn't have the role. try: private_zones = await self._arm_list_all( f"/subscriptions/{sub_id}/providers/Microsoft.Network/privateDnsZones", _API_PRIVATE_DNS, ) except _HttpError as exc: if exc.status in (403, 404): private_zones = [] else: raise for zone in private_zones: zone_name = str(zone.get("name") or "") zone_id = str(zone.get("id") or "") if not zone_id: continue for record_type in ( "A", "AAAA", "CNAME", "MX", "PTR", "SOA", "SRV", "TXT", ): try: records = await self._arm_list_all( f"{zone_id}/{record_type}", _API_PRIVATE_DNS, ) except _HttpError as exc: if exc.status in (403, 404): continue raise for rr in records: out.extend( _expand_private_dns_recordset( rr, zone_name, sub_id, record_type, ) ) return out # ── Storage (Storage Accounts) ─────────────────────────────── async def list_object_storage(self) -> list[dict[str, Any]]: return await self._fan_out_per_subscription( self._list_storage_for_subscription, capability="object_storage", ) async def _list_storage_for_subscription( self, sub_id: str, ) -> list[dict[str, Any]]: try: accounts = await self._arm_list_all( f"/subscriptions/{sub_id}/providers/Microsoft.Storage/storageAccounts", _API_STORAGE, ) except _HttpError as exc: if exc.status in (403, 404): return [] raise return [_storage_account_to_dict(a, sub_id) for a in accounts] # ── Load Balancers (LB + App Gateway) ──────────────────────── async def list_load_balancers(self) -> list[CloudLoadBalancer]: return await self._fan_out_per_subscription( self._list_lbs_for_subscription, capability="load_balancers", ) async def _list_lbs_for_subscription( self, sub_id: str, ) -> list[CloudLoadBalancer]: out: list[CloudLoadBalancer] = [] try: lbs = await self._arm_list_all( f"/subscriptions/{sub_id}/providers/Microsoft.Network/loadBalancers", _API_NETWORK, ) except _HttpError as exc: if exc.status in (403, 404): lbs = [] else: raise for lb in lbs: out.append(_lb_to_record(lb, sub_id, kind="loadBalancer")) try: app_gws = await self._arm_list_all( f"/subscriptions/{sub_id}/providers/Microsoft.Network/applicationGateways", _API_APPGW, ) except _HttpError as exc: if exc.status in (403, 404): app_gws = [] else: raise for gw in app_gws: out.append(_lb_to_record(gw, sub_id, kind="applicationGateway")) return out # ── Monthly billing via Cost Management ────────────────────── # # Cost Management is subscription-scoped — there's no tenant # aggregate endpoint. We POST one ``Usage`` query per enabled # subscription, group by ``ServiceName``, and SUM ``PreTaxCost``. # The breakdown in ``cost_by_service`` is summed across # subscriptions; ``raw_response.per_subscription`` preserves the # per-sub contribution so audit can drill in. # # Endpoint: # POST /subscriptions/{sub}/providers/Microsoft.CostManagement/query # ?api-version=2023-11-01 # Body shape mirrors the unified driver's # ``list_billing_by_dimension`` impl (PR #49). async def get_monthly_billing(self, month: date | None = None) -> dict: """Fetch one month's pretax cost across every Azure subscription. Cost Management's ``query`` endpoint: * is subscription-scoped (no tenant aggregate exists) * accepts ``timeframe=Custom`` with an ISO ``timePeriod`` * returns a ``properties.rows`` matrix whose columns are described by ``properties.columns`` — order matters * uses the subscription's billing currency (USD for most tenants but per-subscription override exists; we read currency from each row and pin the response currency to the FIRST non-empty value) Per-subscription errors (403 SP-not-authorized on a single sub, 404 Cost Management not enabled) are logged and skipped rather than nuking the whole call — partial result still beats empty. """ target = ( month or (date.today().replace(day=1) - timedelta(days=1)) ).replace(day=1) if target.month == 12: next_month = target.replace( year=target.year + 1, month=1, day=1, ) else: next_month = target.replace(month=target.month + 1, day=1) # Cost Management's ``to`` is INCLUSIVE; pass the last day of # the target month, not the first of the next month. end_inclusive = next_month - timedelta(days=1) body = { "type": "Usage", "timeframe": "Custom", "timePeriod": { "from": f"{target.isoformat()}T00:00:00+00:00", "to": f"{end_inclusive.isoformat()}T23:59:59+00:00", }, "dataset": { "granularity": "None", "aggregation": { "totalCost": { "name": "PreTaxCost", "function": "Sum", }, }, "grouping": [ {"type": "Dimension", "name": "ServiceName"}, ], }, } sub_ids = await self._list_subscription_ids() cost_by_service: dict[str, float] = {} total = 0.0 currency = "USD" currencies_seen: set[str] = set() per_subscription: dict[str, dict[str, Any]] = {} token = await self._token(_ARM_BASE) client = self._http() for sub_id in sub_ids: url = ( f"{_ARM_BASE}/subscriptions/{sub_id}" "/providers/Microsoft.CostManagement/query" ) try: resp = await _request_with_transport_retry( "POST", client, url, endpoint_label=f"get_monthly_billing sub={sub_id}", params={"api-version": _API_COST_MANAGEMENT}, headers={ "Authorization": f"Bearer {token}", "Content-Type": "application/json", }, json=body, ) except httpx.HTTPError as exc: # Transport retry exhausted (5 attempts) — record the # per-sub failure and continue with remaining subs so # one bad backend doesn't nuke the whole monthly aggregate. logger.warning( "Azure Cost Management call failed for sub %s: %s", sub_id, format_exception_for_log(exc, max_len=200), ) continue if resp.status_code == 429: retry_after = int(resp.headers.get("Retry-After") or 5) await asyncio.sleep(min(retry_after, 30)) resp = await _request_with_transport_retry( "POST", client, url, endpoint_label=f"get_monthly_billing sub={sub_id} (post-429)", params={"api-version": _API_COST_MANAGEMENT}, headers={ "Authorization": f"Bearer {token}", "Content-Type": "application/json", }, json=body, ) if resp.status_code == 401: raise ProviderAuthError( "Azure Cost Management 401: " f"{redact_provider_error(resp.text)}", ) if resp.status_code in (403, 404): logger.info( "Azure Cost Management %s on sub %s: %s", resp.status_code, sub_id, redact_provider_error(resp.text, max_len=200), ) continue if resp.status_code >= 500: raise ProviderTransientError( "Azure Cost Management 5xx: " f"{redact_provider_error(resp.text)}", ) if resp.status_code >= 400: logger.warning( "Azure Cost Management %s on sub %s: %s", resp.status_code, sub_id, resp.text[:200], ) continue payload = resp.json() props = payload.get("properties") or {} columns = props.get("columns") or [] rows = props.get("rows") or [] cost_idx = next( (i for i, c in enumerate(columns) if c.get("name", "").lower() == "pretaxcost"), None, ) svc_idx = next( (i for i, c in enumerate(columns) if c.get("name", "").lower() == "servicename"), None, ) cur_idx = next( (i for i, c in enumerate(columns) if c.get("name", "").lower() == "currency"), None, ) sub_total = 0.0 row_currencies: set[str] = set() for row in rows: # Bounds-check every positional index: Azure occasionally # returns a 200 whose ``rows`` entry is narrower than # ``columns`` (partial / aggregated responses). A bare # ``row[cost_idx]`` would then IndexError and crash the whole # monthly-billing aggregate mid-subscription. n = len(row) cost_cell = row[cost_idx] if (cost_idx is not None and cost_idx < n) else None try: amt = float(cost_cell) if cost_cell is not None else 0.0 except (TypeError, ValueError): amt = 0.0 svc_cell = row[svc_idx] if (svc_idx is not None and svc_idx < n) else None svc = str(svc_cell) if svc_cell is not None else "Unknown" cur_cell = row[cur_idx] if (cur_idx is not None and cur_idx < n) else None if cur_cell: row_currencies.add(str(cur_cell)) cost_by_service[svc] = ( cost_by_service.get(svc, 0.0) + amt ) sub_total += amt # Azure can enrol subs in EUR / USD / CNY (Mooncake) / # other currencies; if a single sub returns mixed-currency # rows we surface that loudly rather than letting the last # row's currency win as the dict's label. Aggregator is # responsible for FX-converting before summing. if row_currencies: if len(row_currencies) == 1: sub_currency = next(iter(row_currencies)) currencies_seen.add(sub_currency) currency = sub_currency else: currencies_seen.update(row_currencies) currency = "MIXED" per_subscription[sub_id] = { "subscription_id": sub_id, "total": sub_total, "rows": len(rows), "currency": ( next(iter(row_currencies)) if len(row_currencies) == 1 else ("MIXED" if row_currencies else None) ), } # Group per-subscription subtotals by their OWN currency. The # headline ``total`` must never be a raw cross-currency add: a # sub billed in CNY and one in USD would otherwise fold ~7×- # inflated CNY yuan straight into ``total_cost_usd`` (#1663). subtotal_by_currency: dict[str, float] = {} for entry in per_subscription.values(): cur = (entry.get("currency") or "USD").upper() subtotal_by_currency[cur] = ( subtotal_by_currency.get(cur, 0.0) + float(entry.get("total") or 0.0) ) # If subs span multiple currencies, ``currency`` is "MIXED" — # callers MUST consult ``raw_response.currencies`` for the # per-currency breakdown (cmdb_billing_monthly stores total in # the integration-level reporting currency, set by the admin # at integration creation). if len(currencies_seen) > 1: currency = "MIXED" if currency == "MIXED": # FX-convert each currency bucket to USD before summing. # ``app.core.fx_rate`` only knows CNY↔USD, so USD passes # through, CNY converts, and ANY other non-USD currency is # un-convertible → keep the existing MIXED / native-None # behaviour (do NOT silently sum) so downstream never trusts # a bogus USD headline. When every bucket IS convertible we # emit a real USD total and relabel the row USD. usd_total, all_converted = await self._sum_mixed_to_usd( subtotal_by_currency, ) if all_converted: total = usd_total currency = "USD" total_cost_native = None else: # At least one currency has no FX rate — fall back to the # loud MIXED signal. ``total`` stays the raw sum for # forensic ``raw_response`` inspection only; ``currency`` # stays MIXED and the handler treats native as None # (skips the monthly upsert rather than persisting a # cross-currency sum as USD). total = float(sum(subtotal_by_currency.values())) total_cost_native = None else: # Single-currency subscription set. Preserve prior behaviour. total = float(sum(subtotal_by_currency.values())) # Surface the native amount for a SINGLE non-USD currency so # the billing_sync handler's FX rebase can fire (it keys on # ``currency=='CNY' and total_cost_native is not None`` — # Azure China/Mooncake subs report CNY and would otherwise be # persisted as if USD, ~7x overstated; same bug class as # Tencent/UCloud). ``None`` for USD (already USD). total_cost_native = total if currency != "USD" else None return { "month": target, "total_cost_usd": total, "total_cost_native": total_cost_native, "cost_by_service": cost_by_service, "currency": currency, "raw_response": { "per_subscription": per_subscription, "subscription_count": len(sub_ids), "currencies": sorted(currencies_seen), "subtotal_by_currency": subtotal_by_currency, }, } async def _sum_mixed_to_usd( self, subtotal_by_currency: dict[str, float], ) -> tuple[float, bool]: """FX-convert per-currency subtotals to a single USD figure. Returns ``(usd_total, all_converted)``. ``all_converted`` is ``False`` when any bucket's currency has no resolvable FX rate (``app.core.fx_rate`` only supports CNY↔USD) — the caller then keeps the loud MIXED / native-None behaviour rather than trusting a partial sum. Resolves the CNY rate at most once, against the platform setting (no per-integration override is available in the provider layer — the platform weekly-refreshed rate is the correct fallback).""" usd_total = 0.0 cny_per_usd: float | None = None for cur, amt in subtotal_by_currency.items(): if cur == "USD": usd_total += amt continue if cur == "CNY": if cny_per_usd is None: from app.core import fx_rate as _fx_rate from app.core.database import async_session_factory try: async with async_session_factory() as fx_session: cny_per_usd = await _fx_rate.resolve_cny_per_usd( fx_session, ) except RuntimeError as exc: logger.warning( "Azure get_monthly_billing: CNY→USD rate " "unavailable for MIXED aggregate (%s); keeping " "MIXED", exc, ) return usd_total, False usd_total += amt / cny_per_usd continue # Unknown currency (EUR, GBP, …) — no FX rate. Fail closed. logger.warning( "Azure get_monthly_billing: no FX rate for currency %r in " "MIXED aggregate; keeping MIXED (native None)", cur, ) return usd_total, False return usd_total, True # ── HTTP error plumbing ─────────────────────────────────────────── class _HttpError(Exception): """Internal error carrying the ARM HTTP status + body.""" def __init__(self, status: int, body: str) -> None: super().__init__(f"ARM {status}: {body[:200]}") self.status = status self.body = body def _raise_http(resp: httpx.Response) -> None: raise _HttpError(resp.status_code, resp.text) # ── HTTP transport-retry primitive ─────────────────────────────────── # # Combined transport-layer retry for every outbound request the CMDB # resource-sync Azure driver does. Mirrors the cost driver's # ``_post_with_retry`` (azure_cloud.py, fc08c02e) but kept module-local # here because this driver's call sites raise Provider* exceptions on # status codes (handled by each caller) and only need transport-layer # retry from the helper — status-code retries are kept inline so each # call site can preserve its existing 429 / 5xx semantics. # # Why this is needed: # Pre-fix every outbound request in this driver — ``_token``, # ``_arm_get``, the ``_arm_list_all`` nextLink hop, and the cost-mgmt # POSTs in ``get_monthly_billing`` — relied on httpx raising a status # code we could classify. Transport-layer exceptions # (``httpx.ReadTimeout``, ``ConnectError``, ``PoolTimeout``, # ``RemoteProtocolError``, ``WriteError``, ``ReadError``, # ``ConnectTimeout``, ``WriteTimeout``) raise BEFORE a status code # exists. Pre-fix those exceptions bypassed every retry path and # propagated straight to ``_fan_out_per_subscription``'s ``except # Exception`` fallback (line 353), where they landed in # ``partial_failures`` and one cron tick had to wait for the next # tick to retry — exactly the prod 2026-05-24 incident with the # ``object_storage`` leg ReadTimeout on subscription # ``d3a296ee-...``. # # Retry-budget: 5 attempts, exponential backoff ``min(2 ** attempt, 60)`` # seconds. Matches the cost driver's helper (fc08c02e) so a permanently # failing endpoint is detected within ~30s rather than retrying for # minutes — the failure-notifier still fires on real outages, just one # tick later rather than every tick. # # Status-code (429/5xx) retries are NOT done here — each call site # already does its own (``_arm_get`` honours Retry-After up to 60s for # 429; ``get_monthly_billing`` does a single 429-retry). Centralising # both would force every call site to a single status policy, which # this driver intentionally varies (ARM vs Cost Management have # different throttle envelopes). _TRANSPORT_RETRY_EXCEPTIONS: tuple[type[BaseException], ...] = ( # ConnectError covers DNS failure + TCP RST + TLS handshake. httpx.ConnectError, httpx.ConnectTimeout, # ReadError + ReadTimeout cover mid-stream connection loss after # the request was sent — Azure ARM endpoints occasionally drop the # socket instead of returning a status (504 silent path). The # prod 2026-05-24 ``object_storage`` failure raised # ``httpx.ReadTimeout``. httpx.ReadError, httpx.ReadTimeout, httpx.WriteError, httpx.WriteTimeout, # PoolTimeout fires when our local connection pool can't lease a # socket within the timeout — usually a sign upstream is slow and # earlier requests are still in flight. httpx.PoolTimeout, # RemoteProtocolError covers HTTP/1.1 + HTTP/2 framing violations # from misconfigured intermediate proxies. httpx.RemoteProtocolError, ) _DEFAULT_TRANSPORT_RETRY_ATTEMPTS = 5 async def _request_with_transport_retry( method: str, client: httpx.AsyncClient, url: str, *, endpoint_label: str, max_attempts: int = _DEFAULT_TRANSPORT_RETRY_ATTEMPTS, **request_kwargs: Any, ) -> httpx.Response: """Issue an HTTP request with transport-layer retry. Returns the final ``httpx.Response`` regardless of status code — the caller handles status semantics (this helper only retries on transport exceptions, not 4xx/5xx). Re-raises the LAST transport exception once the budget is exhausted, preserving the original traceback so :func:`format_exception_for_log` surfaces the real httpx exception class in failure-notifier text (the observability fix from PR #465 — do NOT wrap in a generic RuntimeError here). ``method`` is the upper-case HTTP verb (``"GET"`` / ``"POST"``). Additional keyword arguments are forwarded to ``client.request`` verbatim (``params=``, ``headers=``, ``json=``, ``data=``). ``endpoint_label`` is a short, log-safe description of the call site (e.g. ``"_arm_get /subscriptions/abc"``) — used in retry log lines so an admin grepping worker output can locate the failing leg. """ # Dispatch via ``getattr(client, "get"/"post")`` rather than # ``client.request(method, ...)`` so existing test fakes that only # stub ``get`` / ``post`` keep working without each fake needing a # ``.request`` shim. Real ``httpx.AsyncClient`` exposes both forms. verb = method.lower() attempt = 0 while True: try: return await getattr(client, verb)(url, **request_kwargs) except _TRANSPORT_RETRY_EXCEPTIONS as exc: if attempt >= max_attempts - 1: # Budget exhausted — re-raise the ORIGINAL exception so # the caller's failure path / failure-notifier sees the # real ``httpx.ReadTimeout`` / ``ConnectError`` class # (PR #465 observability guarantee). raise delay = min(2 ** attempt, 60) logger.warning( "Azure HTTP 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 # ── Azure 4xx classifier (Round-26, 2026-05-18) ───────────────────── # # Same kind of fix Round-25 shipped for GCP, applied to Azure: ARM # returns 4xx for both "real RBAC gap" and "resource provider not # registered on this subscription" / "subscription disabled" / # "subscription in deletion". The latter group is admin-controlled # state that cron retries won't fix, so we route them to ``skipped_legs`` # instead of letting them flood ``partial_failures`` and page IT admin # every cron tick. # # Recognized ``error.code`` values (well-documented by ARM): # * ``MissingSubscriptionRegistration`` — the resource provider (e.g. # ``Microsoft.Compute``) was never registered on this subscription; # listing always returns 404/409. Admin must run # ``az provider register --namespace Microsoft.Compute``. # * ``SubscriptionNotFound`` — the SP sees the sub through tenant # visibility but the sub was moved / deleted under it. # * ``DisabledSubscription`` — billing disabled; the sub still # enumerates but every resource call returns 4xx. # * ``SubscriptionDoesNotExist`` — close variant of NotFound emitted # by some ARM endpoints. # # Anything else (real ``AuthorizationFailed`` / ``Forbidden``) stays # in ``partial_failures`` so the failure-notifier surfaces the real # RBAC gap. _AZURE_BODY_PARSE_MAX_BYTES = 64 * 1024 _AZURE_ADMIN_SKIP_CODES = { "MissingSubscriptionRegistration": "rp_not_registered", "SubscriptionNotFound": "sub_not_found", "DisabledSubscription": "sub_disabled", "SubscriptionDoesNotExist": "sub_not_found", } _AZURE_SAFE_TOKEN_RE = __import__("re").compile(r"[^a-zA-Z0-9._/-]") def _parse_azure_error_body(body: str) -> dict[str, Any] | None: """Parse the ``{"error": {"code": ..., "message": ...}}`` envelope. Returns the inner ``error`` dict, or None if the body isn't a JSON ARM error envelope or is too large to parse safely. """ if not body: return None if len(body) > _AZURE_BODY_PARSE_MAX_BYTES: return None try: import json as _json # local import keeps cold-start light payload = _json.loads(body) except (TypeError, ValueError): return None if not isinstance(payload, dict): return None error = payload.get("error") if isinstance(error, dict): return error return None def _classify_azure_4xx_admin_skip( body: str, ) -> tuple[str, str | None] | None: """Return ``(category, detail)`` when the ARM body looks like an "admin-config-state" 4xx (resource provider not registered, sub deleted, sub disabled, etc.), else None. Detail carries the resource provider namespace when the body includes it under ``error.details[*].target`` (current ARM shape) so the admin sees ``rp_not_registered:Microsoft.Compute`` rather than ``rp_not_registered:unknown``. Sanitized through :data:`_AZURE_SAFE_TOKEN_RE` before returning. """ error = _parse_azure_error_body(body) if error is None: return None code = error.get("code") if not isinstance(code, str): return None category = _AZURE_ADMIN_SKIP_CODES.get(code) if category is None: return None detail: str | None = None # ARM ``MissingSubscriptionRegistration`` puts the RP namespace in # ``error.message`` and sometimes ``error.details[0].target``. details = error.get("details") if isinstance(details, list): for det in details: if not isinstance(det, dict): continue target = det.get("target") if isinstance(target, str) and target: cleaned = _AZURE_SAFE_TOKEN_RE.sub("", target)[:128] if cleaned: detail = cleaned break if detail is None: # Fall back to parsing the namespace out of the message: ARM # writes "The subscription is not registered to use namespace # 'Microsoft.Compute'". Grab the first ``Microsoft.*`` token. message = error.get("message") if isinstance(message, str): import re as _re m = _re.search(r"\b(Microsoft\.[A-Za-z0-9]+)\b", message) if m: detail = m.group(1) return (category, detail) def _azure_4xx_error_message(body: str) -> str | None: """Return the redacted ``error.message`` from an ARM 4xx body, capped at 160 chars (same budget as GCP). Used to populate the optional ``error_message`` leg field so admins can read what ARM actually said without grepping logs. """ error = _parse_azure_error_body(body) if error is None: return None msg = error.get("message") if not isinstance(msg, str) or not msg.strip(): return None return redact_provider_error(msg, max_len=160) or None # ── VM → CloudServer ────────────────────────────────────────────── _POWER_STATE_MAP = { "running": "running", "starting": "running", "stopping": "stopped", "stopped": "stopped", "deallocating": "stopped", "deallocated": "stopped", "unknown": "unknown", } def _vm_to_server( vm: dict[str, Any], sub_id: str, nic_lookup: dict[str, tuple[list[str], list[str]] | None], power_state_lookup: dict[str, str], sku_table: dict[str, tuple[int | None, int | None]], ) -> CloudServer | None: # ``id`` is the full ARM path (~180 chars), which overflows the # ``cmdb_servers.instance_id`` VARCHAR(128) column. Use # ``properties.vmId`` — a 36-char GUID Azure assigns at VM creation # that's globally unique and stable for the VM's lifetime. Align # with AWS (``i-…``), GCP (numeric id), Aliyun (``i-…``) which all # use a short provider-native identifier. # # Fallback: if ``vmId`` is unexpectedly missing (shouldn't happen # for normal VMs), hash the ARM path to a stable 40-char SHA1 — # still unique, still fits, but not human-readable. The full ARM # path stays in ``raw`` either way for debugging. arm_id = str(vm.get("id") or "") props = vm.get("properties") or {} vm_guid = str(props.get("vmId") or "").strip() if vm_guid: instance_id = vm_guid elif arm_id: instance_id = hashlib.sha1(arm_id.encode(), usedforsecurity=False).hexdigest() else: instance_id = arm_id # last resort — likely empty vm_id = arm_id # keep the ARM path for NIC/power-state lookups name = str(vm.get("name") or "") location = str(vm.get("location") or "") hardware = props.get("hardwareProfile") or {} vm_size = str(hardware.get("vmSize") or "") storage = props.get("storageProfile") or {} os_disk = storage.get("osDisk") or {} os_type_raw = str(os_disk.get("osType") or "").lower() or None if os_type_raw == "windows": os_type: str | None = "windows" elif os_type_raw: os_type = "linux" else: os_type = None image_ref = storage.get("imageReference") or {} image_id = ( image_ref.get("id") or "/".join( str(image_ref.get(k) or "") for k in ("publisher", "offer", "sku", "version") if image_ref.get(k) ) or None ) private_ips: list[str] = [] public_ips: list[str] = [] for nic_ref in (props.get("networkProfile") or {}).get( "networkInterfaces", ) or []: ref = str(nic_ref.get("id") or "") if not ref: continue # ``None`` ≠ "no IPs" — it means the NIC fetch failed. Skip # the whole VM rather than write empty IPs over a healthy row. nic_entry = nic_lookup.get(ref, ([], [])) if nic_entry is None: return None p_priv, p_pub = nic_entry private_ips.extend(p_priv) public_ips.extend(p_pub) power = power_state_lookup.get(vm_id) or str( props.get("provisioningState") or "", ).lower() status = _POWER_STATE_MAP.get(power.lower(), power.lower() or "unknown") vcpu, memory_mb = sku_table.get(vm_size, (None, None)) tags_raw = vm.get("tags") or {} tags = {str(k): str(v) for k, v in tags_raw.items()} raw = dict(vm) raw["_subscription_id"] = sub_id raw["_power_state"] = power return CloudServer( instance_id=instance_id, name=name, status=status, region=location, zone=_first_zone(vm), vcpu=vcpu, memory_mb=memory_mb, public_ips=public_ips, private_ips=private_ips, image_id=str(image_id) if image_id else None, os_type=os_type, tags=tags, raw=raw, ) def _first_zone(vm: dict[str, Any]) -> str | None: zones = vm.get("zones") or [] if zones: return str(zones[0]) return None def _parse_azure_metrics( body: dict[str, Any], instance_id: str, ) -> list[ServerMetricPoint]: """Turn an Azure Monitor ``metrics`` response into per-timestamp points for one VM. Maps ``Percentage CPU`` → cpu (avg/max) and ``Network In/Out Total`` → net bytes (the ``total`` aggregation).""" merged: dict[Any, dict[str, float]] = {} for metric in body.get("value") or []: name = str(((metric.get("name") or {}).get("value")) or "") series = metric.get("timeseries") or [] if not series: continue for dp in series[0].get("data") or []: ts_raw = dp.get("timeStamp") if not ts_raw: continue try: ts = datetime.fromisoformat(str(ts_raw).replace("Z", "+00:00")) except ValueError: continue bucket = merged.setdefault(ts, {}) if name == "Percentage CPU": if dp.get("average") is not None: bucket["cpu"] = float(dp["average"]) if dp.get("maximum") is not None: bucket["cpu_max"] = float(dp["maximum"]) elif name == "Network In Total": if dp.get("total") is not None: bucket["net_in"] = float(dp["total"]) elif name == "Network Out Total": if dp.get("total") is not None: bucket["net_out"] = float(dp["total"]) points: list[ServerMetricPoint] = [] for ts, f in merged.items(): if not f: continue points.append(ServerMetricPoint( instance_id=instance_id, timestamp=ts, cpu_pct=f.get("cpu"), cpu_pct_max=f.get("cpu_max"), net_in_bytes=f.get("net_in"), net_out_bytes=f.get("net_out"), )) return points # ── DNS record expansion ────────────────────────────────────────── # Public DNS record types carry per-type arrays (aRecords, aaaaRecords, etc.) # We walk known keys and emit one Helix row per value so the sync engine # can dedupe by (zone, name, type, value). _PUBLIC_DNS_VALUE_EXTRACTORS: list[tuple[str, str, Any]] = [ ("aRecords", "A", lambda v: v.get("ipv4Address")), ("aaaaRecords", "AAAA", lambda v: v.get("ipv6Address")), ("mxRecords", "MX", lambda v: f"{v.get('preference')} {v.get('exchange')}"), ("nsRecords", "NS", lambda v: v.get("nsdname")), ("ptrRecords", "PTR", lambda v: v.get("ptrdname")), ("srvRecords", "SRV", lambda v: ( f"{v.get('priority')} {v.get('weight')} " f"{v.get('port')} {v.get('target')}" )), ("txtRecords", "TXT", lambda v: " ".join(v.get("value") or [])), ("caaRecords", "CAA", lambda v: ( f"{v.get('flags')} {v.get('tag')} \"{v.get('value')}\"" )), ] def _expand_dns_recordset( rr: dict[str, Any], zone_name: str, sub_id: str, zone_scope: str, ) -> list[CloudDnsRecord]: out: list[CloudDnsRecord] = [] props = rr.get("properties") or {} rr_name = str(rr.get("name") or "") rr_id = str(rr.get("id") or "") ttl = int(props.get("TTL") or props.get("ttl") or 0) # CNAME records use ``cnameRecord`` singular. cname = (props.get("cnameRecord") or {}).get("cname") if cname: out.append(CloudDnsRecord( zone_name=zone_name, record_id=f"{rr_id}::CNAME::{cname}", name=rr_name, record_type="CNAME", value=str(cname), ttl=ttl, line=None, status="enable", raw={"subscription_id": sub_id, "zone_scope": zone_scope, "_raw": rr}, )) soa = props.get("soaRecord") if soa: value = ( f"{soa.get('host')} {soa.get('email')} " f"{soa.get('serialNumber')} {soa.get('refreshTime')} " f"{soa.get('retryTime')} {soa.get('expireTime')} " f"{soa.get('minimumTTL')}" ) out.append(CloudDnsRecord( zone_name=zone_name, record_id=f"{rr_id}::SOA", name=rr_name, record_type="SOA", value=value, ttl=ttl, line=None, status="enable", raw={"subscription_id": sub_id, "zone_scope": zone_scope, "_raw": rr}, )) for key, rtype, extract in _PUBLIC_DNS_VALUE_EXTRACTORS: for entry in props.get(key) or []: value = extract(entry) if not value: continue out.append(CloudDnsRecord( zone_name=zone_name, record_id=f"{rr_id}::{rtype}::{value}", name=rr_name, record_type=rtype, value=str(value), ttl=ttl, line=None, status="enable", raw={ "subscription_id": sub_id, "zone_scope": zone_scope, "_raw": rr, }, )) return out def _expand_private_dns_recordset( rr: dict[str, Any], zone_name: str, sub_id: str, record_type: str, ) -> list[CloudDnsRecord]: # Private DNS recordsets are keyed by type in the URL, and the # response carries the type-specific values inline — schema is # otherwise identical to public. return _expand_dns_recordset( rr, zone_name, sub_id, zone_scope="private", ) # ── Storage account → dict ──────────────────────────────────────── def _storage_account_to_dict( acct: dict[str, Any], sub_id: str, ) -> dict[str, Any]: props = acct.get("properties") or {} sku = acct.get("sku") or {} access = "private" if props.get("allowBlobPublicAccess"): access = "public" kind = str(acct.get("kind") or "") return { "bucket_name": str(acct.get("name") or ""), "region": str(acct.get("location") or "").lower() or None, "storage_class": str(sku.get("name") or "") or None, "access_type": access, "size_bytes": None, # needs Azure Monitor metrics — separate job "object_count": None, "status": "active" if props.get("provisioningState") == "Succeeded" else "provisioning", "owner_raw": f"subscription:{sub_id}", "raw": {**acct, "_kind": kind, "_subscription_id": sub_id}, } # ── Load balancer → record ──────────────────────────────────────── def _lb_to_record( lb: dict[str, Any], sub_id: str, kind: str, ) -> CloudLoadBalancer: props = lb.get("properties") or {} frontends = props.get("frontendIPConfigurations") or [] ip_address = "" network_type = "internal" for fe in frontends: fe_props = fe.get("properties") or {} pip = fe_props.get("publicIPAddress") or {} if pip.get("id"): network_type = "public" addr = fe_props.get("privateIPAddress") or (pip.get("id") and "") if addr: ip_address = str(addr) break lb_type = "application" if kind == "applicationGateway" else ( (lb.get("sku") or {}).get("name", "classic").lower() or "classic" ) if lb_type not in {"classic", "application", "network"}: lb_type = "network" if kind == "loadBalancer" else "application" return CloudLoadBalancer( lb_id=str(lb.get("id") or ""), name=str(lb.get("name") or ""), lb_type=lb_type, network_type=network_type, ip_address=ip_address, region=str(lb.get("location") or ""), status=str(props.get("provisioningState") or "unknown").lower(), raw={**lb, "_kind": kind, "_subscription_id": sub_id}, ) # ── Small parsing helpers ───────────────────────────────────────── def _safe_int(value: Any) -> int | None: if value is None: return None try: return int(str(value)) except (TypeError, ValueError): return None def _safe_float(value: Any) -> float | None: if value is None: return None try: return float(str(value)) except (TypeError, ValueError): return None