"""AWS provider — EC2 instances + Route53 registered domains.""" from __future__ import annotations import asyncio from datetime import UTC, date, datetime, timedelta from decimal import Decimal from typing import Any try: import boto3 from botocore.config import Config as BotoConfig from botocore.exceptions import BotoCoreError, ClientError, NoCredentialsError _BOTO3_AVAILABLE = True except ImportError: boto3 = None # type: ignore[assignment] BotoConfig = None # type: ignore[assignment, misc] BotoCoreError = Exception # type: ignore[assignment, misc] ClientError = Exception # type: ignore[assignment, misc] NoCredentialsError = Exception # type: ignore[assignment, misc] _BOTO3_AVAILABLE = False # Pre-launch hardening 2026-05-16: boto3's default ``read_timeout=60s`` # combined with infinite TCP-level retries means a hung regional # endpoint can pin an ``asyncio.to_thread`` worker thread indefinitely. # ``_CAPABILITY_SYNC_TIMEOUT_S`` in sync.py cancels the asyncio task # but the underlying socket keeps running — thread-pool leaks accumulate # per cron tick. Every client built off ``AWSProvider._session`` inherits # this Config (see __init__ wrapping) so connect/read both have hard # caps and retries are bounded. _AWS_CLIENT_CONFIG_KWARGS: dict[str, Any] = { "connect_timeout": 10, "read_timeout": 30, "retries": {"max_attempts": 3, "mode": "standard"}, } def _json_safe(obj: Any) -> Any: """Recursively convert boto3 response objects into JSON-serializable values. boto3 responses frequently contain ``datetime`` instances (e.g. ``LaunchTime``, ``ExpirationDate``) which are not JSON serializable. We convert those to ISO strings so the sync engine can persist them as ``raw_data``. """ if isinstance(obj, datetime): return obj.isoformat() if isinstance(obj, 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 from app.domains.cmdb.providers.base import ( CloudCdnDomain, CloudCertificate, CloudDnsRecord, CloudDomain, CloudLoadBalancer, CloudProvider, CloudResourceInventoryItem, CloudServer, ProviderCapability, ProviderCredentials, ServerMetricPoint, ServerMetricTarget, classify_resource_category, register_provider, ) from app.domains.cmdb.providers.exceptions import ( CapabilityNotSupported, ProviderAuthError, ProviderTransientError, redact_provider_error, ) # Pre-launch hardening 2026-05-16: boto3 ``ClientError`` / ``NoCredentialsError`` # / ``ParamValidationError`` stringifications can include path hints # (``~/.aws/credentials``), full IAM ARNs (revealing account id + user # name), and on some failure modes the failing request snippet which # may carry a signed URL or X-Amz-* token. Every ``raise ProviderError(...)`` # in this module funnels through ``_err`` so the message is redacted + # length-bounded before it propagates to: # * ``cmdb_sync_jobs.capability_results['error']`` (DB-persisted) # * the admin-facing sync-history UI (any cloud_admin can read) # * Feishu failure notifications # * log aggregator retention # Aliyun's classic provider has its own stricter sign-URL redactor; # this is the AWS-specific entry point. def _err(prefix: str, exc: object) -> str: """Compose a short, credential-safe error prefix for AWS provider raises.""" return f"{prefix}: {redact_provider_error(exc)}" @register_provider class AWSProvider(CloudProvider): @classmethod def provider_type(cls) -> str: return "aws" @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, ProviderCapability.METRICS, }) def __init__(self, credentials: ProviderCredentials) -> None: if not _BOTO3_AVAILABLE: raise RuntimeError("boto3 is not installed. Install it with: pip install boto3") super().__init__(credentials) self._default_region = credentials.fields.get("default_region", "us-east-1") # Wrap session.client so every client (ec2/route53/iam/ce/s3/elbv2/ # cloudfront/route53domains) inherits the timeout + retry Config. # Callers that already pass ``config=`` keep their override. self._client_config = BotoConfig(**_AWS_CLIENT_CONFIG_KWARGS) self._session = self._wrap_client_timeout(boto3.Session( aws_access_key_id=credentials.fields["access_key_id"], aws_secret_access_key=credentials.fields["secret_access_key"], region_name=self._default_region, )) # ── AWS Organizations cross-account fan-out (opt-in) ────────── # When ``org_fanout`` is on, servers + metrics enumerate EVERY # active member account of the AWS Organization by STS-assuming # ``org_member_role`` (default OrganizationAccountAccessRole) into # each — the management/payer account itself usually runs no EC2 # (all compute lives in member accounts). Off by default: a plain # single-account integration keeps its current behaviour. f = credentials.fields self._org_fanout = str(f.get("org_fanout") or "").strip().lower() in ( "1", "true", "yes", "on", ) self._org_member_role = ( (f.get("org_member_role") or "").strip() or "OrganizationAccountAccessRole" ) # Region denylist — opt-in regions that are enabled on some account # but hold no resources and whose endpoint is unreachable from the # deploy host (e.g. me-south-1). Scanning them yields nothing but a # per-tick connect-timeout ``partial_failure`` that pages IT for # noise. Excluded from region discovery entirely. Accepts a comma- # or whitespace-separated string, or a list. Case-insensitive. raw_deny = f.get("region_denylist") or "" if isinstance(raw_deny, (list, tuple, set)): deny_iter = [str(x) for x in raw_deny] else: deny_iter = str(raw_deny).replace(",", " ").split() self._region_denylist = {r.strip().lower() for r in deny_iter if r.strip()} self._resource_inventory_enabled = str( f.get("resource_inventory_enabled") or "", ).strip().lower() in ("1", "true", "yes", "on") self._resource_explorer_region = ( str(f.get("resource_explorer_region") or "").strip() or self._default_region ) def _wrap_client_timeout(self, session: Any) -> Any: """Make ``session.client(...)`` default to the timeout+retry Config so every client built off this session (incl. assumed-role member sessions) inherits the bounded connect/read timeouts.""" orig_client = session.client def _client_with_timeout(*args: Any, **kwargs: Any) -> Any: kwargs.setdefault("config", self._client_config) return orig_client(*args, **kwargs) session.client = _client_with_timeout # type: ignore[assignment] return session # ── Organizations fan-out helpers ───────────────────────────────── _ORG_FANOUT_CONCURRENCY = 8 # STS AssumeRole denial codes that mean "this member account is # structurally out of reach" (no assumable member role — typical of # accounts INVITED into the org rather than created by it) rather than # a transient error. Routed to skipped_legs, not partial_failures, so # they never page the failure-notifier every sync tick. _ASSUME_ROLE_SKIP_CODES = frozenset({"AccessDenied", "AccessDeniedException"}) def _list_member_accounts_sync(self) -> list[tuple[str, str]]: """Active member accounts of the org, EXCLUDING the management account (which the base session already covers). Returns ``[(account_id, name), ...]``. Needs ``organizations:ListAccounts`` on the management/delegated credential.""" mgmt = self._verify_caller_identity_sync().get("account_id") org = self._session.client("organizations", region_name="us-east-1") out: list[tuple[str, str]] = [] for page in org.get_paginator("list_accounts").paginate(): for a in page.get("Accounts", []): if a.get("Status") == "ACTIVE" and str(a.get("Id")) != mgmt: out.append((str(a["Id"]), str(a.get("Name") or a["Id"]))) return out def _assume_session_sync(self, account_id: str) -> Any: """STS-assume ``org_member_role`` in ``account_id`` and return a timeout-wrapped boto3 Session on the temp creds. Raises on denial (caller records a partial_failure / skipped_leg).""" sts = self._session.client("sts", region_name="us-east-1") arn = f"arn:aws:iam::{account_id}:role/{self._org_member_role}" creds = sts.assume_role( RoleArn=arn, RoleSessionName="helix-cmdb-efficiency", DurationSeconds=3600, )["Credentials"] return self._wrap_client_timeout(boto3.Session( aws_access_key_id=creds["AccessKeyId"], aws_secret_access_key=creds["SecretAccessKey"], aws_session_token=creds["SessionToken"], region_name=self._default_region, )) def _verify_caller_identity_sync(self) -> dict[str, str]: """Return ``{"account_id": ..., "arn": ..., "user_id": ...}``. ``sts:GetCallerIdentity`` is the cheapest credential probe AWS offers: it requires no IAM permission (every caller can call it), burns no service-specific quota, and returns enough to verify both that the AK pair is live AND that the bound account matches the integration's expected identity. Surfaces ``ProviderAuthError`` on credential failure so test_connection / sync can short-circuit before the expensive list_servers fan- out across every region. """ try: sts = self._session.client("sts", region_name="us-east-1") resp = sts.get_caller_identity() return { "account_id": str(resp.get("Account") or ""), "arn": str(resp.get("Arn") or ""), "user_id": str(resp.get("UserId") or ""), } except NoCredentialsError as exc: raise ProviderAuthError( _err("AWS credential error", exc) ) from exc except ClientError as exc: raise ProviderAuthError(_err("AWS STS error", exc)) from exc except BotoCoreError as exc: # Network / DNS / TLS failures raise BotoCoreError, not # ClientError. Surface as transient so the caller (sync # engine retry / test_connection short-circuit) treats it # as a retryable infrastructure issue rather than a # credential failure. raise ProviderTransientError( _err("AWS STS transient error", exc) ) from exc async def verify_caller_identity(self) -> dict[str, str]: """Async wrapper around ``_verify_caller_identity_sync``.""" return await asyncio.to_thread(self._verify_caller_identity_sync) def _ec2_regions_sync(self, session: Any = None) -> list[str]: session = session or self._session try: ec2 = session.client("ec2", region_name="us-east-1") resp = ec2.describe_regions(Filters=[{"Name": "opt-in-status", "Values": ["opted-in", "opt-in-not-required"]}]) regions = [r["RegionName"] for r in resp["Regions"]] if self._region_denylist: regions = [r for r in regions if r.lower() not in self._region_denylist] return regions except NoCredentialsError as exc: raise ProviderAuthError(_err("AWS credential error", exc)) from exc except ClientError as exc: code = exc.response.get("Error", {}).get("Code") if code in ("AuthFailure", "InvalidClientTokenId"): raise ProviderAuthError(redact_provider_error(exc)) from exc # The IAM principal lacks ``ec2:DescribeRegions`` (e.g. an # ACM-only / narrowly-scoped key used for the certificates # capability). Region discovery is the gateway for EVERY # regional capability (servers / dns / certificates / …), so # this is a structural permission gap, not a blip. Classify # it as ``ProviderAuthError`` — the same call the lifecycle # path makes (``_ec2_lifecycle_sync``) and for the same # reason the route53domains handler spells out: routing a # permanent permission gap to ``ProviderTransientError`` # loops the sync engine forever marking the capability # transient with no actionable fix, and pages the # failure-notifier on every cron tick. Surfacing it as a # failed auth error gives the admin the one actionable # signal: grant ``ec2:DescribeRegions`` on the IAM policy. if code in ("UnauthorizedOperation", "AccessDenied", "AccessDeniedException"): raise ProviderAuthError(redact_provider_error(exc)) from exc # Round-22 audit: NEVER fall back to a hardcoded region # list. The previous behaviour ("return us-east-1 / us- # west-2 / ap-southeast-1 / ap-northeast-1 / eu-west-1 # when DescribeRegions throws") silently truncated tenants # whose primary footprint is in CN, ME, or AFR (regions # never in the fallback). Their EC2 / SG / EIP / EBS rows # would be invisible to Helix on every sync, and the safe- # sweep would treat the absent rows as deletions if the # cumulative drop crossed the 95% floor. Always raise on # non-auth failures so the sync is honestly marked failed # and mark-and-sweep skipped. raise ProviderTransientError(redact_provider_error(exc)) from exc except BotoCoreError as exc: # Round-22: DNS / TLS / network blip during describe_regions # was not previously handled — boto3 raises BotoCoreError # (NOT ClientError) for those. _verify_caller_identity_sync # already wraps this kind; same shape here. raise ProviderTransientError(_err("AWS region-discovery transient", exc)) from exc def _list_servers_sync(self, session: Any = None) -> list[CloudServer]: session = session or self._session servers: list[CloudServer] = [] for region in self._ec2_regions_sync(session): try: ec2 = session.client("ec2", region_name=region) paginator = ec2.get_paginator("describe_instances") for page in paginator.paginate(): for reservation in page["Reservations"]: for inst in reservation["Instances"]: if inst["State"]["Name"] == "terminated": continue public_ips = [] if inst.get("PublicIpAddress"): public_ips = [inst["PublicIpAddress"]] private_ips = [inst["PrivateIpAddress"]] if inst.get("PrivateIpAddress") else [] tags = {t["Key"]: t["Value"] for t in inst.get("Tags", [])} name = tags.get("Name", inst["InstanceId"]) status_map = {"running": "running", "stopped": "stopped"} status = status_map.get(inst["State"]["Name"], inst["State"]["Name"]) platform = inst.get("Platform", "") os_type = "windows" if platform.lower() == "windows" else "linux" cpu_opts = inst.get("CpuOptions", {}) or {} core_count = cpu_opts.get("CoreCount") threads_per_core = cpu_opts.get("ThreadsPerCore") if core_count is not None and threads_per_core is not None: vcpu = int(core_count) * int(threads_per_core) elif core_count is not None: vcpu = int(core_count) else: vcpu = None servers.append(CloudServer( instance_id=inst["InstanceId"], name=name, status=status, region=region, zone=inst.get("Placement", {}).get("AvailabilityZone"), vcpu=vcpu, memory_mb=None, # not in EC2 describe_instances directly public_ips=public_ips, private_ips=private_ips, image_id=inst.get("ImageId"), os_type=os_type, tags=tags, raw=_json_safe(inst), )) except NoCredentialsError as exc: raise ProviderAuthError(_err("AWS credential error", exc)) from exc except ClientError as e: code = e.response["Error"]["Code"] if code in ("AuthFailure", "InvalidClientTokenId"): raise ProviderAuthError(redact_provider_error(e)) # Round-26 (2026-05-18): route AWS ``OptInRequired`` # (region not opted-in on the account — common for # me-south-1 / af-south-1 / ap-east-1 / eu-south-1 etc.) # to ``skipped_legs`` rather than ``partial_failures``. # Same Round-25 rationale as the GCP API-disabled fix: # region not opted-in is admin state, retrying the cron # won't change it, and a tenant with all 30+ AWS regions # described as opt-in would flood IT admin with daily # "OptInRequired × 6" pages. Real auth gaps # (``UnauthorizedOperation`` / ``AccessDenied``) stay in # partial_failures so the failure-notifier still pages. # Note: ``UnauthorizedOperation`` is intentionally NOT # routed here — its message can contain "region not # enabled" text on some endpoints, but boto3 doesn't # surface a stable substring we can pattern-match, so # we keep it as a real failure where the redacted # ``error_message`` lets the admin diagnose. if code == "OptInRequired": self.skipped_legs.append({ "leg": f"region:{region}", "capability": "servers", "reason": "region_not_opted_in", }) continue # Region not available / API disabled / per-region 403: # skip the region but log it as a partial failure so the # sync engine surfaces ``X regions skipped`` on the job # row — pre-launch hardening 2026-05-16. Previously a # quietly-failing region just dropped its instances from # the upsert and the safe-sweep treated it as "real # deletions" if the drop crossed the 95% floor. leg_entry: dict[str, Any] = { "leg": f"region:{region}", "capability": "servers", "reason": str(code), } err_msg = (e.response.get("Error") or {}).get("Message") if isinstance(err_msg, str) and err_msg.strip(): leg_entry["error_message"] = redact_provider_error( err_msg, max_len=160, ) self.partial_failures.append(leg_entry) continue except BotoCoreError as exc: # Round-27 pre-launch (2026-05-25): per-region transport # / DNS / TLS hiccup used to abort the whole loop and # discard every region's instances collected so far. # ``_ec2_regions_sync`` catches BotoCoreError at the # region-discovery layer but not per-region. Record as # ``partial_failure`` and continue so the sync engine # skips safe-sweep for this tick and the other regions # still land. self.partial_failures.append({ "leg": f"region:{region}", "capability": "servers", "reason": "transport_error", "error_message": redact_provider_error( str(exc), max_len=160, ), }) continue return servers async def list_servers(self) -> list[CloudServer]: """Run boto3 (sync) in a worker thread so the FastAPI event loop isn't blocked. With ``org_fanout`` on, enumerate every active member account of the AWS Organization (STS-assume ``org_member_role`` into each) plus the management account itself, concurrently and with per-account failure isolation — one denied/unreachable account records a partial_failure and drops to ``[]`` rather than aborting the whole sync. Each server's ``raw`` is stamped with ``_aws_account_id`` so ``list_server_metrics`` re-assumes the right account when querying CloudWatch. """ if not self._org_fanout: return await asyncio.to_thread(self._list_servers_sync) try: members = await asyncio.to_thread(self._list_member_accounts_sync) except (BotoCoreError, ClientError) as exc: raise ProviderAuthError(_err("AWS Organizations ListAccounts", exc)) from exc sem = asyncio.Semaphore(self._ORG_FANOUT_CONCURRENCY) async def _mgmt() -> list[CloudServer]: # Management/payer account itself (no assume-role needed); it # usually runs 0 EC2 but we include it for completeness. async with sem: try: return await asyncio.to_thread(self._list_servers_sync) except ( ProviderAuthError, ProviderTransientError, BotoCoreError, ClientError, ) as exc: self.partial_failures.append({ "leg": "account:management", "capability": "servers", "reason": "management_list_error", "error_message": redact_provider_error(str(exc), max_len=160), }) return [] async def _member(account_id: str, name: str) -> list[CloudServer]: async with sem: try: msession = await asyncio.to_thread( self._assume_session_sync, account_id, ) except ClientError as exc: code = (exc.response.get("Error") or {}).get("Code", "") if code in self._ASSUME_ROLE_SKIP_CODES: # The member account has no assumable ``org_member_role`` # (accounts INVITED into the org — as opposed to # created by it — don't get the built-in # ``OrganizationAccountAccessRole``) or its trust # policy doesn't admit us. This is stable admin # state: retrying every tick won't change it, and # a org with N invited accounts would otherwise # flood the failure-notifier with N ``AccessDenied`` # legs on every sync. Route to skipped_legs (admin- # visible, capability stays green, no page) — same # rationale as OptInRequired regions / GCP projects # without the API enabled. Roll a member role out to # those accounts (or a HelixCloudReadOnly StackSet) # and they light up automatically next sync. self.skipped_legs.append({ "leg": f"account:{account_id}", "capability": "servers", "reason": "member_role_unavailable", }) return [] self.partial_failures.append({ "leg": f"account:{account_id}", "capability": "servers", "reason": "assume_role_failed", "error_message": redact_provider_error(str(exc), max_len=160), }) return [] except BotoCoreError as exc: # Transport/DNS/TLS hiccup assuming the role — transient, # keep as a partial_failure so the sync skips safe-sweep # this tick and retries next. self.partial_failures.append({ "leg": f"account:{account_id}", "capability": "servers", "reason": "assume_role_transport_error", "error_message": redact_provider_error(str(exc), max_len=160), }) return [] try: srv = await asyncio.to_thread(self._list_servers_sync, msession) except ( ProviderAuthError, ProviderTransientError, BotoCoreError, ClientError, ) as exc: self.partial_failures.append({ "leg": f"account:{account_id}", "capability": "servers", "reason": "list_error", "error_message": redact_provider_error(str(exc), max_len=160), }) return [] for s in srv: if isinstance(s.raw, dict): s.raw["_aws_account_id"] = account_id s.raw["_aws_account_name"] = name return srv results = await asyncio.gather(_mgmt(), *[_member(a, n) for a, n in members]) out: list[CloudServer] = [] for r in results: out.extend(r) return out def _list_resource_inventory_sync( self, session: Any | None = None, *, account_id: str | None = None, ) -> list[CloudResourceInventoryItem]: if not self._resource_inventory_enabled: raise CapabilityNotSupported( "AWS Resource Explorer inventory is disabled; enable it after an " "aggregator index and read-only view are configured", ) active_session = session or self._session try: client = active_session.client( "resource-explorer-2", region_name=self._resource_explorer_region, ) out: list[CloudResourceInventoryItem] = [] token: str | None = None while True: request: dict[str, Any] = {"QueryString": "*", "MaxResults": 1000} if token: request["NextToken"] = token response = client.search(**request) for item in response.get("Resources") or []: resource_id = str(item.get("Arn") or "").strip() resource_type = str(item.get("ResourceType") or "unknown").strip() if not resource_id: continue service = ( resource_type.split("::", 2)[1].casefold() if "::" in resource_type else "other" ) category, sensitivity = classify_resource_category(resource_type, service) out.append(CloudResourceInventoryItem( provider_resource_id=resource_id, resource_type=resource_type, service=service, category=category, name=resource_id.rsplit("/", 1)[-1].rsplit(":", 1)[-1] or resource_id, scope_external_id=str(item.get("OwningAccountId") or account_id or "") or None, region=str(item.get("Region") or "") or None, metadata={ "properties": item.get("Properties") or [], "last_reported_at": item.get("LastReportedAt"), }, sensitivity=sensitivity, )) token = str(response.get("NextToken") or "") or None if not token: return out except ClientError as exc: code = str((exc.response.get("Error") or {}).get("Code") or "") message = _err("AWS Resource Explorer", exc) if code in {"AccessDenied", "AccessDeniedException", "UnauthorizedException"}: raise ProviderAuthError(message) from exc if code in {"ResourceNotFoundException", "ValidationException"}: raise CapabilityNotSupported( "AWS Resource Explorer aggregator index/default view is not configured", ) from exc raise ProviderTransientError(message) from exc except BotoCoreError as exc: raise ProviderTransientError(_err("AWS Resource Explorer", exc)) from exc async def list_resource_inventory(self) -> list[CloudResourceInventoryItem]: if not self._resource_inventory_enabled: raise CapabilityNotSupported( "AWS Resource Explorer inventory is disabled; enable it after an " "aggregator index and read-only view are configured", ) if not self._org_fanout: return await asyncio.to_thread(self._list_resource_inventory_sync) try: members = await asyncio.to_thread(self._list_member_accounts_sync) management_id = await asyncio.to_thread( lambda: self._verify_caller_identity_sync().get("account_id"), ) except (BotoCoreError, ClientError) as exc: raise ProviderAuthError(_err("AWS Organizations ListAccounts", exc)) from exc sem = asyncio.Semaphore(self._ORG_FANOUT_CONCURRENCY) async def scan(account_id: str, session: Any | None) -> list[CloudResourceInventoryItem]: async with sem: try: active = session or await asyncio.to_thread(self._assume_session_sync, account_id) return await asyncio.to_thread( self._list_resource_inventory_sync, active, account_id=account_id, ) except CapabilityNotSupported: self.skipped_legs.append({ "leg": f"account:{account_id}", "capability": "resource_inventory", "reason": "resource_explorer_not_configured", }) return [] except Exception as exc: # noqa: BLE001 - isolate account legs self.partial_failures.append({ "leg": f"account:{account_id}", "capability": "resource_inventory", "reason": "inventory_failed", "error_message": redact_provider_error(str(exc), max_len=160), }) return [] tasks = [scan(str(management_id or "management"), self._session)] tasks.extend(scan(account_id, None) for account_id, _ in members) groups = await asyncio.gather(*tasks) return [item for group in groups for item in group] # ── Utilization metrics (CloudWatch) ────────────────────────────── # # CPU + network land on every EC2 instance with no in-guest agent # (namespace ``AWS/EC2``). Memory and disk *usage* require the # CloudWatch agent (``CWAgent`` namespace, per-tenant metric/dimension # naming) and are intentionally left ``None`` here — the efficiency # engine treats missing memory as "unknown", never as idle. Reads only # ``cloudwatch:GetMetricData``. _METRIC_SPECS = ( ("cpu", "CPUUtilization", "Average"), ("cpumax", "CPUUtilization", "Maximum"), ("netin", "NetworkIn", "Sum"), ("netout", "NetworkOut", "Sum"), ) # Bound per GetMetricData call: 40 instances × 4 queries × 336 hourly # points (14d) ≈ 54k datapoints, safely under the 100 800 cap and the # 500-query cap. _METRIC_BATCH = 40 def _list_metrics_sync( self, targets: list[ServerMetricTarget], lookback_days: int, period_seconds: int, session: Any = None, ) -> list[ServerMetricPoint]: session = session or self._session by_region: dict[str, list[ServerMetricTarget]] = {} for t in targets: if t.region and t.instance_id: by_region.setdefault(t.region, []).append(t) end = datetime.now(UTC) start = end - timedelta(days=lookback_days) points: list[ServerMetricPoint] = [] for region, region_targets in by_region.items(): try: cw = session.client("cloudwatch", region_name=region) except (BotoCoreError, ClientError) as exc: self.partial_failures.append({ "leg": f"region:{region}", "capability": "metrics", "reason": "client_init_error", "error_message": redact_provider_error(str(exc), max_len=160), }) continue for i in range(0, len(region_targets), self._METRIC_BATCH): chunk = region_targets[i:i + self._METRIC_BATCH] queries: list[dict[str, Any]] = [] idmap: dict[str, tuple[str, str]] = {} for idx, t in enumerate(chunk): for prefix, metric, stat in self._METRIC_SPECS: qid = f"{prefix}_{idx}" idmap[qid] = (t.instance_id, prefix) queries.append({ "Id": qid, "MetricStat": { "Metric": { "Namespace": "AWS/EC2", "MetricName": metric, "Dimensions": [ {"Name": "InstanceId", "Value": t.instance_id}, ], }, "Period": period_seconds, "Stat": stat, }, "ReturnData": True, }) try: series: dict[str, list[tuple[Any, float]]] = {} next_token: str | None = None while True: kwargs: dict[str, Any] = { "MetricDataQueries": queries, "StartTime": start, "EndTime": end, "ScanBy": "TimestampAscending", } if next_token: kwargs["NextToken"] = next_token resp = cw.get_metric_data(**kwargs) for res in resp.get("MetricDataResults", []): qid = res["Id"] acc = series.setdefault(qid, []) for ts, val in zip( res.get("Timestamps", []), res.get("Values", []), ): acc.append((ts, val)) next_token = resp.get("NextToken") if not next_token: break except (BotoCoreError, ClientError) as exc: self.partial_failures.append({ "leg": f"region:{region}", "capability": "metrics", "reason": "get_metric_data_error", "error_message": redact_provider_error(str(exc), max_len=160), }) continue # Merge the per-metric series into one point per (instance, ts). merged: dict[tuple[str, Any], dict[str, float]] = {} for qid, pairs in series.items(): instance_id, kind = idmap[qid] for ts, val in pairs: merged.setdefault((instance_id, ts), {})[kind] = val for (instance_id, ts), fields in merged.items(): points.append(ServerMetricPoint( instance_id=instance_id, timestamp=ts, cpu_pct=fields.get("cpu"), cpu_pct_max=fields.get("cpumax"), net_in_bytes=fields.get("netin"), net_out_bytes=fields.get("netout"), )) return points async def list_server_metrics( self, targets: list[ServerMetricTarget], *, lookback_days: int = 14, period_seconds: int = 3600, ) -> list[ServerMetricPoint]: if not self._org_fanout: return await asyncio.to_thread( self._list_metrics_sync, targets, lookback_days, period_seconds, ) # Group targets by the owning member account (stamped into raw by the # list_servers fan-out) and query CloudWatch in each account under an # assumed-role session. Unstamped targets fall back to the base # (management) session. by_acct: dict[str | None, list[ServerMetricTarget]] = {} for t in targets: acct = (t.raw or {}).get("_aws_account_id") if isinstance(t.raw, dict) else None by_acct.setdefault(acct, []).append(t) sem = asyncio.Semaphore(self._ORG_FANOUT_CONCURRENCY) async def _one(acct: str | None, tgts: list[ServerMetricTarget]) -> list[ServerMetricPoint]: async with sem: session = None if acct: try: session = await asyncio.to_thread(self._assume_session_sync, acct) except ClientError as exc: code = (exc.response.get("Error") or {}).get("Code", "") if code in self._ASSUME_ROLE_SKIP_CODES: # Member role gone since inventory (or never # existed) — skip quietly, don't page. Mirrors # the servers-leg classification. self.skipped_legs.append({ "leg": f"account:{acct}", "capability": "metrics", "reason": "member_role_unavailable", }) return [] self.partial_failures.append({ "leg": f"account:{acct}", "capability": "metrics", "reason": "assume_role_failed", "error_message": redact_provider_error(str(exc), max_len=160), }) return [] except BotoCoreError as exc: self.partial_failures.append({ "leg": f"account:{acct}", "capability": "metrics", "reason": "assume_role_transport_error", "error_message": redact_provider_error(str(exc), max_len=160), }) return [] try: return await asyncio.to_thread( self._list_metrics_sync, tgts, lookback_days, period_seconds, session, ) except (BotoCoreError, ClientError) as exc: self.partial_failures.append({ "leg": f"account:{acct or 'management'}", "capability": "metrics", "reason": "get_metric_data_error", "error_message": redact_provider_error(str(exc), max_len=160), }) return [] results = await asyncio.gather(*[_one(a, t) for a, t in by_acct.items()]) out: list[ServerMetricPoint] = [] for r in results: out.extend(r) return out # ── Server lifecycle (EC2 actions) ──────────────────────── # # AWS EC2 lifecycle actions are inherently idempotent: calling # ``start_instances`` on a running instance returns # ``CurrentState=running`` + ``PreviousState=running`` with no # error. We do not need an explicit "already-in-target" catch. # ``terminate_instances`` on an already-terminated instance # returns success (it's been terminated, that's the target state). # Only ``InvalidInstanceID.NotFound`` requires special handling: # treat as idempotent success for destroy (matches "the row # already vanished from EC2") and as a real error for start/stop/ # reboot (you can't start what doesn't exist). # # All three calls run in a worker thread because boto3 is sync. def _ec2_lifecycle_sync( self, op: str, *, instance_id: str, region: str | None, idempotent_on_not_found: bool = False, ) -> None: if not region: raise ValueError( f"AWS {op} requires region — server row has no region", ) try: ec2 = self._session.client("ec2", region_name=region) method = getattr(ec2, op) method(InstanceIds=[instance_id]) except NoCredentialsError as exc: raise ProviderAuthError(_err(f"AWS {op} credential error", exc)) from exc except ClientError as exc: code = exc.response.get("Error", {}).get("Code", "") if code in ("AuthFailure", "InvalidClientTokenId", "UnauthorizedOperation"): raise ProviderAuthError(redact_provider_error(exc)) from exc if code in ( "InvalidInstanceID.NotFound", "InvalidInstanceID.Malformed", ): if idempotent_on_not_found: # Destroy on a row whose EC2 instance is already gone — # the safe-sweep on the next sync would have removed # the cmdb row anyway. Treat as success. return # For start/stop/reboot on a non-existent instance, the # condition is PERMANENT (a retry in 5 min won't bring # it back) so ``ProviderTransientError`` would be # misclassified — surface as ValueError → router 422 # with an admin-actionable message ("instance no longer # exists in cloud — refresh / wait for next sync to # safe-sweep the stale row"). Cleaner than a confusing # "transient" error in a permanent-state situation. raise ValueError( f"AWS {op} target instance not found in {region} " f"(code={code}) — cmdb row references a non-existent " f"cloud instance; the next cmdb.servers sync will " f"safe-sweep it", ) from exc raise ProviderTransientError(redact_provider_error(exc)) from exc except BotoCoreError as exc: raise ProviderTransientError(_err(f"AWS {op} transient", exc)) from exc async def power_on( self, *, instance_id: str, region: str | None = None, project_id: str | None = None, ) -> None: await asyncio.to_thread( self._ec2_lifecycle_sync, "start_instances", instance_id=instance_id, region=region, ) async def power_off( self, *, instance_id: str, region: str | None = None, project_id: str | None = None, force: bool = False, ) -> None: await asyncio.to_thread( self._ec2_lifecycle_sync, "stop_instances", instance_id=instance_id, region=region, ) async def restart_server( self, *, instance_id: str, region: str | None = None, project_id: str | None = None, ) -> None: await asyncio.to_thread( self._ec2_lifecycle_sync, "reboot_instances", instance_id=instance_id, region=region, ) async def destroy_server( self, *, instance_id: str, region: str | None = None, project_id: str | None = None, ) -> None: await asyncio.to_thread( self._ec2_lifecycle_sync, "terminate_instances", instance_id=instance_id, region=region, idempotent_on_not_found=True, ) def _list_domains_sync(self) -> list[CloudDomain]: domains: list[CloudDomain] = [] try: r53 = self._session.client("route53domains", region_name="us-east-1") paginator = r53.get_paginator("list_domains") for page in paginator.paginate(): for d in page.get("Domains", []): detail = r53.get_domain_detail(DomainName=d["DomainName"]) expiry = detail.get("ExpirationDate") expiry_date = expiry.date() if isinstance(expiry, datetime) else expiry domains.append(CloudDomain( domain_name=d["DomainName"], registrar="aws", expiry_date=expiry_date, auto_renew=d.get("AutoRenew", False), icp_filing_no=None, status="active", raw=_json_safe(detail), )) except NoCredentialsError as exc: raise ProviderAuthError(_err("AWS credential error", exc)) from exc except ClientError as e: code = e.response["Error"]["Code"] if code in ("AuthFailure", "InvalidClientTokenId"): raise ProviderAuthError(redact_provider_error(e)) # Migration 173: route53domains is region-locked to us-east-1 # and gated behind its own IAM action — most existing AWS # integrations don't grant ``route53domains:ListDomains`` / # ``GetDomainDetail``. Treat that as graceful degradation # (empty list + partial-failure note) rather than a transient # error: it's a permanent state of the integration, raising # transient would loop the sync engine forever marking the # capability "transient_error" with no actionable fix. if code in ("AccessDeniedException", "AccessDenied", "UnauthorizedOperation"): self.partial_failures.append({ "leg": "route53domains", "capability": "domains", "reason": ( f"{code}: integration lacks route53domains permission. " "Grant route53domains:ListDomains + GetDomainDetail " "on the IAM policy to enable domain-registry sync." ), }) return [] # Round-22 audit: non-auth ClientError used to fall through # and ``return domains`` with a partially-populated list — # safe-sweep would then nuke the missing rows from # ``cmdb_domains`` if the cumulative drop crossed the 95% # floor. Raise transient so the sync engine marks the # capability failed and the mark-and-sweep is skipped this # tick. raise ProviderTransientError(redact_provider_error(e)) from e return domains async def list_domains(self) -> list[CloudDomain]: """List Route53 registered domains (Route53 is global — us-east-1 endpoint).""" return await asyncio.to_thread(self._list_domains_sync) def _list_dns_records_sync(self) -> list[CloudDnsRecord]: records: list[CloudDnsRecord] = [] try: r53 = self._session.client("route53", region_name="us-east-1") paginator = r53.get_paginator("list_hosted_zones") for page in paginator.paginate(): for zone in page["HostedZones"]: zone_id = zone["Id"].split("/")[-1] zone_name = zone["Name"].rstrip(".") rr_paginator = r53.get_paginator("list_resource_record_sets") for rr_page in rr_paginator.paginate(HostedZoneId=zone_id): for rr in rr_page["ResourceRecordSets"]: rr_name = rr["Name"].rstrip(".") rr_type = rr["Type"] if "ResourceRecords" in rr: value = "\n".join(v["Value"] for v in rr["ResourceRecords"]) elif "AliasTarget" in rr: value = rr["AliasTarget"]["DNSName"] else: value = "" ttl = int(rr.get("TTL", 300)) relative = rr_name if rr_name == zone_name: relative = "@" elif rr_name.endswith("." + zone_name): relative = rr_name[: -(len(zone_name) + 1)] or "@" record_id = f"{zone_id}/{relative}/{rr_type}" records.append(CloudDnsRecord( zone_name=zone_name, record_id=record_id, name=relative, record_type=rr_type, value=value, ttl=ttl, line=None, status="enable", raw=_json_safe(rr), )) except NoCredentialsError as exc: raise ProviderAuthError(_err("AWS credential error", exc)) from exc except ClientError as e: code = e.response["Error"]["Code"] if code in ("AuthFailure", "InvalidClientTokenId"): raise ProviderAuthError(redact_provider_error(e)) # Round-22: same fix as _list_domains_sync — raise transient # so safe-sweep doesn't nuke partial-list survivors. raise ProviderTransientError(redact_provider_error(e)) from e return records async def list_dns_records(self) -> list[CloudDnsRecord]: return await asyncio.to_thread(self._list_dns_records_sync) def _list_certificates_sync(self) -> list[CloudCertificate]: """List ACM certificates across all opted-in regions. ACM is regional — the same logical cert imported into two regions has two ARNs and is two rows (provider-faithful).""" certs: list[CloudCertificate] = [] for region in self._ec2_regions_sync(): try: acm = self._session.client("acm", region_name=region) paginator = acm.get_paginator("list_certificates") for page in paginator.paginate(): for summary in page.get("CertificateSummaryList", []): arn = summary.get("CertificateArn") if not arn: continue detail = acm.describe_certificate(CertificateArn=arn) cert = detail.get("Certificate", {}) or {} certs.append(CloudCertificate( cloud_cert_id=arn, common_name=cert.get("DomainName"), sans=list(cert.get("SubjectAlternativeNames", []) or []), issuer=cert.get("Issuer"), not_before=cert.get("NotBefore"), not_after=cert.get("NotAfter"), status=cert.get("Status", "") or "", serial_number=cert.get("Serial"), raw=_json_safe(cert), )) except NoCredentialsError as exc: raise ProviderAuthError(_err("AWS credential error", exc)) from exc except ClientError as e: code = e.response["Error"]["Code"] if code in ("AuthFailure", "InvalidClientTokenId"): raise ProviderAuthError(redact_provider_error(e)) # Transient / per-region issue → raise transient so the # safe-sweep doesn't nuke partial-list survivors. raise ProviderTransientError(redact_provider_error(e)) from e return certs async def list_certificates(self) -> list[CloudCertificate]: return await asyncio.to_thread(self._list_certificates_sync) def _get_zone_id_by_name_sync(self, zone_name: str) -> str | None: """Find the Route53 hosted zone ID for a given zone name.""" r53 = self._session.client("route53", region_name="us-east-1") resp = r53.list_hosted_zones_by_name(DNSName=zone_name + ".", MaxItems="1") for zone in resp.get("HostedZones", []): if zone["Name"].rstrip(".") == zone_name: return zone["Id"].split("/")[-1] return None def _create_dns_record_sync(self, record) -> str: try: zone_id = self._get_zone_id_by_name_sync(record.zone_name) if not zone_id: raise ValueError(f"Hosted zone not found for {record.zone_name}") fqdn = f"{record.name}.{record.zone_name}" if record.name != "@" else record.zone_name r53 = self._session.client("route53", region_name="us-east-1") r53.change_resource_record_sets( HostedZoneId=zone_id, ChangeBatch={ "Changes": [{ "Action": "UPSERT", "ResourceRecordSet": { "Name": fqdn, "Type": record.record_type, "TTL": record.ttl, "ResourceRecords": [{"Value": record.value}], }, }] }, ) return f"{zone_id}/{record.name}/{record.record_type}" except NoCredentialsError as exc: raise ProviderAuthError(_err("AWS credential error", exc)) from exc except ClientError as e: code = e.response["Error"]["Code"] if code in ("AuthFailure", "InvalidClientTokenId"): raise ProviderAuthError(redact_provider_error(e)) raise ProviderTransientError(redact_provider_error(e)) 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: # For Route53, update = UPSERT with new values try: parts = record_id.split("/") zone_id = parts[0] if len(parts) >= 3 else self._get_zone_id_by_name_sync(zone_name) fqdn = f"{record.name}.{zone_name}" if record.name != "@" else zone_name r53 = self._session.client("route53", region_name="us-east-1") r53.change_resource_record_sets( HostedZoneId=zone_id, ChangeBatch={ "Changes": [{ "Action": "UPSERT", "ResourceRecordSet": { "Name": fqdn, "Type": record.record_type, "TTL": record.ttl, "ResourceRecords": [{"Value": record.value}], }, }] }, ) except NoCredentialsError as exc: raise ProviderAuthError(_err("AWS credential error", exc)) from exc except ClientError as e: code = e.response["Error"]["Code"] if code in ("AuthFailure", "InvalidClientTokenId"): raise ProviderAuthError(redact_provider_error(e)) raise ProviderTransientError(redact_provider_error(e)) 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: try: parts = record_id.split("/") # record_id = "zone_id/name/type" if len(parts) < 3: raise ValueError(f"Cannot parse record_id: {record_id}") zone_id, rec_name, rec_type = parts[0], parts[1], parts[2] fqdn = f"{rec_name}.{zone_name}" if rec_name != "@" else zone_name # Need TTL + value — fetch current record first r53 = self._session.client("route53", region_name="us-east-1") rr_resp = r53.list_resource_record_sets( HostedZoneId=zone_id, StartRecordName=fqdn, StartRecordType=rec_type, MaxItems="1", ) rrsets = rr_resp.get("ResourceRecordSets", []) if not rrsets: return # already gone rrset = rrsets[0] # StartRecordName/StartRecordType is a CURSOR into the sorted # zone, not a filter: when the target record is already gone # (the common reconciliation case), Route53 returns the NEXT # record at-or-after the key. Blindly deleting rrsets[0] would # then destroy an unrelated neighbouring record. Verify the # returned set actually matches the intended name+type before # issuing the DELETE (Route53 returns Names with a trailing # dot — normalise both sides). Non-match == already gone. got_name = str(rrset.get("Name", "")).rstrip(".").lower() want_name = fqdn.rstrip(".").lower() if got_name != want_name or str(rrset.get("Type", "")) != rec_type: return # target already removed; do not delete a neighbour r53.change_resource_record_sets( HostedZoneId=zone_id, ChangeBatch={ "Changes": [{ "Action": "DELETE", "ResourceRecordSet": rrset, }] }, ) except NoCredentialsError as exc: raise ProviderAuthError(_err("AWS credential error", exc)) from exc except ClientError as e: code = e.response["Error"]["Code"] if code in ("AuthFailure", "InvalidClientTokenId"): raise ProviderAuthError(redact_provider_error(e)) raise ProviderTransientError(redact_provider_error(e)) 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 _list_load_balancers_sync(self) -> list[CloudLoadBalancer]: lbs: list[CloudLoadBalancer] = [] for region in self._ec2_regions_sync(): try: client = self._session.client("elbv2", region_name=region) paginator = client.get_paginator("describe_load_balancers") for page in paginator.paginate(): for lb in page["LoadBalancers"]: network_type = "internal" if lb.get("Scheme") == "internal" else "public" lb_type_map = {"application": "application", "network": "network"} lb_type = lb_type_map.get(lb.get("Type", ""), "classic") lbs.append(CloudLoadBalancer( lb_id=lb["LoadBalancerArn"], name=lb.get("LoadBalancerName", ""), lb_type=lb_type, network_type=network_type, ip_address=lb.get("DNSName", ""), region=region, status=lb.get("State", {}).get("Code", "unknown"), raw=_json_safe(lb), )) except NoCredentialsError as exc: raise ProviderAuthError(_err("AWS credential error", exc)) from exc except ClientError as e: code = e.response["Error"]["Code"] if code in ("AuthFailure", "InvalidClientTokenId"): raise ProviderAuthError(redact_provider_error(e)) # Round-27 pre-launch (2026-05-25): route ``OptInRequired`` # to ``skipped_legs`` (admin state — retrying won't help) # symmetrically with ``_list_servers_sync``. Previously # an opt-in-required region (me-south-1 / af-south-1 / # ap-east-1 / eu-south-1 / etc.) for a tenant with LBs # scoped to enabled regions only would page IT admin # every cron tick as a partial_failure. if code == "OptInRequired": self.skipped_legs.append({ "leg": f"region:{region}", "capability": "load_balancers", "reason": "region_not_opted_in", }) continue # Round-22 audit: record the per-region failure so the # sync engine surfaces "N regions skipped" and the # safe-sweep does NOT treat the missing rows as # deletions. Mirrors the _list_servers_sync pattern. self.partial_failures.append({ "leg": f"region:{region}", "capability": "load_balancers", "reason": str(code), }) continue except BotoCoreError as exc: # Mirror _list_servers_sync: per-region transport hiccup # records partial_failure and continues; do NOT abort # the whole LB sweep and lose other regions' data. self.partial_failures.append({ "leg": f"region:{region}", "capability": "load_balancers", "reason": "transport_error", "error_message": redact_provider_error( str(exc), max_len=160, ), }) continue return lbs async def list_load_balancers(self) -> list[CloudLoadBalancer]: return await asyncio.to_thread(self._list_load_balancers_sync) def _list_cdn_domains_sync(self) -> list[CloudCdnDomain]: domains: list[CloudCdnDomain] = [] try: cf = self._session.client("cloudfront", region_name="us-east-1") paginator = cf.get_paginator("list_distributions") for page in paginator.paginate(): dist_list = page.get("DistributionList", {}) for dist in dist_list.get("Items", []): aliases = dist.get("Aliases", {}).get("Items", []) domain_name = aliases[0] if aliases else dist.get("DomainName", "") cname = dist.get("DomainName", "") origins = dist.get("Origins", {}).get("Items", []) source_type, source_address = "domain", "" if origins: origin = origins[0] source_address = origin.get("DomainName", "") source_type = "oss" if "s3" in source_address.lower() else "domain" viewer_proto = dist.get("DefaultCacheBehavior", {}).get("ViewerProtocolPolicy", "") is_https = viewer_proto in ("https-only", "redirect-to-https") status_map = {"Deployed": "online", "InProgress": "configuring"} status = status_map.get(dist.get("Status", ""), "offline") domains.append(CloudCdnDomain( domain_name=domain_name, cname=cname, status=status, source_type=source_type, source_address=source_address, is_https=is_https, raw=_json_safe(dist), )) except NoCredentialsError as exc: raise ProviderAuthError(_err("AWS credential error", exc)) from exc except ClientError as e: code = e.response["Error"]["Code"] if code in ("AuthFailure", "InvalidClientTokenId"): raise ProviderAuthError(redact_provider_error(e)) # Round-22: same fix as _list_domains_sync. raise ProviderTransientError(redact_provider_error(e)) from e return domains async def list_cdn_domains(self) -> list[CloudCdnDomain]: return await asyncio.to_thread(self._list_cdn_domains_sync) def _list_object_storage_sync(self) -> list[dict]: try: s3 = self._session.client("s3", region_name="us-east-1") account_id = str( self._session.client("sts", region_name="us-east-1") .get_caller_identity().get("Account") or "" ) resp = s3.list_buckets() result = [] for b in resp.get("Buckets", []): # Pre-launch hardening 2026-05-16 follow-up (CR1 P2): # ``list_buckets`` doesn't carry per-bucket region; # ``get_bucket_location`` returns LocationConstraint # which is the canonical region (or "" for us-east-1 # — AWS's historical wart, since us-east-1 was the # original "no constraint" region). Per-bucket # round-trip is fine for typical fleets (≤ a few # hundred buckets); per-bucket failure logs + falls # back to None so a 403 on one bucket doesn't # cascade. Cached client per call so the connection # pool stays warm. region = None try: loc = s3.get_bucket_location(Bucket=b["Name"]) region_raw = loc.get("LocationConstraint") region = ( "us-east-1" if region_raw in (None, "") else str(region_raw) ) except ClientError as loc_err: loc_code = loc_err.response["Error"]["Code"] if loc_code in ("AuthFailure", "InvalidClientTokenId"): raise # Per-bucket 403 / NoSuchBucket / cross-region # access — record as partial_failure so the admin # UI surfaces "N buckets without region" instead # of pretending every bucket lives in us-east-1. self.partial_failures.append({ "leg": f"bucket:{b['Name']}", "capability": "object_storage", "reason": f"get_bucket_location_{loc_code}", }) result.append({ "bucket_name": b["Name"], "region": region, "storage_class": None, "access_type": None, "status": "active", "size_bytes": None, "object_count": None, "raw": {**_json_safe(b), "_aws_account_id": account_id}, }) return result except NoCredentialsError as exc: raise ProviderAuthError(_err("AWS credential error", exc)) from exc except ClientError as e: code = e.response["Error"]["Code"] if code in ("AuthFailure", "InvalidClientTokenId"): raise ProviderAuthError(redact_provider_error(e)) # Re-raise as ProviderTransientError so sync marks the # capability ``failed`` instead of returning ``[]`` and # letting mark-and-sweep nuke every previously-seen # bucket. The sweep guard added in sync.py is a # belt-and-braces second line; this is the suspenders. raise ProviderTransientError( f"AWS S3 list_buckets failed (code={code}): {e}" ) from e async def list_object_storage(self) -> list[dict]: """List S3 buckets via boto3 list_buckets().""" return await asyncio.to_thread(self._list_object_storage_sync) # ─── IAM user sync ───────────────────────────────────────────── def _list_iam_users_sync(self) -> list[dict]: users: list[dict] = [] try: iam = self._session.client("iam", region_name="us-east-1") paginator = iam.get_paginator("list_users") for page in paginator.paginate(): for u in page.get("Users", []): user_name = u["UserName"] # Enrich with attached/inline policies + groups + MFA # state in a single place — sync is low-frequency and # the admin UI needs this at query time, so paying # per-user API cost here beats N+1 at read time. # Round-22 audit: previously each enrichment call # used a bare ``except ClientError: pass`` which # collapsed AuthFailure (must propagate) and # AccessDenied (real "I cannot read this user's # policies") into the same "render as no policies" # silence. Helix would then render the user as # "no attached policies" — a downstream offboarding # heuristic that uses "0 policies = nothing to # clean up" gets a wrong-positive. Distinguish # auth-fatal (re-raise), permission-denied / not- # found (record + continue), and surface to admin. def _enrichment_err( leg: str, exc: ClientError, ) -> None: # Round-2 review-2: re-raise the explicit # ``exc`` argument rather than the bare # ``raise``. Bare-``raise`` reads from # ``sys.exc_info()`` and silently propagates # whatever happens to be the active exception # if a future refactor invokes this helper # outside an ``except`` block. ``raise exc`` # makes the contract explicit and robust. ecode = ( exc.response.get("Error", {}).get("Code", "") ) if ecode in ("AuthFailure", "InvalidClientTokenId"): raise exc self.partial_failures.append({ "leg": leg, "capability": "iam_users", "reason": ecode or type(exc).__name__, }) attached: list[dict] = [] try: attached_raw = iam.list_attached_user_policies( UserName=user_name, ).get("AttachedPolicies") or [] attached = [ {"name": p["PolicyName"], "arn": p["PolicyArn"]} for p in attached_raw ] except ClientError as exc: _enrichment_err( f"user:{user_name}:attached_policies", exc, ) inline_names: list[str] = [] try: inline_names = iam.list_user_policies( UserName=user_name, ).get("PolicyNames") or [] except ClientError as exc: _enrichment_err( f"user:{user_name}:inline_policies", exc, ) if inline_names: attached.extend( {"name": n, "arn": "inline"} for n in inline_names ) groups: list[str] = [] try: groups = [ g["GroupName"] for g in iam.list_groups_for_user( UserName=user_name, ).get("Groups") or [] ] except ClientError as exc: _enrichment_err(f"user:{user_name}:groups", exc) mfa_enabled = False try: mfa_enabled = bool(iam.list_mfa_devices( UserName=user_name, ).get("MFADevices")) except ClientError as exc: _enrichment_err(f"user:{user_name}:mfa", exc) tags = {} try: tags = { t["Key"]: t["Value"] for t in iam.list_user_tags( UserName=user_name, ).get("Tags") or [] } except ClientError as exc: _enrichment_err(f"user:{user_name}:tags", exc) users.append({ "user_name": user_name, "arn": u.get("Arn"), "user_id_external": u.get("UserId"), "create_date": u.get("CreateDate"), "last_login_at": u.get("PasswordLastUsed"), "attached_policies": attached, "groups": groups, "mfa_enabled": mfa_enabled, # Really probed (list_mfa_devices), so the # tri-state is a genuine answer either way. "mfa_state": ( "enabled" if mfa_enabled else "disabled" ), "tags": tags, "raw": _json_safe(u), }) except NoCredentialsError as exc: raise ProviderAuthError(_err("AWS credential error", exc)) from exc except ClientError as e: code = e.response["Error"]["Code"] if code in ("AuthFailure", "InvalidClientTokenId"): raise ProviderAuthError(redact_provider_error(e)) # Round-27 pre-launch (2026-05-25): non-auth ClientError # (``Throttling`` / ``ServiceFailure`` / ``InternalError`` / # ``AccessDenied`` mid-enrichment) used to fall through and # ``return users`` with a partial list. The sync engine then # safe-swept ``cmdb_cloud_iam_users`` rows that didn't make # the truncated list — if the throttle drop stayed under the # 20% guard, REAL IAM users got silently soft-deleted from # Helix while still active on AWS. Mirror the # ``_list_domains_sync`` / ``_list_cdn_domains_sync`` pattern # (line 355, 607) and propagate as transient so the sync # engine marks the capability failed and SKIPS the sweep. raise ProviderTransientError(redact_provider_error(e)) from e except BotoCoreError as exc: # Same rationale as ClientError above: transport / TLS / DNS # hiccup must NOT silently truncate the list. Propagate. raise ProviderTransientError(_err("AWS IAM transport error", exc)) from exc return users async def list_iam_users(self) -> list[dict]: return await asyncio.to_thread(self._list_iam_users_sync) # ─── Monthly billing via Cost Explorer ───────────────────────── def _get_monthly_billing_sync(self, month: date) -> dict: """Fetch one month's cost summary via AWS Cost Explorer. ``month`` is any date inside the target month; we normalize to the first of the month and fetch [first, next-first). Cost Explorer is a us-east-1 global endpoint. """ start = month.replace(day=1) if start.month == 12: end = start.replace(year=start.year + 1, month=1, day=1) else: end = start.replace(month=start.month + 1, day=1) cost_by_service: dict[str, float] = {} total = Decimal("0") currency = "USD" raw_pages: list[dict] = [] # ``GetCostAndUsage`` isn't in boto3's paginator registry — iterate # ``NextPageToken`` ourselves. For a single-month × group-by- # service call the response normally fits in one page, but a busy # org with hundreds of services can spill, and the silent truncation # is exactly the failure mode we want to avoid. try: ce = self._session.client("ce", region_name="us-east-1") next_token: str | None = None while True: kwargs: dict[str, Any] = { "TimePeriod": { "Start": start.isoformat(), "End": end.isoformat(), }, "Granularity": "MONTHLY", "Metrics": ["UnblendedCost"], "GroupBy": [{"Type": "DIMENSION", "Key": "SERVICE"}], } if next_token: kwargs["NextPageToken"] = next_token resp = ce.get_cost_and_usage(**kwargs) raw_pages.append(resp) for results_by_time in resp.get("ResultsByTime", []): for group in results_by_time.get("Groups", []): service = (group.get("Keys") or ["Unknown"])[0] metrics = group.get("Metrics", {}).get("UnblendedCost") or {} amount = Decimal(str(metrics.get("Amount") or 0)) currency = metrics.get("Unit") or currency total += amount cost_by_service[service] = float( Decimal(str(cost_by_service.get(service, 0))) + amount, ) next_token = resp.get("NextPageToken") if not next_token: break except NoCredentialsError as exc: raise ProviderAuthError(_err("AWS credential error", exc)) from exc except ClientError as e: code = e.response["Error"]["Code"] if code in ("AuthFailure", "InvalidClientTokenId"): raise ProviderAuthError(redact_provider_error(e)) # Cost Explorer not enabled etc. — propagate as transient raise ProviderTransientError(_err("Cost Explorer error", e)) from e return { "month": start, "total_cost_usd": float(total), "cost_by_service": cost_by_service, "currency": currency, # Keep the last raw response for debugging / audit; earlier # pages are available in logs if ever needed. "raw_response": _json_safe(raw_pages[-1]) if raw_pages else {}, } async def get_monthly_billing(self, month: date | None = None) -> dict: """Return last full month's billing by default.""" target = month or (date.today().replace(day=1) - timedelta(days=1)) return await asyncio.to_thread(self._get_monthly_billing_sync, target) # ─── IAM user provisioning (for Bedrock API key auto-grant) ─── def _create_iam_user_with_policy_sync( self, user_name: str, policy_name: str, policy_document: dict, tags: dict[str, str] | None = None, group_names: list[str] | None = None, ) -> dict: """Create an IAM user, attach an inline policy, optionally add the user to one or more existing IAM groups, then mint an access key. Returns ``{"access_key_id", "secret_access_key", "user_arn", "groups_attached"}``. The secret is returned ONCE by AWS — the caller must persist it immediately. Subsequent reads from this driver cannot recover it. Order matters: create_user → put_policy → add_to_group → create_access_key. We attach the user to groups *before* minting the credential pair so a missing or misconfigured group fails loudly without leaving an active credential pair behind. A ``NoSuchEntity`` from ``add_user_to_group`` propagates as ``ProviderTransientError`` (the caller's autogrant flow records it in ``detail.bedrock.error`` and the IT admin can repair the group, then retrigger via the rebind endpoint). """ import json as _json iam = self._session.client("iam", region_name="us-east-1") groups_attached: list[str] = [] try: create_resp = iam.create_user( UserName=user_name, Tags=[{"Key": k, "Value": v} for k, v in (tags or {}).items()], ) user_arn = create_resp["User"]["Arn"] iam.put_user_policy( UserName=user_name, PolicyName=policy_name, PolicyDocument=_json.dumps(policy_document), ) for group_name in (group_names or []): # AWS rejects empty / whitespace group names with # ValidationError; skip defensively so a stray "" in # the caller's list doesn't blow up the whole grant. gn = (group_name or "").strip() if not gn: continue iam.add_user_to_group(GroupName=gn, UserName=user_name) groups_attached.append(gn) ak_resp = iam.create_access_key(UserName=user_name) ak = ak_resp["AccessKey"] return { "user_arn": user_arn, "access_key_id": ak["AccessKeyId"], "secret_access_key": ak["SecretAccessKey"], "groups_attached": groups_attached, } except NoCredentialsError as exc: raise ProviderAuthError(_err("AWS credential error", exc)) from exc except ClientError as e: code = e.response["Error"]["Code"] if code in ("AuthFailure", "InvalidClientTokenId"): raise ProviderAuthError(redact_provider_error(e)) raise ProviderTransientError(_err("IAM user provisioning failed", e)) from e async def create_iam_user_with_policy( self, user_name: str, policy_name: str, policy_document: dict, tags: dict[str, str] | None = None, group_names: list[str] | None = None, ) -> dict: return await asyncio.to_thread( self._create_iam_user_with_policy_sync, user_name, policy_name, policy_document, tags, group_names, ) # ── IAM access-key rotation primitives ─────────────────────────── # Used by the API-key AK/SK rotation flow (apikey/rotation.py). An # IAM user may hold at most TWO access keys at once — that hard AWS # limit is exactly what dual-active rotation exploits: mint a second # key, hand it to the employee, delete the first after a grace # period. The rotation service prechecks ``list_access_keys`` count # so a second rotation can't attempt a third key (which AWS rejects # with ``LimitExceeded``). def _list_access_keys_sync(self, user_name: str) -> list[dict[str, str]]: iam = self._session.client("iam", region_name="us-east-1") try: resp = iam.list_access_keys(UserName=user_name) return [ { "access_key_id": md["AccessKeyId"], "status": md.get("Status", ""), "create_date": ( md["CreateDate"].isoformat() if md.get("CreateDate") else "" ), } for md in resp.get("AccessKeyMetadata", []) ] except NoCredentialsError as exc: raise ProviderAuthError(_err("AWS credential error", exc)) from exc except ClientError as e: code = e.response["Error"]["Code"] if code in ("AuthFailure", "InvalidClientTokenId"): raise ProviderAuthError(redact_provider_error(e)) if code == "NoSuchEntity": # User already gone (revoked / offboarding teardown). The # caller treats an absent user as "nothing to rotate" — # signal via the distinct exception class. raise raise ProviderTransientError( _err("IAM list_access_keys failed", e), ) from e async def list_access_keys(self, user_name: str) -> list[dict[str, str]]: return await asyncio.to_thread(self._list_access_keys_sync, user_name) def _create_access_key_sync(self, user_name: str) -> dict[str, str]: """Mint a second access key on an existing IAM user. Returns ``{"access_key_id", "secret_access_key"}`` — the secret is returned ONCE by AWS; the caller must persist it immediately. Raises on ``LimitExceeded`` (user already has 2 keys) so the rotation flow surfaces "a rotation is already in flight — retire the old key first". """ iam = self._session.client("iam", region_name="us-east-1") try: ak_resp = iam.create_access_key(UserName=user_name) ak = ak_resp["AccessKey"] return { "access_key_id": ak["AccessKeyId"], "secret_access_key": ak["SecretAccessKey"], } except NoCredentialsError as exc: raise ProviderAuthError(_err("AWS credential error", exc)) from exc except ClientError as e: code = e.response["Error"]["Code"] if code in ("AuthFailure", "InvalidClientTokenId"): raise ProviderAuthError(redact_provider_error(e)) if code in ("NoSuchEntity", "LimitExceeded"): # NoSuchEntity → user gone; LimitExceeded → already 2 # keys. Both are actionable states the rotation service # maps to a specific error; re-raise the typed ClientError # so the caller can inspect ``e.response['Error']['Code']``. raise raise ProviderTransientError( _err("IAM create_access_key failed", e), ) from e async def create_access_key(self, user_name: str) -> dict[str, str]: return await asyncio.to_thread(self._create_access_key_sync, user_name) def _delete_access_key_sync( self, user_name: str, access_key_id: str, ) -> dict[str, Any]: """Delete a specific access key from an IAM user. Idempotent: ``NoSuchEntity`` (key or user already gone) returns ``already_absent=True`` rather than raising, so the retire cron is safe to re-run and safe to fire after a revoke deleted the whole user. """ iam = self._session.client("iam", region_name="us-east-1") try: iam.delete_access_key( UserName=user_name, AccessKeyId=access_key_id, ) return {"deleted": True, "already_absent": False} except NoCredentialsError as exc: raise ProviderAuthError(_err("AWS credential error", exc)) from exc except ClientError as e: code = e.response["Error"]["Code"] if code in ("AuthFailure", "InvalidClientTokenId"): raise ProviderAuthError(redact_provider_error(e)) if code == "NoSuchEntity": return {"deleted": False, "already_absent": True} raise ProviderTransientError( _err("IAM delete_access_key failed", e), ) from e async def delete_access_key( self, user_name: str, access_key_id: str, ) -> dict[str, Any]: return await asyncio.to_thread( self._delete_access_key_sync, user_name, access_key_id, ) def _tag_iam_user_sync( self, user_name: str, tags: dict[str, str], ) -> dict[str, Any]: """Apply (upsert) the given tag key/value pairs onto an existing IAM user. AWS ``iam.tag_user`` is upsert-shaped — it OVERWRITES the value when the key already exists and adds it when it doesn't. Other tags on the user are NOT touched. Used by the Bedrock Owner-tag backfill (PR 2026-05-12) to retro-fit ``Owner=`` on IAM users that were created before the autogrant added the tag step OR were tagged with the wrong value (e.g. ``Owner=`` from a pre-PR-220 revision). The backfill is idempotent — re-running it is a no-op once the tag is correct. Returns ``{"user_name", "tags_applied"}``. AWS ``tag_user`` returns no data on success. """ iam = self._session.client("iam", region_name="us-east-1") tag_list = [ {"Key": k, "Value": v} for k, v in (tags or {}).items() if v is not None ] if not tag_list: return {"user_name": user_name, "tags_applied": []} try: iam.tag_user(UserName=user_name, Tags=tag_list) return { "user_name": user_name, "tags_applied": [t["Key"] for t in tag_list], } except NoCredentialsError as exc: raise ProviderAuthError(_err("AWS credential error", exc)) from exc except ClientError as e: code = e.response["Error"]["Code"] if code in ("AuthFailure", "InvalidClientTokenId"): raise ProviderAuthError(redact_provider_error(e)) if code == "NoSuchEntity": # User was deleted between Helix's read and our call — # the backfill should skip this row, not crash the run. # Caller distinguishes via the exception class. raise raise ProviderTransientError(_err("IAM tag_user failed", e)) from e async def tag_iam_user( self, user_name: str, tags: dict[str, str], ) -> dict[str, Any]: return await asyncio.to_thread( self._tag_iam_user_sync, user_name, tags, )