"""Deterministic cloud-resource governance evaluation and trend rollups.""" from __future__ import annotations import hashlib import json import uuid from datetime import UTC, date, datetime, timedelta from typing import Any from sqlalchemy import exists, or_, select, update from sqlalchemy.dialects.mysql import insert as mysql_insert from sqlalchemy.ext.asyncio import AsyncSession from app.domains.cmdb.models import ( SCOPE_TYPE_AZURE_RESOURCE_GROUP, CmdbCloudResource, CmdbCloudResourceDailyRollup, CmdbCloudScope, CmdbResourceFinding, CmdbResourceGovernance, CmdbResourceGovernanceRule, ) from app.domains.cmdb.resource_directory_coverage import ResourceCoverage, resolve_resource_coverage from app.domains.cmdb.resource_directory_ownership import resolve_effective_resource_owners SUPPORTED_EVALUATORS = frozenset({ "missing_owner", "public_exposure", "stale_observation", "restricted_review", "production_unclassified", "monitoring_missing", "cost_unavailable", }) _ACTIVE_FINDING_STATUSES = ("open", "acknowledged") _PAGE_SIZE = 1_000 def finding_fingerprint(rule_key: str, resource_id: str) -> str: return hashlib.sha256(f"{rule_key}\0{resource_id}".encode()).hexdigest() def _applies(rule: CmdbResourceGovernanceRule, resource: CmdbCloudResource) -> bool: applies = rule.applies_to or {} dimensions = { "providers": resource.provider, "categories": resource.category, "resource_types": resource.resource_type, } for key, value in dimensions.items(): allowed = applies.get(key) if isinstance(allowed, list) and allowed and value not in allowed: return False return True def _public_exposure(resource: CmdbCloudResource) -> tuple[bool, dict[str, Any]]: normalized = resource.normalized_metadata or {} raw = resource.raw_metadata or {} public_ips = normalized.get("public_ips") if isinstance(public_ips, list) and any(str(value).strip() for value in public_ips): return True, {"signal": "public_ips", "count": len(public_ips)} candidates = { "access_type": normalized.get("access_type") or raw.get("access_type"), "network_type": normalized.get("network_type") or raw.get("network_type"), "public_access": raw.get("public_access") or raw.get("publicAccess"), } for key, value in candidates.items(): normalized_value = str(value or "").strip().casefold() if normalized_value in { "true", "public", "internet", "public-read", "public-read-write", }: return True, {"signal": key, "value": normalized_value} return False, {} def evaluate_rule( rule: CmdbResourceGovernanceRule, resource: CmdbCloudResource, governance: CmdbResourceGovernance | None, *, owned_scope_pairs: set[tuple[str, str]], owned_resource_groups: set[tuple[str, str]], now: datetime, effective_owner_person_id: str | None = None, coverage: ResourceCoverage | None = None, ) -> tuple[bool, dict[str, Any]]: if resource.deleted_at is not None or not _applies(rule, resource): return False, {} evaluator = rule.evaluator if evaluator == "missing_owner": resource_group = str((resource.raw_metadata or {}).get("resource_group") or "") scope_owned = ( (resource.integration_id, str(resource.scope_external_id or "")) in owned_scope_pairs or (resource.integration_id, resource_group) in owned_resource_groups ) governed = bool( governance and (governance.business_owner_person_id or governance.technical_owner_person_id) ) hit = not effective_owner_person_id and not governed and not scope_owned return hit, {"scope_external_id": resource.scope_external_id} if evaluator == "public_exposure": return _public_exposure(resource) if evaluator == "stale_observation": hours = int((rule.parameters or {}).get("hours") or 48) hours = max(1, min(hours, 24 * 30)) hit = resource.last_seen_at < now - timedelta(hours=hours) return hit, {"last_seen_at": resource.last_seen_at.isoformat(), "threshold_hours": hours} if evaluator == "restricted_review": exception_expires_at = governance.exception_expires_at if governance else None if exception_expires_at is not None and exception_expires_at.tzinfo is not None: exception_expires_at = exception_expires_at.astimezone(UTC).replace(tzinfo=None) reviewed = bool( governance and ( governance.governance_status in {"reviewed", "compliant"} or ( governance.governance_status == "exception_approved" and exception_expires_at is not None and exception_expires_at > now ) ) ) return resource.sensitivity == "restricted" and not reviewed, { "sensitivity": resource.sensitivity, } if evaluator == "production_unclassified": return resource.environment == "prod" and not (governance and governance.criticality), { "environment": resource.environment, } if evaluator == "monitoring_missing": hit = bool(coverage and coverage.monitoring_status == "missing") return hit, { "monitoring_status": coverage.monitoring_status if coverage else None, "monitoring_latest_at": coverage.monitoring_latest_at if coverage else None, } if evaluator == "cost_unavailable": hit = bool(coverage and coverage.cost_status == "not_available") return hit, { "cost_status": coverage.cost_status if coverage else None, "cost_source": coverage.cost_source if coverage else None, } return False, {} async def evaluate_governance(db: AsyncSession) -> dict[str, int]: # MySQL/TiDB DATETIME columns in this schema have second precision. Keeping # Python microseconds here makes a finding written during this run compare # as ``last_seen_at < run_started`` after the database truncates it, and the # closing sweep would immediately resolve the finding it just opened. run_started = datetime.now(UTC).replace(tzinfo=None, microsecond=0) rules = list((await db.scalars(select(CmdbResourceGovernanceRule).where( CmdbResourceGovernanceRule.enabled.is_(True), CmdbResourceGovernanceRule.evaluator.in_(SUPPORTED_EVALUATORS), ))).all()) scope_rows = (await db.execute(select( CmdbCloudScope.integration_id, CmdbCloudScope.external_id, CmdbCloudScope.scope_type, ).where( CmdbCloudScope.deleted_at.is_(None), CmdbCloudScope.owner_person_id.is_not(None), ))).all() owned_scope_pairs: set[tuple[str, str]] = set() owned_resource_groups: set[tuple[str, str]] = set() for integration_id, external_id, scope_type in scope_rows: pair = (str(integration_id), str(external_id)) if scope_type == SCOPE_TYPE_AZURE_RESOURCE_GROUP: owned_resource_groups.add(pair) else: owned_scope_pairs.add(pair) summary = {"rules": len(rules), "resources": 0, "opened": 0, "updated": 0, "resolved": 0} last_id: uuid.UUID | None = None while True: stmt = select(CmdbCloudResource).order_by(CmdbCloudResource.id).limit(_PAGE_SIZE) if last_id is not None: stmt = stmt.where(CmdbCloudResource.id > last_id) resources = list((await db.scalars(stmt)).all()) if not resources: break resource_ids = [str(resource.id) for resource in resources] governance_rows = (await db.scalars(select(CmdbResourceGovernance).where( CmdbResourceGovernance.resource_id.in_(resource_ids), ))).all() governance_by_resource = {row.resource_id: row for row in governance_rows} effective_owners = await resolve_effective_resource_owners(db, resources) coverage_by_resource = await resolve_resource_coverage(db, resources) hits: list[tuple[CmdbResourceGovernanceRule, CmdbCloudResource, dict[str, Any]]] = [] for resource in resources: governance = governance_by_resource.get(str(resource.id)) for rule in rules: hit, evidence = evaluate_rule( rule, resource, governance, owned_scope_pairs=owned_scope_pairs, owned_resource_groups=owned_resource_groups, now=run_started, effective_owner_person_id=( effective_owners[str(resource.id)].person_id if str(resource.id) in effective_owners else None ), coverage=coverage_by_resource.get(str(resource.id)), ) if hit: hits.append((rule, resource, evidence)) fingerprints = [finding_fingerprint(rule.rule_key, str(resource.id)) for rule, resource, _ in hits] existing_rows = (await db.scalars(select(CmdbResourceFinding).where( CmdbResourceFinding.fingerprint.in_(fingerprints), ))).all() if fingerprints else [] existing = {row.fingerprint: row for row in existing_rows} for rule, resource, evidence in hits: fingerprint = finding_fingerprint(rule.rule_key, str(resource.id)) finding = existing.get(fingerprint) if finding is None: db.add(CmdbResourceFinding( resource_id=str(resource.id), rule_id=str(rule.id), fingerprint=fingerprint, severity=rule.severity, status="open", title=rule.name, description=rule.description, evidence=evidence, recommendation=rule.recommendation, first_seen_at=run_started, last_seen_at=run_started, )) summary["opened"] += 1 else: finding.severity = rule.severity finding.title = rule.name finding.description = rule.description finding.evidence = evidence finding.recommendation = rule.recommendation finding.last_seen_at = run_started if finding.status == "resolved": finding.status = "open" finding.resolved_at = None finding.resolution_note = None elif ( finding.status == "suppressed" and finding.due_at is not None and finding.due_at <= run_started ): finding.status = "open" finding.resolution_note = "治理例外已到期" summary["updated"] += 1 summary["resources"] += len(resources) last_id = resources[-1].id await db.commit() result = await db.execute(update(CmdbResourceFinding).where( CmdbResourceFinding.status.in_(_ACTIVE_FINDING_STATUSES), CmdbResourceFinding.last_seen_at < run_started, exists(select(CmdbResourceGovernanceRule.id).where( CmdbResourceGovernanceRule.id == CmdbResourceFinding.rule_id, or_( CmdbResourceGovernanceRule.enabled.is_(False), CmdbResourceGovernanceRule.evaluator.in_(SUPPORTED_EVALUATORS), ), ).correlate(CmdbResourceFinding)), ).values(status="resolved", resolved_at=run_started, resolution_note="规则条件已不再命中")) summary["resolved"] = int(getattr(result, "rowcount", 0) or 0) await db.commit() return summary def _dimension_key(values: dict[str, str | None]) -> str: encoded = json.dumps(values, sort_keys=True, separators=(",", ":")) return hashlib.sha256(encoded.encode()).hexdigest() async def refresh_daily_rollups(db: AsyncSession, snapshot_date: date | None = None) -> dict[str, int]: day = snapshot_date or datetime.now(UTC).date() dimensions: tuple[tuple[str, ...], ...] = ( (), ("provider",), ("category",), ("provider", "category"), ("provider", "resource_type"), ("provider", "scope_external_id"), ) # A full-table GROUP BY with several correlated EXISTS predicates caused # TiDB to materialize enough intermediate state to trigger error 8176 in # production. Fold bounded pages and prefetch their related signals so # memory remains O(_PAGE_SIZE + distinct rollup dimensions). # # Each page also commits: TiDB serves every plain read in a transaction # from that transaction's start TS, so a single scan of the whole table # would hold one read snapshot for the entire run. Production runs take # ~8 minutes against a 10-minute ``tidb_gc_life_time``, which both risks # error 9006 and pins the cluster GC safepoint for the duration. Paging # with a fresh read TS per page smears the counts across a few seconds of # concurrent sync writes, which a daily rollup tolerates -- and is what # ``evaluate_governance`` above already does. aggregates: dict[ tuple[tuple[str, ...], tuple[str | None, ...]], list[int] ] = {} last_id: uuid.UUID | None = None while True: stmt = select(CmdbCloudResource).order_by(CmdbCloudResource.id).limit(_PAGE_SIZE) if last_id is not None: stmt = stmt.where(CmdbCloudResource.id > last_id) resources = list((await db.scalars(stmt)).all()) if not resources: break resource_ids = [str(resource.id) for resource in resources] governance_rows = list((await db.scalars(select(CmdbResourceGovernance).where( CmdbResourceGovernance.resource_id.in_(resource_ids), ))).all()) governance_by_resource = {row.resource_id: row for row in governance_rows} open_finding_resource_ids = set((await db.scalars(select( CmdbResourceFinding.resource_id, ).where( CmdbResourceFinding.resource_id.in_(resource_ids), CmdbResourceFinding.status.in_(_ACTIVE_FINDING_STATUSES), ).distinct())).all()) effective_owners = await resolve_effective_resource_owners(db, resources) for resource in resources: resource_id = str(resource.id) governance = governance_by_resource.get(resource_id) governed_owner = bool( governance and ( governance.business_owner_person_id or governance.technical_owner_person_id ) ) counts = ( 1, int(resource.deleted_at is None), int(resource.deleted_at is not None), int(resource_id not in effective_owners and not governed_owner), int(resource.sensitivity == "restricted"), int(bool(governance and governance.risk_level in {"high", "critical"})), int(resource_id in open_finding_resource_ids), ) for fields in dimensions: field_values = tuple( str(getattr(resource, field)) if getattr(resource, field) is not None else None for field in fields ) aggregate = aggregates.setdefault((fields, field_values), [0] * len(counts)) for index, count in enumerate(counts): aggregate[index] += count # Read the cursor before committing: the app session sets # ``expire_on_commit=False`` today, but a session that expired these # instances would turn this attribute access into a lazy refresh -- # i.e. MissingGreenlet on the async engine. last_id = resources[-1].id await db.commit() written = 0 for (fields, field_values), rollup_counts in aggregates.items(): values = dict(zip(fields, field_values, strict=True)) insert = mysql_insert(CmdbCloudResourceDailyRollup).values( id=str(uuid.uuid4()), snapshot_date=day, dimension_key=_dimension_key(values), provider=values.get("provider"), category=values.get("category"), resource_type=values.get("resource_type"), scope_external_id=values.get("scope_external_id"), total_count=rollup_counts[0], active_count=rollup_counts[1], missing_count=rollup_counts[2], unowned_count=rollup_counts[3], restricted_count=rollup_counts[4], high_risk_count=rollup_counts[5], open_finding_count=rollup_counts[6], ).on_duplicate_key_update( total_count=rollup_counts[0], active_count=rollup_counts[1], missing_count=rollup_counts[2], unowned_count=rollup_counts[3], restricted_count=rollup_counts[4], high_risk_count=rollup_counts[5], open_finding_count=rollup_counts[6], updated_at=datetime.now(UTC), ) await db.execute(insert) written += 1 await db.commit() return {"rollups_written": written} async def evaluate_and_rollup(db: AsyncSession) -> dict[str, int]: findings = await evaluate_governance(db) rollups = await refresh_daily_rollups(db) return {**findings, **rollups}