"""Unified, permission-scoped cloud resource directory APIs.""" from __future__ import annotations import json import uuid from datetime import UTC, date, datetime, timedelta from typing import Any from fastapi import APIRouter, Depends, Query, Request from fastapi.encoders import jsonable_encoder from sqlalchemy import case, exists, func, or_, select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.auth.models import AuthUser from app.core.database import get_db from app.core.dependencies import get_current_user from app.core.dependencies import require_active_employee_for_mutations as _leaver_gate from app.core.exceptions import ConflictError, ForbiddenError, NotFoundError, ValidationError from app.core.idempotent_route import idempotent from app.core.rbac import ( AUDITOR_ROLE, CLOUD_ADMIN_ROLE, GLOBAL_ADMIN_ROLE, can_manage_cloud, can_view_cloud_resources, has_role, ) from app.core.response import paginated_response, success_response from app.domains.cmdb.models import ( CmdbCloudResource, CmdbCloudResourceChange, CmdbCloudResourceDailyRollup, CmdbResourceFinding, CmdbResourceGovernance, CmdbResourceGovernanceRule, CmdbResourceType, ) from app.domains.cmdb.resource_directory import sanitize_metadata from app.domains.cmdb.resource_directory_coverage import ( ResourceCoverage, resolve_resource_coverage, ) from app.domains.cmdb.resource_directory_ownership import ( EffectiveResourceOwner, resolve_effective_resource_owners, ) from app.domains.cmdb.resource_directory_schemas import ( FindingAction, GovernancePatch, GovernanceRuleCreate, GovernanceRulePatch, ) from app.domains.cmdb.resource_directory_visibility import ( DirectoryVisibility, directory_visibility_clause, resolve_directory_visibility, ) from app.domains.cmdb.resource_governance import SUPPORTED_EVALUATORS from app.identity.person_models import IdentityPerson router = APIRouter( prefix="/api/v1/cmdb/resource-directory", tags=["cmdb-resource-directory"], dependencies=[Depends(_leaver_gate)], ) _FINDING_ACTIVE = ("open", "acknowledged") _SORTS: dict[str, Any] = { "last_seen_desc": CmdbCloudResource.last_seen_at.desc(), "last_changed_desc": CmdbCloudResource.last_changed_at.desc(), "name_asc": CmdbCloudResource.name.asc(), "name_desc": CmdbCloudResource.name.desc(), "provider_asc": CmdbCloudResource.provider.asc(), } def _escape_like(value: str) -> str: return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") async def _visibility(db: AsyncSession, user: AuthUser) -> DirectoryVisibility: if not can_view_cloud_resources(user.role_names): raise ForbiddenError() return await resolve_directory_visibility( db, user_id=str(user.id), role_names=user.role_names, ) def _can_view_raw_metadata(user: AuthUser) -> bool: return bool(has_role( user.role_names, GLOBAL_ADMIN_ROLE, CLOUD_ADMIN_ROLE, AUDITOR_ROLE, )) def _can_read_full_directory(user: AuthUser) -> bool: return _can_view_raw_metadata(user) async def _person_labels(db: AsyncSession, person_ids: list[str | None]) -> dict[str, str]: clean = {person_id for person_id in person_ids if person_id} if not clean: return {} rows = (await db.execute(select( IdentityPerson.id, IdentityPerson.display_name, IdentityPerson.work_email, ).where(IdentityPerson.id.in_(clean)))).all() return { str(person_id): display_name or work_email or str(person_id) for person_id, display_name, work_email in rows } def _governance_dict(row: CmdbResourceGovernance | None) -> dict[str, Any] | None: if row is None: return None return { "id": str(row.id), "version": row.version, "business_owner_person_id": row.business_owner_person_id, "technical_owner_person_id": row.technical_owner_person_id, "governance_status": row.governance_status, "criticality": row.criticality, "data_classification": row.data_classification, "risk_level": row.risk_level, "lifecycle_stage": row.lifecycle_stage, "compliance_status": row.compliance_status, "exception_reason": row.exception_reason, "exception_expires_at": row.exception_expires_at, "notes": row.notes, "custom_fields": row.custom_fields or {}, "updated_by_person_id": row.updated_by_person_id, "updated_at": row.updated_at, } def _resource_dict( row: CmdbCloudResource, governance: CmdbResourceGovernance | None, *, open_findings: int = 0, effective_owner: EffectiveResourceOwner | None = None, coverage: ResourceCoverage | None = None, ) -> dict[str, Any]: return { "id": str(row.id), "integration_id": row.integration_id, "provider": row.provider, "service": row.service, "category": row.category, "resource_type": row.resource_type, "provider_resource_id": row.provider_resource_id, "name": row.name, "scope_external_id": row.scope_external_id, "region": row.region, "zone": row.zone, "lifecycle_status": row.lifecycle_status, "environment": row.environment, "owner_person_id": row.owner_person_id, "sensitivity": row.sensitivity, "effective_owner_person_id": effective_owner.person_id if effective_owner else None, "owner_source": effective_owner.source if effective_owner else None, "inherited_from_scope_id": ( effective_owner.scope_id if effective_owner and effective_owner.source != "resource" else None ), "inherited_from_scope_name": ( effective_owner.scope_name if effective_owner and effective_owner.source != "resource" else None ), "tags": row.tags or {}, "first_seen_at": row.first_seen_at, "last_seen_at": row.last_seen_at, "last_changed_at": row.last_changed_at, "missing_since": row.missing_since, "deleted_at": row.deleted_at, "governance": _governance_dict(governance), "open_finding_count": open_findings, "monitoring_status": coverage.monitoring_status if coverage else None, "monitoring_latest_at": coverage.monitoring_latest_at if coverage else None, "cost_status": coverage.cost_status if coverage else None, "cost_source": coverage.cost_source if coverage else None, } def _filters( *, q: str | None, providers: list[str] | None, categories: list[str] | None, resource_types: list[str] | None, statuses: list[str] | None, sensitivities: list[str] | None, integration_id: str | None, scope_external_id: str | None, include_deleted: bool, ) -> list[Any]: clauses: list[Any] = [] if scope_external_id and not integration_id: # The Azure resource-group branch reads bounded JSON metadata. Never # allow it to become a fleet-wide scan; embedded scope workbenches # always know their connector id. raise ValidationError("integration_id is required with scope_external_id") if not include_deleted: clauses.append(CmdbCloudResource.deleted_at.is_(None)) if q and q.strip(): pattern = f"%{_escape_like(q.strip()[:200])}%" clauses.append(or_( CmdbCloudResource.name.ilike(pattern, escape="\\"), CmdbCloudResource.provider_resource_id.ilike(pattern, escape="\\"), CmdbCloudResource.scope_external_id.ilike(pattern, escape="\\"), )) if providers: clauses.append(CmdbCloudResource.provider.in_(providers[:20])) if categories: clauses.append(CmdbCloudResource.category.in_(categories[:20])) if resource_types: clauses.append(CmdbCloudResource.resource_type.in_(resource_types[:50])) if statuses: clauses.append(CmdbCloudResource.lifecycle_status.in_(statuses[:20])) if sensitivities: clauses.append(CmdbCloudResource.sensitivity.in_(sensitivities[:10])) if integration_id: clauses.append(CmdbCloudResource.integration_id == integration_id) if scope_external_id: # Azure resource-group scopes live in provider metadata while the # canonical scope column keeps the parent subscription id. clauses.append(or_( CmdbCloudResource.scope_external_id == scope_external_id, CmdbCloudResource.raw_metadata["resource_group"].as_string() == scope_external_id, )) return clauses @router.get("/resources") async def list_resources( request: Request, q: str | None = Query(default=None, max_length=200), provider: list[str] | None = Query(default=None), category: list[str] | None = Query(default=None), resource_type: list[str] | None = Query(default=None), status: list[str] | None = Query(default=None), sensitivity: list[str] | None = Query(default=None), governance_status: list[str] | None = Query(default=None), finding_severity: list[str] | None = Query(default=None), integration_id: str | None = Query(default=None, max_length=36), scope_external_id: str | None = Query(default=None, max_length=256), include_deleted: bool = False, sort: str = Query(default="last_seen_desc"), page: int = Query(default=1, ge=1), page_size: int = Query(default=50, ge=1, le=200), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): vis = await _visibility(db, user) clauses = [directory_visibility_clause(vis), *_filters( q=q, providers=provider, categories=category, resource_types=resource_type, statuses=status, sensitivities=sensitivity, integration_id=integration_id, scope_external_id=scope_external_id, include_deleted=include_deleted, )] if governance_status: clauses.append(CmdbResourceGovernance.governance_status.in_(governance_status[:10])) if finding_severity: clauses.append(exists(select(CmdbResourceFinding.id).where( CmdbResourceFinding.resource_id == CmdbCloudResource.id, CmdbResourceFinding.status.in_(_FINDING_ACTIVE), CmdbResourceFinding.severity.in_(finding_severity[:10]), ).correlate(CmdbCloudResource))) total = int(await db.scalar(select(func.count(CmdbCloudResource.id)).select_from( CmdbCloudResource, ).outerjoin( CmdbResourceGovernance, CmdbResourceGovernance.resource_id == CmdbCloudResource.id, ).where(*clauses)) or 0) finding_count = select( CmdbResourceFinding.resource_id, func.count(CmdbResourceFinding.id).label("finding_count"), ).where(CmdbResourceFinding.status.in_(_FINDING_ACTIVE)).group_by( CmdbResourceFinding.resource_id, ).subquery() rows = (await db.execute(select( CmdbCloudResource, CmdbResourceGovernance, func.coalesce(finding_count.c.finding_count, 0), ).outerjoin( CmdbResourceGovernance, CmdbResourceGovernance.resource_id == CmdbCloudResource.id, ).outerjoin( finding_count, finding_count.c.resource_id == CmdbCloudResource.id, ).where(*clauses).order_by( _SORTS.get(sort, _SORTS["last_seen_desc"]), CmdbCloudResource.id, ).offset((page - 1) * page_size).limit(page_size))).all() effective_owners = await resolve_effective_resource_owners( db, [resource for resource, _, _ in rows], ) coverage = await resolve_resource_coverage( db, [resource for resource, _, _ in rows], ) label_ids: list[str | None] = [] for resource, governance, _ in rows: effective_owner = effective_owners.get(str(resource.id)) label_ids.extend(( resource.owner_person_id, effective_owner.person_id if effective_owner else None, governance.business_owner_person_id if governance else None, governance.technical_owner_person_id if governance else None, )) labels = await _person_labels(db, label_ids) data: list[dict[str, Any]] = [] for resource, governance, finding_total in rows: effective_owner = effective_owners.get(str(resource.id)) item = _resource_dict( resource, governance, open_findings=int(finding_total or 0), effective_owner=effective_owner, coverage=coverage.get(str(resource.id)), ) item["owner_label"] = labels.get(str(resource.owner_person_id)) if resource.owner_person_id else None item["effective_owner_label"] = ( labels.get(effective_owner.person_id) if effective_owner else None ) if item["governance"]: item["governance"]["business_owner_label"] = labels.get(str(governance.business_owner_person_id)) item["governance"]["technical_owner_label"] = labels.get(str(governance.technical_owner_person_id)) data.append(item) return paginated_response(data, total, page, page_size, request) @router.get("/resources/{resource_id}") async def get_resource( resource_id: uuid.UUID, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): vis = await _visibility(db, user) result = (await db.execute(select( CmdbCloudResource, CmdbResourceGovernance, ).outerjoin( CmdbResourceGovernance, CmdbResourceGovernance.resource_id == CmdbCloudResource.id, ).where( CmdbCloudResource.id == resource_id, directory_visibility_clause(vis), ))).one_or_none() if result is None: raise NotFoundError("CloudResource", str(resource_id)) resource, governance = result finding_total = int(await db.scalar(select(func.count(CmdbResourceFinding.id)).where( CmdbResourceFinding.resource_id == str(resource.id), CmdbResourceFinding.status.in_(_FINDING_ACTIVE), )) or 0) effective_owner = (await resolve_effective_resource_owners(db, [resource])).get(str(resource.id)) coverage = (await resolve_resource_coverage(db, [resource])).get(str(resource.id)) data = _resource_dict( resource, governance, open_findings=finding_total, effective_owner=effective_owner, coverage=coverage, ) data["normalized_metadata"] = resource.normalized_metadata or {} data["raw_metadata"] = resource.raw_metadata or {} if _can_view_raw_metadata(user) else None labels = await _person_labels(db, [ resource.owner_person_id, effective_owner.person_id if effective_owner else None, governance.business_owner_person_id if governance else None, governance.technical_owner_person_id if governance else None, governance.updated_by_person_id if governance else None, ]) data["owner_label"] = labels.get(str(resource.owner_person_id)) if resource.owner_person_id else None data["effective_owner_label"] = labels.get(effective_owner.person_id) if effective_owner else None if data["governance"]: data["governance"]["business_owner_label"] = labels.get(str(governance.business_owner_person_id)) data["governance"]["technical_owner_label"] = labels.get(str(governance.technical_owner_person_id)) data["governance"]["updated_by_label"] = labels.get(str(governance.updated_by_person_id)) return success_response(data=data, request=request) @router.get("/resources/{resource_id}/history") async def get_resource_history( resource_id: uuid.UUID, request: Request, page: int = Query(default=1, ge=1), page_size: int = Query(default=50, ge=1, le=200), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): vis = await _visibility(db, user) visible = await db.scalar(select(CmdbCloudResource.id).where( CmdbCloudResource.id == resource_id, directory_visibility_clause(vis), )) if visible is None: raise NotFoundError("CloudResource", str(resource_id)) where = CmdbCloudResourceChange.resource_id == str(resource_id) total = int(await db.scalar(select(func.count(CmdbCloudResourceChange.id)).where(where)) or 0) rows = (await db.scalars(select(CmdbCloudResourceChange).where(where).order_by( CmdbCloudResourceChange.observed_at.desc(), CmdbCloudResourceChange.id.desc(), ).offset((page - 1) * page_size).limit(page_size))).all() reveal = _can_view_raw_metadata(user) data = [{ "id": str(row.id), "change_type": row.change_type, "changed_fields": row.changed_fields or [], "previous_hash": row.previous_hash, "current_hash": row.current_hash, "snapshot": row.snapshot if reveal else None, "source_sync_job_id": row.source_sync_job_id, "observed_at": row.observed_at, } for row in rows] return paginated_response(data, total, page, page_size, request) @router.get("/types") async def list_resource_types( request: Request, provider: str | None = Query(default=None, max_length=64), category: str | None = Query(default=None, max_length=64), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): vis = await _visibility(db, user) visible_types = select(CmdbCloudResource.resource_type_id).where( directory_visibility_clause(vis), ).distinct() clauses: list[Any] = [CmdbResourceType.id.in_(visible_types), CmdbResourceType.active.is_(True)] if provider: clauses.append(CmdbResourceType.provider == provider) if category: clauses.append(CmdbResourceType.category == category) rows = (await db.scalars(select(CmdbResourceType).where(*clauses).order_by( CmdbResourceType.provider, CmdbResourceType.category, CmdbResourceType.display_name, ))).all() return success_response(data=[{ "id": str(row.id), "provider": row.provider, "resource_type": row.resource_type, "service": row.service, "category": row.category, "display_name": row.display_name, "icon_key": row.icon_key, "sensitivity": row.sensitivity, "display_schema": row.display_schema or {}, "discovered_dynamically": row.discovered_dynamically, } for row in rows], request=request) @router.get("/summary") async def get_summary( request: Request, provider: list[str] | None = Query(default=None), category: list[str] | None = Query(default=None), integration_id: str | None = Query(default=None, max_length=36), scope_external_id: str | None = Query(default=None, max_length=256), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): vis = await _visibility(db, user) clauses = [directory_visibility_clause(vis), *_filters( q=None, providers=provider, categories=category, resource_types=None, statuses=None, sensitivities=None, integration_id=integration_id, scope_external_id=scope_external_id, include_deleted=True, )] open_finding = exists(select(CmdbResourceFinding.id).where( CmdbResourceFinding.resource_id == CmdbCloudResource.id, CmdbResourceFinding.status.in_(_FINDING_ACTIVE), ).correlate(CmdbCloudResource)) def active_rule_finding(rule_key: str) -> Any: return exists(select(CmdbResourceFinding.id).join( CmdbResourceGovernanceRule, CmdbResourceGovernanceRule.id == CmdbResourceFinding.rule_id, ).where( CmdbResourceFinding.resource_id == CmdbCloudResource.id, CmdbResourceFinding.status.in_(_FINDING_ACTIVE), CmdbResourceGovernanceRule.rule_key == rule_key, ).correlate(CmdbCloudResource)) base = select( func.count(CmdbCloudResource.id), func.sum(case((CmdbCloudResource.deleted_at.is_(None), 1), else_=0)), func.sum(case((CmdbCloudResource.deleted_at.is_not(None), 1), else_=0)), func.sum(case((CmdbCloudResource.sensitivity == "restricted", 1), else_=0)), func.sum(case((open_finding, 1), else_=0)), func.sum(case((active_rule_finding("resource.monitoring_missing"), 1), else_=0)), func.sum(case((active_rule_finding("resource.cost_unavailable"), 1), else_=0)), ).where(*clauses) total, active, missing, restricted, with_findings, monitoring_gaps, cost_gaps = ( await db.execute(base) ).one() async def grouped(column: Any, limit: int = 30) -> list[dict[str, Any]]: rows = (await db.execute(select(column, func.count(CmdbCloudResource.id)).where( *clauses, ).group_by(column).order_by(func.count(CmdbCloudResource.id).desc()).limit(limit))).all() return [{"key": key or "unknown", "count": int(count)} for key, count in rows] data = { "total": int(total or 0), "active": int(active or 0), "missing": int(missing or 0), "restricted": int(restricted or 0), "with_open_findings": int(with_findings or 0), "monitoring_gaps": int(monitoring_gaps or 0), "cost_gaps": int(cost_gaps or 0), "by_provider": await grouped(CmdbCloudResource.provider), "by_category": await grouped(CmdbCloudResource.category), "by_resource_type": await grouped(CmdbCloudResource.resource_type, 50), "by_status": await grouped(CmdbCloudResource.lifecycle_status), } return success_response(data=data, request=request) @router.patch("/resources/{resource_id}/governance") async def update_governance( resource_id: uuid.UUID, body: GovernancePatch, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): if not can_manage_cloud(user.role_names): raise ForbiddenError() values = body.model_dump(exclude_unset=True, exclude={"version"}) if not values: raise ValidationError("At least one governance field must be supplied") resource = await db.get(CmdbCloudResource, resource_id) if resource is None: raise NotFoundError("CloudResource", str(resource_id)) row = await db.scalar(select(CmdbResourceGovernance).where( CmdbResourceGovernance.resource_id == str(resource_id), ).with_for_update()) if row is not None and (body.version is None or body.version != row.version): raise ConflictError("Governance record changed; reload before saving") owner_ids = { str(value) for value in (body.business_owner_person_id, body.technical_owner_person_id) if value is not None } if owner_ids: found = set(str(value) for value in (await db.scalars(select(IdentityPerson.id).where( IdentityPerson.id.in_(owner_ids), IdentityPerson.employment_status != "left", ))).all()) if found != owner_ids: raise ValidationError("Governance owners must be active Helix people") before = _governance_dict(row) if row is None: now = datetime.now(UTC).replace(tzinfo=None) row = CmdbResourceGovernance( resource_id=str(resource_id), updated_by_person_id=str(user.id), created_at=now, updated_at=now, ) db.add(row) if "custom_fields" in values: values["custom_fields"] = sanitize_metadata(values["custom_fields"] or {}) if len(json.dumps(values["custom_fields"], ensure_ascii=False, default=str)) > 32_768: raise ValidationError("custom_fields is too large") for key, value in values.items(): if key.endswith("_person_id") and value is not None: value = str(value) if key == "exception_expires_at" and value is not None: value = value.astimezone(UTC).replace(tzinfo=None) setattr(row, key, value) if row.governance_status == "exception_approved": if not (row.exception_reason or "").strip(): raise ValidationError("exception_reason is required for exception_approved") now = datetime.now(UTC).replace(tzinfo=None) if row.exception_expires_at is None or row.exception_expires_at <= now: raise ValidationError("exception_approved requires a future exception_expires_at") if row.exception_expires_at > now + timedelta(days=366): raise ValidationError("Governance exception cannot exceed 366 days") row.updated_by_person_id = str(user.id) row.updated_at = datetime.now(UTC).replace(tzinfo=None) row.version = (row.version or 0) + 1 if before is not None else 1 try: await db.flush() except IntegrityError as exc: await db.rollback() raise ConflictError("Governance record was created concurrently; reload and retry") from exc after = _governance_dict(row) await AuditService(db).log( trace_id=getattr(request.state, "trace_id", str(uuid.uuid4())), actor_type="user", actor_user_id=user.id, action="cmdb.resource_governance.update", status="success", resource_type="cmdb_cloud_resource", resource_id=str(resource_id), detail=jsonable_encoder({"before": before, "after": after}), ) await db.commit() return success_response(data=after, request=request) def _finding_dict( finding: CmdbResourceFinding, resource: CmdbCloudResource, rule: CmdbResourceGovernanceRule, ) -> dict[str, Any]: return { "id": str(finding.id), "resource_id": str(resource.id), "resource_name": resource.name, "provider": resource.provider, "category": resource.category, "resource_type": resource.resource_type, "rule_id": str(rule.id), "rule_key": rule.rule_key, "rule_name": rule.name, "severity": finding.severity, "status": finding.status, "title": finding.title, "description": finding.description, "evidence": finding.evidence or {}, "recommendation": finding.recommendation, "assigned_to_person_id": finding.assigned_to_person_id, "due_at": finding.due_at, "first_seen_at": finding.first_seen_at, "last_seen_at": finding.last_seen_at, "acknowledged_at": finding.acknowledged_at, "resolved_at": finding.resolved_at, "resolution_note": finding.resolution_note, } @router.get("/findings") async def list_findings( request: Request, status: list[str] | None = Query(default=None), severity: list[str] | None = Query(default=None), rule_key: str | None = Query(default=None, max_length=96), assigned_to_person_id: str | None = Query(default=None, max_length=36), page: int = Query(default=1, ge=1), page_size: int = Query(default=50, ge=1, le=200), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): vis = await _visibility(db, user) clauses: list[Any] = [directory_visibility_clause(vis)] if status: clauses.append(CmdbResourceFinding.status.in_(status[:10])) if severity: clauses.append(CmdbResourceFinding.severity.in_(severity[:10])) if rule_key: clauses.append(CmdbResourceGovernanceRule.rule_key == rule_key) if assigned_to_person_id: clauses.append(CmdbResourceFinding.assigned_to_person_id == assigned_to_person_id) base = select(CmdbResourceFinding, CmdbCloudResource, CmdbResourceGovernanceRule).join( CmdbCloudResource, CmdbCloudResource.id == CmdbResourceFinding.resource_id, ).join( CmdbResourceGovernanceRule, CmdbResourceGovernanceRule.id == CmdbResourceFinding.rule_id, ).where(*clauses) total = int(await db.scalar(select(func.count()).select_from(base.subquery())) or 0) rows = (await db.execute(base.order_by( case( (CmdbResourceFinding.severity == "critical", 0), (CmdbResourceFinding.severity == "high", 1), (CmdbResourceFinding.severity == "medium", 2), else_=3, ), CmdbResourceFinding.last_seen_at.desc(), ).offset((page - 1) * page_size).limit(page_size))).all() data = [_finding_dict(finding, resource, rule) for finding, resource, rule in rows] labels = await _person_labels(db, [item["assigned_to_person_id"] for item in data]) for item in data: item["assigned_to_label"] = labels.get(str(item["assigned_to_person_id"])) return paginated_response(data, total, page, page_size, request) @router.post("/findings/{finding_id}/action") @idempotent( "cmdb.resource_finding.action", key_params=("finding_id", "user.id"), ) async def act_on_finding( finding_id: uuid.UUID, body: FindingAction, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): vis = await _visibility(db, user) result = (await db.execute(select( CmdbResourceFinding, CmdbCloudResource, CmdbResourceGovernanceRule, ).join( CmdbCloudResource, CmdbCloudResource.id == CmdbResourceFinding.resource_id, ).join( CmdbResourceGovernanceRule, CmdbResourceGovernanceRule.id == CmdbResourceFinding.rule_id, ).where( CmdbResourceFinding.id == finding_id, directory_visibility_clause(vis), ).with_for_update())).one_or_none() if result is None: raise NotFoundError("ResourceFinding", str(finding_id)) finding, resource, rule = result admin = can_manage_cloud(user.role_names) if has_role(user.role_names, AUDITOR_ROLE) and not admin: raise ForbiddenError() if body.action != "acknowledge" and not admin: raise ForbiddenError() if (body.assigned_to_person_id or body.due_at) and not admin: raise ForbiddenError() before = {"status": finding.status, "assigned_to_person_id": finding.assigned_to_person_id, "due_at": finding.due_at} now = datetime.now(UTC).replace(tzinfo=None) due_at = ( body.due_at.astimezone(UTC).replace(tzinfo=None) if body.due_at is not None else None ) if body.action == "acknowledge": if finding.status not in {"open", "acknowledged"}: raise ConflictError("Only an open finding can be acknowledged") finding.status = "acknowledged" finding.acknowledged_at = now finding.acknowledged_by_person_id = str(user.id) elif body.action == "reopen": finding.status = "open" finding.resolved_at = None finding.resolution_note = body.note else: if due_at is None or due_at <= now: raise ValidationError("Suppressing a finding requires a future due_at") if due_at > now + timedelta(days=366): raise ValidationError("Finding suppression cannot exceed 366 days") if not (body.note or "").strip(): raise ValidationError("Suppressing a finding requires a note") finding.status = "suppressed" finding.resolution_note = body.note if body.assigned_to_person_id is not None: assignee = await db.get(IdentityPerson, body.assigned_to_person_id) if assignee is None or assignee.employment_status == "left": raise ValidationError("Finding assignee must be an active Helix person") finding.assigned_to_person_id = str(body.assigned_to_person_id) if due_at is not None: finding.due_at = due_at await AuditService(db).log( trace_id=getattr(request.state, "trace_id", str(uuid.uuid4())), actor_type="user", actor_user_id=user.id, action=f"cmdb.resource_finding.{body.action}", status="success", resource_type="cmdb_resource_finding", resource_id=str(finding.id), detail=jsonable_encoder({ "resource_id": str(resource.id), "rule_key": rule.rule_key, "before": before, "after": {"status": finding.status, "assigned_to_person_id": finding.assigned_to_person_id, "due_at": finding.due_at, "note": body.note}, }), ) await db.commit() return success_response(data=_finding_dict(finding, resource, rule), request=request) def _rule_dict(rule: CmdbResourceGovernanceRule) -> dict[str, Any]: return { "id": str(rule.id), "rule_key": rule.rule_key, "name": rule.name, "description": rule.description, "category": rule.category, "severity": rule.severity, "evaluator": rule.evaluator, "applies_to": rule.applies_to or {}, "parameters": rule.parameters or {}, "recommendation": rule.recommendation, "enabled": rule.enabled, "built_in": rule.built_in, "version": rule.version, "created_at": rule.created_at, "updated_at": rule.updated_at, } def _validate_rule_config( evaluator: str, applies_to: dict[str, list[str]] | None, parameters: dict[str, Any] | None, ) -> None: if not applies_to: applies_to = {} else: unknown = set(applies_to) - {"providers", "categories", "resource_types"} if unknown: raise ValidationError( f"Unsupported applies_to dimensions: {', '.join(sorted(unknown))}", ) if any(len(values) > 100 for values in applies_to.values()): raise ValidationError("Each applies_to dimension supports at most 100 values") if any( not value.strip() or len(value) > 192 for values in applies_to.values() for value in values ): raise ValidationError("applies_to values must be 1-192 non-blank characters") parameters = parameters or {} if evaluator == "stale_observation": if set(parameters) - {"hours"}: raise ValidationError("stale_observation only supports the hours parameter") hours = parameters.get("hours", 48) if isinstance(hours, bool) or not isinstance(hours, int) or not 1 <= hours <= 720: raise ValidationError("stale_observation hours must be an integer from 1 to 720") elif parameters: raise ValidationError(f"{evaluator} does not accept parameters") config = {"applies_to": applies_to, "parameters": parameters} if len(json.dumps(config, ensure_ascii=False, separators=(",", ":"))) > 32_768: raise ValidationError("Rule configuration is too large") if sanitize_metadata(parameters) != parameters: raise ValidationError("Rule parameters cannot contain credentials or secret-like values") @router.get("/rules") async def list_rules( request: Request, enabled: bool | None = None, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): if not _can_read_full_directory(user): raise ForbiddenError() stmt = select(CmdbResourceGovernanceRule) if enabled is not None: stmt = stmt.where(CmdbResourceGovernanceRule.enabled.is_(enabled)) rows = (await db.scalars(stmt.order_by( CmdbResourceGovernanceRule.category, CmdbResourceGovernanceRule.name, ))).all() return success_response(data=[_rule_dict(row) for row in rows], request=request) @router.post("/rules", status_code=201) async def create_rule( body: GovernanceRuleCreate, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): if not can_manage_cloud(user.role_names): raise ForbiddenError() if body.evaluator not in SUPPORTED_EVALUATORS: raise ValidationError("Unsupported deterministic evaluator") _validate_rule_config(body.evaluator, body.applies_to, body.parameters) if await db.scalar(select(CmdbResourceGovernanceRule.id).where( CmdbResourceGovernanceRule.rule_key == body.rule_key, )) is not None: raise ConflictError("rule_key already exists") now = datetime.now(UTC).replace(tzinfo=None) row = CmdbResourceGovernanceRule( **body.model_dump(), built_in=False, created_by_person_id=str(user.id), updated_by_person_id=str(user.id), created_at=now, updated_at=now, ) db.add(row) try: await db.flush() except IntegrityError as exc: await db.rollback() raise ConflictError("rule_key already exists") from exc await AuditService(db).log( trace_id=getattr(request.state, "trace_id", str(uuid.uuid4())), actor_type="user", actor_user_id=user.id, action="cmdb.resource_governance_rule.create", status="success", resource_type="cmdb_resource_governance_rule", resource_id=str(row.id), detail=jsonable_encoder(_rule_dict(row)), ) await db.commit() return success_response(data=_rule_dict(row), request=request) @router.patch("/rules/{rule_id}") async def update_rule( rule_id: uuid.UUID, body: GovernanceRulePatch, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): if not can_manage_cloud(user.role_names): raise ForbiddenError() row = await db.scalar(select(CmdbResourceGovernanceRule).where( CmdbResourceGovernanceRule.id == rule_id, ).with_for_update()) if row is None: raise NotFoundError("ResourceGovernanceRule", str(rule_id)) if row.version != body.version: raise ConflictError("Governance rule changed; reload before saving") values = body.model_dump(exclude_unset=True, exclude={"version"}) if not values: raise ValidationError("At least one rule field must be supplied") _validate_rule_config( row.evaluator, values.get("applies_to", row.applies_to), values.get("parameters", row.parameters), ) before = _rule_dict(row) for key, value in values.items(): setattr(row, key, value) row.version += 1 row.updated_by_person_id = str(user.id) row.updated_at = datetime.now(UTC).replace(tzinfo=None) await AuditService(db).log( trace_id=getattr(request.state, "trace_id", str(uuid.uuid4())), actor_type="user", actor_user_id=user.id, action="cmdb.resource_governance_rule.update", status="success", resource_type="cmdb_resource_governance_rule", resource_id=str(row.id), detail=jsonable_encoder({"before": before, "after": _rule_dict(row)}), ) await db.commit() return success_response(data=_rule_dict(row), request=request) @router.get("/rollups") async def list_rollups( request: Request, days: int = Query(default=90, ge=1, le=730), provider: str | None = Query(default=None, max_length=64), category: str | None = Query(default=None, max_length=64), resource_type: str | None = Query(default=None, max_length=192), scope_external_id: str | None = Query(default=None, max_length=256), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): if not _can_read_full_directory(user): raise ForbiddenError() clauses: list[Any] = [ CmdbCloudResourceDailyRollup.snapshot_date >= date.today() - timedelta(days=days - 1), ] values = { "provider": provider, "category": category, "resource_type": resource_type, "scope_external_id": scope_external_id, } for key, value in values.items(): column = getattr(CmdbCloudResourceDailyRollup, key) clauses.append(column == value if value is not None else column.is_(None)) rows = (await db.scalars(select(CmdbCloudResourceDailyRollup).where(*clauses).order_by( CmdbCloudResourceDailyRollup.snapshot_date, ))).all() data = [{ "snapshot_date": row.snapshot_date, "provider": row.provider, "category": row.category, "resource_type": row.resource_type, "scope_external_id": row.scope_external_id, "total_count": row.total_count, "active_count": row.active_count, "missing_count": row.missing_count, "unowned_count": row.unowned_count, "restricted_count": row.restricted_count, "high_risk_count": row.high_risk_count, "open_finding_count": row.open_finding_count, } for row in rows] return success_response(data=data, request=request)