from __future__ import annotations import abc import enum from dataclasses import dataclass from datetime import date, datetime from typing import Any class ProviderCapability(enum.StrEnum): SERVERS = "servers" DOMAINS = "domains" DNS_RECORDS = "dns_records" CERTIFICATES = "certificates" CDN = "cdn" OBJECT_STORAGE = "object_storage" LOAD_BALANCERS = "load_balancers" APPLICATIONS = "applications" # Reserved: per-capability driver not yet implemented. Providers that # declare IAM in ``supported_capabilities()`` must also override # ``list_iam_users()``; the sync engine raises CapabilityNotSupported # otherwise. Kept in the enum so scheduled_jobs rows + UI can show # "IAM 同步" as a capability before any driver ships it. IAM = "iam" # Runtime utilization (CPU / memory / disk / network) pulled from each # cloud's monitoring service — CloudWatch (AWS), CloudMonitor/Cms # (Aliyun), Azure Monitor, Cloud Monitoring (GCP). Distinct from # SERVERS (which syncs *capacity*: vcpu / memory_mb). Providers that # declare METRICS must override ``list_server_metrics()``. Drives the # cloud-efficiency / cost-optimization engine. METRICS = "metrics" @dataclass class CloudResourceInventoryItem: """Provider-neutral record from a broad, read-only inventory API.""" provider_resource_id: str resource_type: str service: str category: str name: str scope_external_id: str | None = None region: str | None = None zone: str | None = None status: str | None = None tags: dict[str, Any] | None = None metadata: dict[str, Any] | None = None sensitivity: str = "internal" def classify_resource_category(resource_type: str, service: str = "") -> tuple[str, str]: """Conservative category and sensitivity defaults for dynamic types.""" value = f"{service} {resource_type}".casefold() if any(part in value for part in ("iam", "identity", "secret", "credential", "keyvault", "key pair", "keypair")): return "identity", "restricted" categories = ( ("ai", ( "aiplatform", "artificialintelligence", "bedrock", "cognitiveservices", "machinelearning", "openai", "sagemaker", "vertex", )), ("analytics", ("bigquery", "datawarehouse", "analytics", "synapse")), ("database", ("database", "sql", "dynamodb", "cosmos", "redis", "rds")), ("storage", ("storage", "bucket", "blob", "filesystem", "disk", "volume")), ("network", ("network", "vpc", "subnet", "dns", "loadbalanc", "gateway", "cdn", "firewall")), ("security", ("security", "certificate", "kms", "vault", "defender")), ("compute", ("compute", "instance", "virtualmachine", "container", "kubernetes", "lambda", "function")), ("observability", ("monitor", "logging", "trace", "metric")), ("integration", ("queue", "event", "message", "workflow", "integration")), ) for category, needles in categories: if any(needle in value for needle in needles): return category, "internal" return "other", "internal" @dataclass class CloudServer: instance_id: str name: str status: str # normalized: "running" | "stopped" | "terminated" region: str zone: str | None vcpu: int | None memory_mb: int | None public_ips: list[str] private_ips: list[str] image_id: str | None os_type: str | None # "linux" | "windows" tags: dict[str, str] raw: dict[str, Any] @dataclass class ServerMetricTarget: """The identity a metrics adapter needs to query one instance's utilization. Built by the metrics-sync handler from a ``cmdb_servers`` row — the adapter does NOT re-discover the fleet. Each cloud needs a different subset: * AWS — ``region`` + ``instance_id`` (CloudWatch dimension) * Aliyun — ``region`` + ``instance_id`` (CloudMonitor dimension) * Azure — the full ARM resourceId, reconstructed from ``raw`` (the synced VM record) or ``instance_id`` when it already is the resourceId * GCP — ``project_id`` + ``instance_id`` (numeric id / name label) """ instance_id: str region: str | None zone: str | None = None project_id: str | None = None # Provider-native record (``cmdb_servers.raw_data``) so adapters that # need an id the DTO doesn't carry (Azure ARM resourceId, GCP numeric # instance id) can reconstruct it without a second list call. raw: dict[str, Any] | None = None @dataclass class ServerMetricPoint: """One time-bucket (default 1h) of normalized utilization for a single cloud instance. Adapters emit these; the metrics-sync handler buckets them by local date into ``cmdb_server_metrics_daily`` and the efficiency engine aggregates avg / p95 / max over the window. Every metric field is nullable and degrades gracefully. CPU and network are available at the hypervisor level on all four clouds with no in-guest agent. Memory and disk *usage* require a guest agent (CloudWatch agent / CloudMonitor agent / Azure diagnostics / GCP Ops agent) and are ``None`` when the datapoint is absent — the rule engine treats missing memory as "unknown", never as "0% / idle". """ instance_id: str timestamp: datetime # bucket start (UTC) cpu_pct: float | None = None # avg CPU utilization % over the bucket cpu_pct_max: float | None = None # in-bucket peak, when the provider returns it mem_pct: float | None = None # avg memory utilization % (agent-only) disk_used_pct: float | None = None # filesystem used % (agent-only) net_in_bytes: float | None = None # total inbound bytes over the bucket net_out_bytes: float | None = None # total outbound bytes over the bucket disk_read_bytes: float | None = None disk_write_bytes: float | None = None @dataclass class CloudDomain: domain_name: str registrar: str expiry_date: date | None auto_renew: bool icp_filing_no: str | None status: str # "active" | "expired" raw: dict[str, Any] @dataclass class CloudCertificate: """A TLS/SSL certificate managed by a cloud cert manager (AWS ACM, Aliyun SSL, Tencent SSL). ``cloud_cert_id`` is the provider-side id (ARN / CertId) and the dedup anchor for the synced row.""" cloud_cert_id: str common_name: str | None sans: list[str] issuer: str | None not_before: datetime | None not_after: datetime | None status: str # provider status passed through (e.g. ISSUED / EXPIRED) serial_number: str | None raw: dict[str, Any] # Normalized lowercase-hex SHA-256 fingerprint (64 chars, no colons) # when the provider exposes one — enables cross-source dedup against # probed rows. ``None`` when the provider's list/describe call doesn't # return a SHA-256 (ACM / Tencent list don't); dedup then falls back # to the covered-name set. fingerprint_sha256: str | None = None @dataclass class CloudDnsRecord: zone_name: str record_id: str name: str record_type: str value: str ttl: int line: str | None status: str # "enable" | "disable" raw: dict[str, Any] @dataclass class CloudLoadBalancer: lb_id: str name: str lb_type: str # "classic" | "application" | "network" network_type: str # "public" | "internal" ip_address: str region: str status: str raw: dict[str, Any] @dataclass class CloudCdnDomain: domain_name: str cname: str status: str # "online" | "offline" | "configuring" source_type: str # "domain" | "ip" | "oss" | "other" source_address: str is_https: bool raw: dict[str, Any] @dataclass class DnsRecordWrite: """Input for create/update DNS record write operations.""" zone_name: str name: str # host record: "@" for apex, "www" for subdomain record_type: str # "A" | "AAAA" | "CNAME" | "MX" | "TXT" | "NS" | "SRV" | "CAA" value: str ttl: int = 600 line: str | None = None # Aliyun/Tencent line routing, None for AWS # Smallest TTL each DNS provider accepts on a record write, keyed by the # ``integrations.provider`` string. # # Why this table exists (2026-08-14): a self-service request submitted with # TTL=300 passed Helix's ``ge=60`` schema bound, went through three levels # of human approval, and only THEN got rejected by DNSPod with # ``LimitExceeded.RecordTtlLimit`` — the requester had burned a day for # nothing. Pre-validating at submit turns a post-approval dead end into an # instant, fixable form error. # # Tencent DNSPod: the floor is plan-dependent (免费版 600, paid plans go to # 60). 600 is the correct floor for the account Helix drives today — # verified against the live tenant: 900 records across 10 zones, not one # below 600. An account on a paid plan can lower its own floor via the # integration's ``config_public.min_dns_ttl`` override rather than editing # this table. # # Providers absent from the table inherit ``DEFAULT_MIN_DNS_TTL`` — i.e. no # provider-specific floor beyond the schema bound. Only add an entry once # the floor has been confirmed against a real tenant; a guessed floor # rejects writes the provider would have accepted. MIN_DNS_TTL_BY_PROVIDER: dict[str, int] = { "tencent": 600, } DEFAULT_MIN_DNS_TTL = 60 def min_dns_ttl( provider: str | None, *, config_public: dict[str, Any] | None = None, ) -> int: """Return the smallest TTL ``provider`` will accept on a record write. ``config_public.min_dns_ttl`` (per-integration) wins over the table so an account that upgrades its DNS plan can lower the floor without a code change. A non-positive or non-integer override is ignored rather than trusted — a typo there would otherwise disable the guard silently. """ override = (config_public or {}).get("min_dns_ttl") if isinstance(override, int) and not isinstance(override, bool) and override > 0: return override return MIN_DNS_TTL_BY_PROVIDER.get((provider or "").lower(), DEFAULT_MIN_DNS_TTL) @dataclass class ProviderCredentials: provider_id: str provider_type: str account_alias: str fields: dict[str, str] class CloudProvider(abc.ABC): def __init__(self, credentials: ProviderCredentials) -> None: self._credentials = credentials # Pre-launch hardening 2026-05-16: per-provider scratch list # the multi-region / multi-subscription / multi-project fan-out # helpers append to whenever a single leg of the fan-out fails # but the rest succeeded. The sync engine reads this list after # each capability completes and stamps it into # ``capability_results['partial_failures']`` so admins see # "synced 17/20 Azure subscriptions, 3 skipped (403)" instead # of an apparent green sync that's actually missing rows. # # Entries are free-form dicts; conventional keys: # ``leg``: human-readable identifier (region, sub_id, project_id) # ``capability``: which capability triggered (servers / dns_records) # ``reason``: short string (HTTP status, exception class) # # Drivers that don't fan out leave the list empty. self.partial_failures: list[dict[str, Any]] = [] # Pre-launch hardening 2026-05-17: legs that should NOT count as # sync failures because the underlying state is "structurally # expected" — the leg is functioning correctly, the resource # type just doesn't exist on that leg. The canonical case is GCP # returning 403 for ``compute.instances.aggregatedList`` on # projects that never had Compute Engine API enabled (BigQuery # / Storage / AppEngine / Firebase-only projects). Those 403s # are indistinguishable at the HTTP layer from real permission # denials, but the structured error body carries # ``reason=SERVICE_DISABLED`` or ``accessNotConfigured`` which # drivers parse and route here. # # Distinct from ``partial_failures``: legs here do NOT bump # ``capability_results['status']`` to ``partial_failure`` and do # NOT page the failure-notifier. They surface in a separate # ``capability_results['skipped_legs']`` field for admin # visibility (which project has which API disabled) without # turning every sync red. Production incident 2026-05-17: GCP # servers cron stayed red on every tick because 474 projects in # a 500-employee org legitimately have no GCE, polluting # ``partial_failures`` with ``http_403`` legs on each run. # # Entries are free-form dicts; conventional keys: # ``leg``: human-readable identifier (project_id, sub_id, region) # ``capability``: which capability triggered (servers / dns_records) # ``reason``: short string (e.g. ``api_disabled:compute.googleapis.com``) self.skipped_legs: list[dict[str, Any]] = [] def consume_partial_failures(self) -> list[dict[str, Any]]: """Return + clear the partial-failure list. Called by the sync engine between capabilities so a per-capability result row reflects only that capability's failures (not all previous capabilities' failures piled up). """ out = self.partial_failures self.partial_failures = [] return out def consume_skipped_legs(self) -> list[dict[str, Any]]: """Return + clear the skipped-legs list. Mirror of ``consume_partial_failures`` for "structurally expected" skips (e.g. GCP project with Compute API disabled). Drained per-capability so each result row reflects only that capability's skips, not all preceding capabilities piled up. """ out = self.skipped_legs self.skipped_legs = [] return out @classmethod @abc.abstractmethod def provider_type(cls) -> str: ... @classmethod @abc.abstractmethod def supported_capabilities(cls) -> frozenset[ProviderCapability]: ... def supports(self, cap: ProviderCapability) -> bool: return cap in self.supported_capabilities() async def list_servers(self) -> list[CloudServer]: raise CapabilityNotSupported(f"{self.provider_type()} does not support servers") async def list_domains(self) -> list[CloudDomain]: raise CapabilityNotSupported(f"{self.provider_type()} does not support domains") async def list_dns_records(self) -> list[CloudDnsRecord]: raise CapabilityNotSupported(f"{self.provider_type()} does not support dns_records") async def list_certificates(self) -> list[CloudCertificate]: raise CapabilityNotSupported(f"{self.provider_type()} does not support certificates") async def list_load_balancers(self) -> list[CloudLoadBalancer]: raise CapabilityNotSupported(f"{self.provider_type()} does not support load_balancers") async def list_cdn_domains(self) -> list[CloudCdnDomain]: raise CapabilityNotSupported(f"{self.provider_type()} does not support cdn") async def list_object_storage(self) -> list[dict]: """List object storage buckets. Override in providers that support it.""" raise CapabilityNotSupported(f"{self.provider_type()} does not support object storage listing") async def list_applications(self) -> list[dict]: """List applications. Not typically cloud-provider-driven; returns empty by default.""" return [] async def list_resource_inventory(self) -> list[CloudResourceInventoryItem]: """Enumerate the provider's broad resource catalog without mutations. Providers only override this when an explicit inventory service or tenant-wide resource-list API is available. A raised unsupported error is an operator-visible setup state, never an empty authoritative snapshot, so the directory will not sweep previously observed rows. """ raise CapabilityNotSupported( f"{self.provider_type()} broad resource inventory is not configured", ) async def list_iam_users(self) -> list[dict]: """List IAM / sub-account users on this cloud tenant. Reserved; no driver ships an implementation yet.""" raise CapabilityNotSupported(f"{self.provider_type()} does not support IAM listing") async def list_server_metrics( self, targets: list[ServerMetricTarget], *, lookback_days: int = 14, period_seconds: int = 3600, ) -> list[ServerMetricPoint]: """Pull runtime utilization for the given instances from the cloud's monitoring service. Contract: * ``targets`` are the instances the caller already knows about (from ``cmdb_servers``); the adapter must NOT re-list the fleet. * Return a flat list of hourly (``period_seconds``) buckets across all targets, tagged by ``instance_id``. Missing metrics are ``None``, not ``0`` — the efficiency engine distinguishes "no datapoint" from "measured zero". * Fan-out failures (a region / subscription / project the monitoring API rejects) go to ``self.partial_failures`` / ``self.skipped_legs`` exactly like the inventory list methods, so a partial pull is visible without failing the whole sync. * Providers that declare ``ProviderCapability.METRICS`` must override this; others inherit the ``CapabilityNotSupported`` raise (the metrics-sync handler catches it and returns a skipped status). """ raise CapabilityNotSupported( f"{self.provider_type()} does not support server metrics", ) async def create_dns_record(self, record: DnsRecordWrite) -> str: """Push a new DNS record to the cloud provider. Returns provider record_id.""" raise CapabilityNotSupported(f"{self.provider_type()} does not support DNS write") async def update_dns_record(self, record_id: str, zone_name: str, record: DnsRecordWrite) -> None: """Update an existing DNS record in the cloud provider.""" raise CapabilityNotSupported(f"{self.provider_type()} does not support DNS write") async def delete_dns_record(self, record_id: str, zone_name: str) -> None: """Delete a DNS record from the cloud provider.""" raise CapabilityNotSupported(f"{self.provider_type()} does not support DNS write") async def set_dns_record_status(self, record_id: str, zone_name: str, enable: bool) -> None: """Flip the cloud-side enable/disable flag for an existing DNS record. DNSPod (Tencent) and Aliyun DNS expose a dedicated "modify record status" API for this. AWS Route 53 has no equivalent — records exist or they don't — so the AWS driver leaves this at the default ``CapabilityNotSupported`` raise. Callers (DNS request executor + admin direct-action router) catch it and return a 501 with a precise message. """ raise CapabilityNotSupported( f"{self.provider_type()} does not support DNS record enable/disable", ) # ── Server lifecycle (PR #559) ───────────────────────────── # # Replicates the legacy CMDB's 开机/关机/重启/删除 action set on # cloud servers. Drivers that don't implement these inherit the # ``CapabilityNotSupported`` raise — the router catches it and # returns 501 with a precise message, and the frontend grays out # the corresponding "更多" dropdown items per-provider. # # Contract: every method is fire-and-forget. The cloud API call is # the authoritative trigger; we DO NOT poll for terminal state in # the request path (legacy CMDB had the same shape). The next # scheduled sync reconciles the actual cloud-side status into # ``cmdb_servers.status``. The router updates the row's status to # an optimistic transitional value (``starting`` / ``stopping`` / # ``rebooting`` / ``terminated``) on success so the UI reflects # the intent immediately. # # ``region`` / ``project_id`` are passed by the router from the # ``cmdb_servers`` row — the driver does NOT re-discover them. # ``instance_id`` is the provider-native id (UHostId / ECS # InstanceId / EC2 InstanceId). async def power_on( self, *, instance_id: str, region: str | None = None, project_id: str | None = None, ) -> None: raise CapabilityNotSupported( f"{self.provider_type()} does not support server power-on", ) async def power_off( self, *, instance_id: str, region: str | None = None, project_id: str | None = None, force: bool = False, ) -> None: raise CapabilityNotSupported( f"{self.provider_type()} does not support server power-off", ) async def restart_server( self, *, instance_id: str, region: str | None = None, project_id: str | None = None, ) -> None: raise CapabilityNotSupported( f"{self.provider_type()} does not support server restart", ) async def destroy_server( self, *, instance_id: str, region: str | None = None, project_id: str | None = None, ) -> None: raise CapabilityNotSupported( f"{self.provider_type()} does not support server termination", ) async def close(self) -> None: pass _REGISTRY: dict[str, type[CloudProvider]] = {} def register_provider(cls: type[CloudProvider]) -> type[CloudProvider]: _REGISTRY[cls.provider_type()] = cls return cls def get_provider_class(provider_type: str) -> type[CloudProvider]: cls = _REGISTRY.get(provider_type) if cls is None: raise ValueError(f"Unknown provider type: {provider_type!r}. Known: {list(_REGISTRY)}") return cls # Fix circular import — exceptions needed in base from app.domains.cmdb.providers.exceptions import CapabilityNotSupported # noqa: E402