"""Cloud account lifecycle REST endpoints. Mounted under ``/api/v1/cmdb`` alongside the existing router. Three logical blocks: * ``/scopes`` — admin list/create/update/close scopes, assign owner, update budget. * ``/scopes/:id/grants`` — list + create grants (owner-direct), plus ``/approve`` and ``/reject`` for pending requests. * ``/my-scopes`` + ``/my-grants`` — employee self-service: scopes I own (with member + billing + alerts), grants I hold. * ``/billing/daily`` — scope-level daily costs for the drill-down chart. RBAC: ``can_manage_cloud`` for tenant-wide admin writes; ``is_cloud_scope_owner_of`` for per-scope owner actions; the owner of a pending grant can also approve. Employees can read their own grants and request new ones. """ from __future__ import annotations import json as _json import re import uuid from datetime import UTC, date, datetime, timedelta from decimal import Decimal from typing import Any, Literal import sqlalchemy as sa from fastapi import APIRouter, BackgroundTasks, Depends, Query, Request from pydantic import BaseModel, Field, field_validator from sqlalchemy import and_, case, distinct, func, or_, select 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 ( enforce_action_token_subject, get_current_user, require_browser_session, ) from app.core.dependencies import ( require_active_employee_for_mutations as _leaver_runtime_gate, ) from app.core.distributed_lock import LockBusy, distributed_lock from app.core.exceptions import ( AppException, ConflictError, ForbiddenError, NotFoundError, ValidationError, format_exception_for_log, ) from app.core.idempotent_route import idempotent from app.core.rbac import ( can_manage_cloud, can_manage_cloud_scope_lifecycle, can_read_cloud, can_view_cloud_resources, can_view_finance, is_cloud_scope_owner_of, person_owns_any_cloud_scope, ) from app.core.response import paginated_response, success_response from app.domains.cmdb.lifecycle_service import ( CloudLifecycleService, _load_cloud_driver, list_idle_scope_candidates, scope_budget_measure, scope_self_request_block_reason, sync_grants_for_integration, sync_roles_for_integration, sync_roles_for_scope, sync_scopes_for_integration, ) from app.domains.cmdb.models import ( GRANT_STATUS_ACTIVE, GRANT_STATUS_PENDING, GRANT_TTL_DEFAULT_DAYS, GRANT_TTL_HARD_CAP_DAYS, PROD_ENVIRONMENTS, SCOPE_REQUEST_STATUS_PROVISIONED, SCOPE_STATUS_CLOSED, SCOPE_TYPE_AZURE_RESOURCE_GROUP, SCOPE_TYPE_GCP_BILLING_ACCOUNT, SCOPE_TYPE_LEGACY_BILLING_UNIT, SCOPE_TYPE_STANDALONE_IAM, TENANT_LEVEL_SCOPE_TYPES, CmdbCloudAccessGrant, CmdbCloudBillingDaily, CmdbCloudRole, CmdbCloudScope, CmdbCloudScopeRequest, CmdbCostFact, employee_directory_scope_criteria, ) from app.domains.cmdb.prod_write_lock import ( ensure_prod_scope_writable, prod_cloud_writes_locked, ) from app.domains.cmdb.providers.exceptions import redact_provider_error # Sub-tenant scope types — scopes whose cost lives in # ``cmdb_cost_facts.resource_group`` rather than in # ``cmdb_cloud_billing_daily`` (the daily cron writes daily aggregates # only at tenant level; RG-level cost is not provider-returned at # that granularity). When ``list_billing_daily`` is asked for one of # these scope types it routes through cost_facts instead — see the # RG branch in the handler for the SQL. ADD new RG-typed scope_type # constants here when a future provider lands one (e.g. Aliyun # resource_group, GCP folder); the same caveat applies. _RG_SCOPE_TYPES: frozenset[str] = frozenset({ SCOPE_TYPE_AZURE_RESOURCE_GROUP, }) from app.core.safe_logging import get_safe_logger from app.core.txn import savepoint from app.domains.cmdb.scope_request_service import CloudScopeRequestService from app.domains.cmdb.timeline_service import ( fetch_grant_timeline, fetch_scope_request_timeline, ) from app.identity.person_models import IdentityPerson from app.integrations.driver_base import CloudDriver, CloudScopeRef from app.integrations.models import Integration logger = get_safe_logger(__name__) router = APIRouter( prefix="/api/v1/cmdb", tags=["cmdb-lifecycle"], dependencies=[Depends(_leaver_runtime_gate)], ) def _utc_now_naive() -> datetime: # MySQL/TiDB returns naive datetimes for ``DateTime(timezone=True)`` # columns. Use a naive UTC ``now`` for arithmetic against them so the # subtraction doesn't raise ``can't subtract offset-naive and # offset-aware datetimes`` and the serialized output stays naive # (matching the rest of the codebase's convention). return datetime.now(UTC).replace(tzinfo=None) # ── Request / response schemas ───────────────────────────────── class ScopeResponse(BaseModel): id: uuid.UUID integration_id: str integration_name: str | None = None provider: str | None = None scope_type: str external_id: str display_name: str | None = None owner_person_id: str | None = None owner_label: str | None = None # Leaver visibility: 工号 + 在职状态 for PersonTag on the scope owner. owner_employee_no: str | None = None owner_employment_status: str | None = None # Cost-attribution department: the owner's full org chain (BU / 大组 / # 小组), leaf last, tenant root already stripped. Sourced from # ``IdentityPerson.department_chain``. Empty list when unowned or the # owner has no primary org membership. owner_department_chain: list[str] = Field(default_factory=list) budget_usd_monthly: float | None = None # Cost basis the budget figure is denominated in — ``"net"`` (实付/发票) # or ``"list"`` (业务折前). Always present so the UI can label the # number instead of rendering a bare USD figure whose basis is # unknowable; percentage alert rules inherit it by construction. budget_measure: str = "net" auto_suspend_after_days: int default_grant_ttl_days: int status: str environment: str = "dev" notes: str | None = None # Account contact email (migration 450). ``cloud_account_email`` is # the value the cloud driver last reported (AWS Organizations account # ``Email``) — read-only, refreshed every sync, rendered as "云上值". # ``account_email_override`` is the owner/admin replacement (null = # use cloud). ``account_email`` is the resolved effective value the # UI shows by default (override or, failing that, cloud). The # ``*_modified_*`` trio records who last set/cleared the override so # the detail page can show "谁改成了什么" — label is "name · email". cloud_account_email: str | None = None account_email_override: str | None = None account_email: str | None = None account_email_modified_by_person_id: str | None = None account_email_modified_by_label: str | None = None account_email_modified_at: datetime | None = None active_grant_count: int = 0 pending_grant_count: int = 0 active_member_count: int = 0 pending_member_count: int = 0 last_synced_at: datetime # Manual-decommission bookkeeping. ``None`` for active scopes. decommissioned_at: datetime | None = None decommissioned_by: str | None = None decommissioned_by_label: str | None = None decommission_reason: str | None = None decommission_provider_state: str | None = None # In-flight takedown (migration 624): a confirmed decommission the # provider has not accepted yet because Helix is still deleting the # blockers the admin confirmed. ``status`` stays ``active`` for this # whole window, so these three are the only signal that a takedown # is under way. decommission_pending_since: datetime | None = None decommission_cleanup_state: str | None = None decommission_cleanup_error: str | None = None # Pivot URL into the cloud vendor's web console — resolved by # ``console_urls.resolve_console_url`` (Integration.config_public # override → platform_settings global → builtin baseline). # ``None`` means the provider has no known console; the frontend # hides the "打开 Console" button. console_url: str | None = None model_config = {"from_attributes": True} class ScopeUpdateRequest(BaseModel): owner_person_id: uuid.UUID | None = None # Owner-transfer modifier: when true AND ``owner_person_id`` is set, # also revoke the OUTGOING owner's owner-equivalent grant on this scope # (the access they held by virtue of being owner). Their unrelated # member grants are left intact. Surfaced as an opt-in checkbox in the # transfer dialog; ignored on non-transfer PATCHes. Prod scopes skip # the revoke (prod IAM stays under the write lock). revoke_previous_owner_grant: bool = False budget_usd_monthly: Decimal | None = Field(default=None, ge=0, le=Decimal("99999999.99")) # Cost basis the submitted ``budget_usd_monthly`` is denominated in. # Omitted is fine for a scope whose budget is (still) on the invoice # basis; on a scope already recorded as ``list`` the service REFUSES a # number without an explicit basis (400 CLOUD_BUDGET_MEASURE_REQUIRED) — # guessing is what produced the 2026-07 false-alert incident. Can also # be sent alone to correct a mis-recorded basis without touching the # number (percentage rules are re-denominated either way). budget_measure: Literal["net", "list"] | None = None auto_suspend_after_days: int | None = Field(default=None, ge=7, le=365) default_grant_ttl_days: int | None = Field(default=None, ge=1, le=365) notes: str | None = Field(default=None, max_length=4000) # Owner/admin override of the cloud-reported account email. Semantics: # * omitted (None) → leave the override untouched # * non-empty string → set the override (must be a plausible email) # * empty string ("") → CLEAR the override, reverting to the cloud # value ("云上值") # Only a scope owner or cloud/global admin reaches this route, so no # extra role gate is needed (unlike environment/status, which are # admin-only). Pure Helix metadata — never pushed to the provider. account_email_override: str | None = Field(default=None, max_length=320) # Status is intentionally locked to a Literal — admins must NOT use # PATCH to flip ``closed → active`` (that bypasses the # ``decommission_scope`` provider-side close + audit trail + grant- # disable cascade). The service layer additionally rejects illegal # transitions; this Literal is the schema-level gate that catches # typos like ``"closd"`` (which would otherwise fall out of every # status filter, including auto-revoke / billing / sweep). status: Literal["active", "suspended"] | None = None # Lifecycle environment — must be one of {dev, staging, prod}. Setting # ``prod`` opts the scope OUT of the T+0 expiry auto-revoke; the # owner is notified instead. Default at creation time is ``dev``. environment: str | None = Field(default=None, pattern="^(dev|staging|prod)$") @field_validator("account_email_override") @classmethod def _validate_account_email_override(cls, v: str | None) -> str | None: # None = untouched; "" = clear (kept as-is, the service maps it to # NULL). A non-empty value must be a plausible single email — a # loose ``local@domain.tld`` shape, not RFC-5322-complete, enough # to reject a fat-fingered display name or a copied-in list. if v is None: return v v = v.strip() if v == "": return "" if not re.fullmatch(r"[^@\s]+@[^@\s]+\.[^@\s]+", v): raise ValueError( "account_email_override must be a valid email address", ) return v class AlertRuleInput(BaseModel): # Strict literal — previously accepted any string, including typos # that silently created un-evaluatable rules. type: Literal["daily", "monthly"] # ``threshold_usd`` is the absolute USD figure the evaluator # compares against. Either it OR ``threshold_pct`` (or both) must # be supplied; ``threshold_pct`` wins when both are present and # the scope has a ``budget_usd_monthly`` (the service recomputes # ``threshold_usd`` from ``pct × budget``). 0 / unset is allowed # only when ``threshold_pct`` is present. threshold_usd: float = Field(default=0.0, ge=0, le=1e9) threshold_pct: float | None = Field(default=None, ge=0, le=1000) # cc_person_ids capped at 50 — abuse vector: a scope owner could # otherwise weaponize this list to spam C-suite UUIDs (or any # arbitrary person) with budget alerts. The router-level cc validator # additionally checks each id against active IdentityPerson rows. cc_person_ids: list[uuid.UUID] = Field(default_factory=list, max_length=50) active: bool = True class SetAlertRulesRequest(BaseModel): # Cap at 50 rules to bound the daily evaluator's per-scope cost. rules: list[AlertRuleInput] = Field(max_length=50) class DecommissionScopeRequest(BaseModel): reason: str = Field(min_length=1, max_length=500) # Opt-in: "if the provider refuses because the scope still holds # resources, delete the ones I confirmed and finish the job". Default # ``False`` keeps every existing caller (MCP-less admin scripts, # tests, the mobile page until it ships the preflight UI) on the old # behaviour — an automatic deletion must never be something a client # gets by omission. cleanup_blockers: bool = False # The blocker keys the dialog SHOWED the admin. This is the consent, # and cleanup never touches anything outside it: a resource group # created between rendering the dialog and confirming it is not in # this list, so the backend stops and asks for a fresh confirmation # instead of deleting it. Capped generously — a subscription with # more than 200 resource groups is not something to sweep in one # click anyway. confirmed_blocker_keys: list[str] = Field( default_factory=list, max_length=200, ) class SnoozeIdleScopeRequest(BaseModel): """Body for ``POST /cmdb/scopes/{scope_id}/snooze-idle``. ``days`` defaults to 30 (the duration paired with the 「忽略 30 天」 button in the detail page). Bounded 1..365 by the service to refuse pathological "10-year mute" snoozes that would be indistinguishable from a decommission in the audit log. """ days: int = Field(default=30, ge=1, le=365) class IdleScopeCandidateResponse(BaseModel): scope_id: str integration_id: str integration_name: str | None provider: str | None scope_type: str external_id: str display_name: str | None owner_person_id: str | None last_synced_at: datetime days_since_any_grant_activity: int | None cost_30d_usd: float class GrantResponse(BaseModel): id: uuid.UUID scope_id: str scope_display_name: str | None = None scope_external_id: str | None = None # Lifecycle classification of the parent scope. The frontend reads # this so it can suppress the "expires in N days" alarm on # production scopes — those grants are explicitly held past # expires_at and are owner-managed (see ``sweep_expiring_grants`` # in lifecycle_service). ``None`` only when the scope is gone. scope_environment: str | None = None integration_id: str | None = None integration_name: str | None = None provider: str | None = None # Nullable since migration 131 — orphan-cleanup history rows # carry no Helix person reference. See ``external_principal_*`` # below for the cloud-side identifiers we keep instead. person_id: str | None = None person_label: str | None = None # Leaver visibility: 工号 + 在职状态 for PersonTag on the grantee. # ``None`` on orphan-cleanup history rows (person_id IS NULL). person_employee_no: str | None = None person_employment_status: str | None = None # Helix platform roles held by ``person_id`` (``cloud_admin``, # ``it_admin``, etc.). The frontend uses this to suppress the # "expires in N days" chip for cloud / global admins, whose access # to a scope is held perpetually as part of their job — showing an # expiry countdown for them is misleading. ``[]`` for plain # employees. person_role_names: list[str] = Field(default_factory=list) # Set only on orphan-cleanup history rows (person_id IS NULL). # The cloud-side principal id (e.g. AWS Identity Store user UUID) # plus a best-effort recovered email so the admin "已结束" tab # can attribute the swept assignment. external_principal_id: str | None = None external_principal_email: str | None = None role_key: str role_display_name: str | None = None status: str granted_at: datetime expires_at: datetime | None = None last_renewed_at: datetime | None = None last_login_at: datetime | None = None revoked_at: datetime | None = None revoke_reason: str | None = None # Unified "this grant has ended" descriptors. ``revoked_at`` / # ``revoke_reason`` are only populated for ``status='revoked'`` rows # (terminal human decision; see DECISIONS.md "revoked vs disabled"). # ``status='disabled'`` rows leave the ``revoked_*`` columns NULL on # purpose and stash the disable timestamp + reason in # ``cmdb_cloud_access_grants.detail.disable_result``. Frontend lists # showing an "ended at" / "ended reason" column should read these # virtual fields, which the serializer fills from whichever column # actually carries the data for the row's status (revoke / disable # / expire). Older clients still get the raw ``revoked_*`` fields. ended_at: datetime | None = None ended_reason: str | None = None external_ref: str | None = None days_until_expiry: int | None = None source: str | None = None model_config = {"from_attributes": True} class GrantRequestBody(BaseModel): """One submission. ``role_key`` is the primary role; ``extra_role_keys`` are 0..N additional roles tacked on in the same form. Backend explodes them into N+1 ``CmdbCloudAccessGrant`` rows sharing one ``request_group_id`` so approve / reject / revoke act on the whole submission atomically — picking "Reader + Storage Blob Data Reader" is one approval decision, not two notifications. ``extra_role_keys`` capped at 10 because a single legitimate grant rarely needs more than 2-3, and a bigger list usually means the employee is misusing the form (a role admin should be defining a bundle role instead). """ scope_id: uuid.UUID role_key: str = Field(min_length=1, max_length=256) role_display_name: str | None = Field(default=None, max_length=256) extra_role_keys: list[str] = Field(default_factory=list, max_length=10) reason: str | None = Field(default=None, max_length=1000) ttl_days: int | None = Field(default=None, ge=1, le=365) class OwnerAddMemberBody(BaseModel): person_id: uuid.UUID role_key: str = Field(min_length=1, max_length=256) role_display_name: str | None = Field(default=None, max_length=256) ttl_days: int | None = Field(default=None, ge=1, le=365) class ApprovalBody(BaseModel): ttl_days: int | None = Field(default=None, ge=1, le=365) class RejectBody(BaseModel): reason: str | None = Field(default=None, max_length=500) class RevokeBody(BaseModel): reason: str = Field(default="admin_removed", max_length=64) # ── Serialization helpers ────────────────────────────────────── async def _serialize_scopes( db: AsyncSession, rows: list[CmdbCloudScope], *, viewer: AuthUser | None = None, ) -> list[dict[str, Any]]: """Serialize scope rows for API response. ``viewer``: when set, ``alert_rules.cc_person_ids`` is redacted for non-privileged viewers (anyone without ``can_read_cloud``). The cc list is the company's billing-incident escalation roster — often contains C-suite person_ids — and combined with ``pickable-people`` can be reverse-resolved to names + emails. Owners + admins still see the full list so they can edit it. """ if not rows: return [] redact_cc = viewer is not None and not can_read_cloud(viewer.role_names) # Global prod-write lock state, resolved once: every prod scope in # this batch carries ``write_locked`` so the frontend can disable # action buttons + show the explainer before a blocked call. See # ``app/domains/cmdb/prod_write_lock.py``. prod_locked = await prod_cloud_writes_locked(db) integ_ids = {r.integration_id for r in rows} owner_ids = {r.owner_person_id for r in rows if r.owner_person_id} decommissioner_ids = { r.decommissioned_by for r in rows if r.decommissioned_by } email_modifier_ids = { r.account_email_modified_by_person_id for r in rows if r.account_email_modified_by_person_id } person_ids = owner_ids | decommissioner_ids | email_modifier_ids integ_rows = (await db.execute( select(Integration).where(Integration.id.in_( [uuid.UUID(i) for i in integ_ids], )) )).scalars().all() integ_by_id = {str(i.id): i for i in integ_rows} person_rows = (await db.execute( select(IdentityPerson).where(IdentityPerson.id.in_( [uuid.UUID(i) for i in person_ids], )) )).scalars().all() if person_ids else [] person_by_id = {str(p.id): p for p in person_rows} # Backward-compatible alias: keeps the rest of the function readable. owner_by_id = person_by_id # Count grants AND distinct members in one grouped query per scope. # ``count(id)`` = total role assignments (grants); ``count(distinct # person_id)`` = unique people. The "成员" column needs the latter # (one row per person), the "生效 grant" hero stat the former. grant_counts: dict[str, dict[str, int]] = {} member_counts: dict[str, dict[str, int]] = {} if rows: c_rows = (await db.execute( select( CmdbCloudAccessGrant.scope_id, CmdbCloudAccessGrant.status, func.count(CmdbCloudAccessGrant.id), func.count(distinct(CmdbCloudAccessGrant.person_id)), ).where( CmdbCloudAccessGrant.scope_id.in_([str(r.id) for r in rows]), CmdbCloudAccessGrant.status.in_( (GRANT_STATUS_ACTIVE, GRANT_STATUS_PENDING), ), ).group_by(CmdbCloudAccessGrant.scope_id, CmdbCloudAccessGrant.status) )).all() for scope_id, status, grant_cnt, member_cnt in c_rows: grant_counts.setdefault(str(scope_id), {})[status] = int(grant_cnt) member_counts.setdefault(str(scope_id), {})[status] = int(member_cnt) # Resolve the console pivot URL once per integration (not once per # scope) — every scope on the same integration shares the same URL, # and resolution touches platform_settings each call. Falls back to # the hardcoded baseline when neither override is set. from app.domains.cmdb.console_urls import resolve_console_url console_url_by_integ: dict[str, str | None] = {} for integ in integ_rows: console_url_by_integ[str(integ.id)] = await resolve_console_url( db, integ, ) out: list[dict[str, Any]] = [] for r in rows: integ = integ_by_id.get(str(r.integration_id)) owner = owner_by_id.get(r.owner_person_id) if r.owner_person_id else None counts = grant_counts.get(str(r.id), {}) members = member_counts.get(str(r.id), {}) # Strip ``last_fired`` from alert rules for external consumers — # it's internal sweep bookkeeping, not part of the config. Also # redact ``cc_person_ids`` for non-privileged viewers (the cc # list IS the org's billing-escalation roster — contains C-suite # person_ids; combined with /scopes/{id}/pickable-people it # reverse-resolves to names + emails of the people you'd phish # to feign a billing-incident escalation). def _scrub(rule: dict) -> dict: scrubbed = {k: v for k, v in rule.items() if k != "last_fired"} if redact_cc: scrubbed.pop("cc_person_ids", None) return scrubbed alert_rules_ext = [_scrub(rule) for rule in (r.alert_rules or [])] _self_request_block = scope_self_request_block_reason(r) out.append({ "id": str(r.id), "integration_id": str(r.integration_id), "integration_name": integ.name if integ else None, "provider": integ.provider if integ else None, "scope_type": r.scope_type, "external_id": r.external_id, "display_name": r.display_name, "owner_person_id": r.owner_person_id, "owner_label": ( f"{owner.display_name} · {owner.work_email}" if owner else None ), "owner_employee_no": owner.employee_no if owner else None, "owner_employment_status": owner.employment_status if owner else None, "owner_department_chain": owner.department_chain if owner else [], "budget_usd_monthly": ( float(r.budget_usd_monthly) if r.budget_usd_monthly is not None else None ), "budget_measure": scope_budget_measure(r), "auto_suspend_after_days": r.auto_suspend_after_days, "default_grant_ttl_days": r.default_grant_ttl_days, "status": r.status, "environment": r.environment, # ``True`` when this is a production scope AND the global # prod-write lock is engaged — frontend disables every # mutating action and renders the lock banner. Backend # enforces independently (``ensure_prod_scope_writable``); # this flag is presentation only. "write_locked": prod_locked and r.environment in PROD_ENVIRONMENTS, # Can an employee self-request access on this scope at all? # Computed from the very guards ``submit_access_request`` # runs (see ``scope_self_request_block_reason``) so the # detail page's 「申请权限」 button is hidden / explained # instead of failing after the form is filled in. "self_requestable": _self_request_block is None, "self_request_block_reason": _self_request_block, "notes": r.notes, # Account email: cloud value + override + resolved effective # value + who last changed the override. ``account_email`` # is what the UI shows by default; the two source columns let # it render "云上值 X · 已改为 Y (张三)". "cloud_account_email": r.cloud_account_email, "account_email_override": r.account_email_override, "account_email": r.account_email_override or r.cloud_account_email, "account_email_modified_by_person_id": ( r.account_email_modified_by_person_id ), "account_email_modified_by_label": ( f"{person_by_id[r.account_email_modified_by_person_id].display_name}" f" · {person_by_id[r.account_email_modified_by_person_id].work_email}" if r.account_email_modified_by_person_id and r.account_email_modified_by_person_id in person_by_id else None ), "account_email_modified_at": r.account_email_modified_at, "alert_rules": alert_rules_ext, "active_grant_count": counts.get(GRANT_STATUS_ACTIVE, 0), "pending_grant_count": counts.get(GRANT_STATUS_PENDING, 0), "active_member_count": members.get(GRANT_STATUS_ACTIVE, 0), "pending_member_count": members.get(GRANT_STATUS_PENDING, 0), "last_synced_at": r.last_synced_at, "decommissioned_at": r.decommissioned_at, "decommissioned_by": r.decommissioned_by, "decommissioned_by_label": ( f"{person_by_id[r.decommissioned_by].display_name}" f" · {person_by_id[r.decommissioned_by].work_email}" if r.decommissioned_by and r.decommissioned_by in person_by_id else None ), "decommission_reason": r.decommission_reason, "decommission_provider_state": r.decommission_provider_state, # In-flight takedown (migration 624). Non-null while a # confirmed decommission is waiting on blocker cleanup, i.e. # while ``status`` is STILL ``active`` — the scope is a live # account with a takedown in progress, and the list/detail # pages render a 下线中 pill from these instead of looking # like nothing happened after the admin's click. "decommission_pending_since": r.decommission_pending_since, "decommission_cleanup_state": r.decommission_cleanup_state, "decommission_cleanup_error": r.decommission_cleanup_error, "console_url": console_url_by_integ.get(str(r.integration_id)), # Idle-sweep dedup + snooze state (migration 199). Both # nullable; null means "never notified" / "not snoozed". "last_idle_notified_at": r.last_idle_notified_at, "idle_snoozed_until": r.idle_snoozed_until, }) return out async def _serialize_grants( db: AsyncSession, rows: list[CmdbCloudAccessGrant], *, viewer: AuthUser | None = None, ) -> list[dict[str, Any]]: if not rows: return [] scope_ids = {r.scope_id for r in rows} # ``person_id`` may be NULL for orphan-cleanup history rows (the # principal was a leaver whose Helix person row no longer exists # — see migration 131). Filter NULLs out before the IdentityPerson # lookup so we don't try to ``UUID(None)``; the per-row serializer # below renders ``person_label`` from ``external_principal_email`` # in that case. person_ids = {r.person_id for r in rows if r.person_id} scope_rows = (await db.execute( select(CmdbCloudScope).where(CmdbCloudScope.id.in_( [uuid.UUID(s) for s in scope_ids], )) )).scalars().all() scope_by_id = {str(s.id): s for s in scope_rows} # Scope owners — used to tell a requester WHO is approving their # still-pending grant (the ``pending_approver`` field below). The # owner is the approval authority for an owned scope; an ownerless # scope routes to the cloud-admin team. Mirrors the routing in # ``CloudLifecycleService._resolve_request_approval_recipients``. owner_ids = {s.owner_person_id for s in scope_rows if s.owner_person_id} owner_rows = (await db.execute( select(IdentityPerson).where(IdentityPerson.id.in_( [uuid.UUID(o) for o in owner_ids], )) )).scalars().all() if owner_ids else [] owner_by_id = {str(p.id): p for p in owner_rows} integ_ids = {str(s.integration_id) for s in scope_rows} integ_rows = (await db.execute( select(Integration).where(Integration.id.in_( [uuid.UUID(i) for i in integ_ids], )) )).scalars().all() if integ_ids else [] integ_by_id = {str(i.id): i for i in integ_rows} # Renewal eligibility, computed server-side once per listing. # # The alternative — letting the client guess from status/expiry — is # how the cloud surface ended up full of buttons that were enabled and # then failed, and chips that claimed a state the backend disagreed # with (the 「永久授权」 chip on a grant the sweep was about to revoke # being the worst case). One authority, shipped to every renderer. # # Both prefetches are single queries for the whole page, so this adds # 2 round-trips regardless of row count. from app.domains.cmdb.lifecycle_service import load_perpetual_person_ids from app.domains.cmdb.models import ( RENEWAL_STATUSES_OPEN, CmdbCloudGrantRenewalRequest, ) from app.domains.cmdb.renewal_service import check_renewal_eligibility perpetual_ids = await load_perpetual_person_ids(db) open_renewal_grant_ids = { str(g) for g in (await db.execute( select(CmdbCloudGrantRenewalRequest.grant_id).where( CmdbCloudGrantRenewalRequest.grant_id.in_( [str(r.id) for r in rows], ), CmdbCloudGrantRenewalRequest.status.in_( RENEWAL_STATUSES_OPEN, ), ) )).scalars().all() } # ``IdentityPerson.user_roles`` is ``lazy="selectin"`` and # ``AuthUserRole.role`` is ``lazy="joined"`` (see person_models.py # and auth/models.py), so loading the persons here is enough — the # role_names property below will not trigger lazy-fetch in the # async session. We still compute role_names eagerly into a dict so # callers don't accidentally re-touch the ORM after session close. person_rows = (await db.execute( select(IdentityPerson).where(IdentityPerson.id.in_( [uuid.UUID(p) for p in person_ids], )) )).scalars().all() if person_ids else [] person_by_id = {str(p.id): p for p in person_rows} # ``person_role_names`` exposes which Helix platform roles each # member holds. The frontend uses this to render the perpetual # cloud-admin chip — but for non-privileged callers (a plain # employee listing the members of their own scope) it's an # information leak: knowing who the auditors / cloud admins / HR # admins are is a useful targeting primitive (phishing, social # engineering). So we only populate the field for: # * privileged readers — admins, auditors, cloud_admin (covered # by ``can_read_cloud``) # * the caller's own row (so the caller's own perpetual chip # still renders on /my/cloud pages, regardless of role) # Everyone else gets ``[]`` for OTHER members. The frontend's # perpetual-chip code path is no-op on ``[]`` so it falls back to # the standard expiry chip — which is the correct UX when we # cannot prove the member is a cloud_admin. privileged_viewer = ( viewer is not None and can_read_cloud(viewer.role_names) ) viewer_id = str(viewer.id) if viewer is not None else None person_role_names_by_id: dict[str, list[str]] = {} if privileged_viewer: person_role_names_by_id = { str(p.id): list(p.role_names) for p in person_rows } elif viewer_id is not None: # Non-privileged caller — only their own row carries roles. for p in person_rows: if str(p.id) == viewer_id: person_role_names_by_id[str(p.id)] = list(p.role_names) break # Global prod-write lock state, resolved once for the batch — every # grant on a prod scope carries ``write_locked`` so the frontend can # disable revoke / renew / release before a blocked call. prod_locked = await prod_cloud_writes_locked(db) out: list[dict[str, Any]] = [] now = _utc_now_naive() for r in rows: scope = scope_by_id.get(str(r.scope_id)) person = ( person_by_id.get(str(r.person_id)) if r.person_id else None ) integ = ( integ_by_id.get(str(scope.integration_id)) if scope else None ) days_until: int | None = None if r.expires_at: # ``r.expires_at`` is normally naive (MySQL/TiDB strips tz on # round-trip), but a grant that was JUST written in this same # session — approve/renew dispatch assigns # ``datetime.now(UTC) + timedelta`` — is still offset-AWARE # because ``expire_on_commit=False`` means ``db.get`` after # commit returns the in-memory object, not a fresh DB load. # Subtracting the naive ``now`` from it raised # "can't subtract offset-naive and offset-aware datetimes" # and 500'd the approve response *after* the grant had already # been persisted. Coerce to naive UTC so the arithmetic is # tz-agnostic regardless of which path produced the value. expires_at = r.expires_at if expires_at.tzinfo is not None: expires_at = expires_at.astimezone(UTC).replace(tzinfo=None) delta = (expires_at - now).days days_until = delta # Resolve the row's "ended_at / ended_reason" virtual fields. # Status-driven mapping so the frontend's "Ended" tab renders # uniformly regardless of which DB column actually carries the # data: # revoked → revoked_at / revoke_reason # disabled → detail.disable_result.at / .reason # expired → expires_at / "expired" # Anything else (active/pending) stays None. ended_at: datetime | None = None ended_reason: str | None = None if r.status == "revoked": ended_at = r.revoked_at ended_reason = r.revoke_reason elif r.status == "disabled": disable_blob = (r.detail or {}).get("disable_result") or {} iso_at = disable_blob.get("at") if isinstance(iso_at, str): try: ended_at = datetime.fromisoformat(iso_at) except ValueError: ended_at = None ended_reason = disable_blob.get("reason") elif r.status == "expired": ended_at = r.expires_at ended_reason = "expired" # Cloud-side revoke health (2026-06-10, execution-plane PR2). # Privileged readers only — the failure string can reference # IAM/internal identifiers, and the action button it powers is # ``can_manage_cloud``-gated anyway. Non-privileged co-members # see plain false/None (fields present so the FE type is # stable). r_detail = r.detail or {} stuck_cloud_revoke = bool( privileged_viewer and r_detail.get("requires_manual_cloud_revoke"), ) # ``pending_approver`` — only for still-pending grants. Tells the # requester (and anyone reading the request log / timeline) WHO # the request is waiting on, so a "Pending" row is never a # dead-end "waiting on … nobody-knows". The approval authority is # the scope owner; an ownerless scope (e.g. GCP billing account) # routes to the cloud-admin team. A departed / missing owner also # falls back to the team, matching the router's approve gate # (``can_manage_cloud`` OR owner) which an admin can still action. # ``kind`` lets the frontend i18n the team label while showing the # owner's name verbatim — no Chinese string baked into the API. pending_approver: dict[str, Any] | None = None if r.status == "pending": owner = ( owner_by_id.get(str(scope.owner_person_id)) if scope and scope.owner_person_id else None ) if ( owner is not None and owner.employment_status not in ("left", "leaver_pending") ): pending_approver = { "kind": "owner", "label": owner.display_name or owner.work_email, } else: pending_approver = {"kind": "cloud_admin_team", "label": None} out.append({ "pending_approver": pending_approver, "id": str(r.id), "scope_id": str(r.scope_id), "requires_manual_cloud_revoke": stuck_cloud_revoke, "manual_revoke_last_error": ( (r_detail.get("last_error") or None) if stuck_cloud_revoke else None ), "manual_revoke_escalated": bool( stuck_cloud_revoke and r_detail.get("manual_revoke_escalated_at"), ), "scope_display_name": scope.display_name if scope else None, "scope_external_id": scope.external_id if scope else None, "scope_environment": scope.environment if scope else None, # ``True`` when the grant's scope is production AND the # global prod-write lock is engaged — frontend disables # revoke / renew / release on this row. Backend enforces # independently at the mutating endpoints. "write_locked": ( prod_locked and scope is not None and scope.environment in PROD_ENVIRONMENTS ), "integration_id": str(scope.integration_id) if scope else None, "integration_name": integ.name if integ else None, "provider": integ.provider if integ else None, # ``person_id`` is None for orphan-cleanup history rows # (revoke_reason='orphan_principal_swept'). The frontend # renders ``person_label`` from # ``external_principal_email`` instead — see the table's # render fn on the 已结束 tab. "person_id": str(r.person_id) if r.person_id else None, "person_label": ( ( f"{person.display_name} · {person.work_email}" if ( privileged_viewer or ( viewer_id is not None and r.person_id is not None and str(r.person_id) == viewer_id ) ) # Non-privileged co-members of the scope see the # peer's display_name only — the full email # broadcast turned a scope-membership list into a # phishing-target enumeration primitive for every # employee with access to any cloud scope. else (person.display_name or "(team member)") ) if person # Orphan rows: privileged viewers see the recovered # email or principal_id; non-privileged see a generic # placeholder so they can't enumerate leavers. else ( ( r.external_principal_email or r.external_principal_id ) if privileged_viewer else "(orphan principal)" ) ), # Leaver visibility: always surface 在职状态 so PersonTag can # strike-through + 「已离职」-tag a departed grantee for every # viewer. 工号 follows the same privacy gate as the full label # (privileged / self only) — for a co-member peer we still mark # them departed but don't leak the internal employee number. "person_employment_status": ( person.employment_status if person else None ), "person_employee_no": ( person.employee_no if person and ( privileged_viewer or ( viewer_id is not None and r.person_id is not None and str(r.person_id) == viewer_id ) ) else None ), "person_role_names": person_role_names_by_id.get( str(r.person_id), [], ) if r.person_id else [], # Orphan-cleanup history rows carry the leaver's recovered # work email. That's PII (a now-departed employee's contact # info) — only privileged readers (admins, auditors, # cloud_admin) should see it. A scope owner / member must # NOT be able to enumerate "who used to have access via # this orphaned binding" by listing revoked grants on a # scope they share. ``person_label`` falls back to the # principal id only for privileged viewers; for everyone # else we redact to "(orphan principal)" so the row is # still visible (the count matters for completeness) but # the leaver's identifier isn't leaked. "external_principal_id": ( r.external_principal_id if privileged_viewer else None ), "external_principal_email": ( r.external_principal_email if privileged_viewer else None ), "role_key": r.role_key, "role_display_name": r.role_display_name, "status": r.status, "granted_at": r.granted_at, "expires_at": r.expires_at, "last_renewed_at": r.last_renewed_at, # ``last_login_at`` was emitted historically by the # retired ``cmdb.scope_inactivity_sweep`` cron. Migration # 138 retired the cron because the underlying signal # could never be populated reliably for SSO-class logins # (AWS Identity Center / Azure portal SSO are not in the # IAM lastUsed APIs). The column lingers in the DB so # historical values are not lost, but the field is no # longer surfaced to clients — every value would either # be NULL or stale, and clients showing "Last login: …" # tooltips would mislead the operator. The frontend # already treats the field as nullable so dropping it # produces graceful "—" placeholders without breaking # the rendering. "revoked_at": r.revoked_at, "revoke_reason": r.revoke_reason, "ended_at": ended_at, "ended_reason": ended_reason, # ``external_ref`` typically carries cloud-side principal # identifiers (AWS ARN with userId/account-id, Azure SSO # principal_id, RAM username). For a non-privileged scope # member listing co-members, exposing this is a recon # primitive — they can harvest every colleague's AWS userId # + account-id and use it against AWS-support / SCP / # cross-account assume attempts. Mirror the # ``person_role_names`` redaction: only privileged viewers # OR the row's own owner see the raw ref; everyone else # gets ``None``. "external_ref": ( r.external_ref if ( privileged_viewer or ( viewer_id is not None and r.person_id is not None and str(r.person_id) == viewer_id ) ) else None ), "days_until_expiry": days_until, "source": r.source, # Surface the requester-submitted reason + requested TTL so # the pending-approvals tab can render the approval card # without a second round-trip. Only meaningful for pending # rows (post-approve the row's ``detail`` is overwritten # with the provider response), but cheap to emit always — # the value is the same string the scope owner already # saw in the Feishu DM, no incremental sensitivity. "requested_ttl_days": (r.detail or {}).get("requested_ttl_days"), "request_reason": (r.detail or {}).get("reason") or None, # {"eligible": bool, "reason": str|None, "is_revival": bool}. # Drives whether 「申请续期」 renders enabled, and what the # tooltip says when it does not. "renewal_eligibility": ( await check_renewal_eligibility( db, grant=r, scope=scope, holder=person, perpetual_person_ids=perpetual_ids, open_request_grant_ids=open_renewal_grant_ids, ) ).as_dict if scope is not None else { "eligible": False, "reason": "scope_inactive", "is_revival": False, }, }) return out async def _require_scope_owner_or_admin( db: AsyncSession, user: AuthUser, scope_id: str, ) -> None: if can_manage_cloud(user.role_names): return if await is_cloud_scope_owner_of(db, str(user.id), scope_id): return raise ForbiddenError() async def _require_scope_reader( db: AsyncSession, user: AuthUser, scope_id: str, ) -> None: """READ-only superset of ``_require_scope_owner_or_admin``. Allows: * ``can_manage_cloud`` (global_admin) — full read/write * Scope owner — full read/write on their own scope * ``can_view_finance`` (finance / auditor / management) — read-only Used on endpoints powering the scope-detail page so a finance viewer who reached the page via /admin/cloud/overview row-click can actually load the data (scope meta, members, billing daily, owner-confirmation timeline). The page itself stays owner-aware: transfer / add-member / alerts / settings tabs only render for the owner (``isOwner`` check), so finance sees a read-only view. Mutating endpoints (PATCH scope, add grant, decommission, etc.) keep using ``_require_scope_owner_or_admin`` so finance's read permission does not silently elevate to write. """ if can_manage_cloud(user.role_names): return if can_view_finance(user.role_names): return if await is_cloud_scope_owner_of(db, str(user.id), scope_id): return raise ForbiddenError() # ── Scopes ──────────────────────────────────────────────── #: Natural direction per sort key, used when the caller does not pass #: ``sort_dir``. A single default would be wrong for one of the two: #: names read A→Z, timestamps read newest-first. Getting this wrong is #: silent — the list still renders, just backwards — so the default #: lives here next to the keys rather than in the signature. _SCOPE_SORT_DEFAULT_DIR = {"display_name": "asc", "decommissioned_at": "desc"} def _scope_order_by( sort: str | None, sort_dir: str | None, *, closed_only: bool, ) -> tuple[Any, ...]: """ORDER BY clause for ``GET /cmdb/scopes``. ``sort`` / ``sort_dir`` are already pattern-validated by FastAPI, so this maps known-good tokens rather than defending against injection. Two things it does that a bare ``ORDER BY col DIR`` would not: * **NULLs always last, in both directions.** ``decommissioned_at`` is NULL for archive rows closed before the column existed (prod carries ~1000 of them, e.g. the "Azure subscription 1" row whose when/who/why are all "—"). MySQL/TiDB sort NULL first ascending, which would bury every dated row the admin actually came for under a wall of blanks. ``col IS NULL`` as the leading key pins them to the end whichever way the arrow points. ``NULLS LAST`` is not MySQL syntax, so this is the portable spelling. * **A deterministic tiebreak.** Bulk decommissions land many rows on the same second; without a second key their relative order is undefined per query, which makes a paginated list silently drop and duplicate rows across pages. """ key = sort or ("decommissioned_at" if closed_only else "display_name") column = { "display_name": CmdbCloudScope.display_name, "decommissioned_at": CmdbCloudScope.decommissioned_at, }[key] direction = sort_dir or _SCOPE_SORT_DEFAULT_DIR[key] primary = column.desc() if direction == "desc" else column.asc() if key == "decommissioned_at": return (column.is_(None).asc(), primary, CmdbCloudScope.id.asc()) return (primary, CmdbCloudScope.id.asc()) @router.get("/scopes") async def list_scopes( request: Request, integration_id: str | None = None, scope_type: str | None = None, include_legacy_billing: bool = Query( default=False, description=( "Include ``legacy_billing_unit`` rows — cost-allocation units " "from decommissioned tenants. They are not cloud accounts and " "cannot be granted; hidden unless explicitly asked for." ), ), status: str | None = None, owner_person_id: str | None = None, unowned: bool = False, mine_as_owner: bool = False, requestable: bool = False, search: str | None = None, # Department node name (``DepartmentTreeSelect`` value). Narrows the # list to scopes whose OWNER's primary org membership sits in that # department's subtree — the same owner-attribution semantic the # Cloud Overview's global department filter applies to every other # surface on the page. Unowned scopes are unattributable and drop out. # ``max_length=128`` mirrors the cap on the /overview/admin + /analytics # department params. department: str | None = Query(None, max_length=128), # Repeatable query params (``?provider=aws&provider=azure``) so the # admin filter row can multi-select. ``Query(None)`` is the FastAPI # idiom for "optional list of strings". provider: list[str] | None = Query(default=None), environment: list[str] | None = Query(default=None), # Closed scopes — decommissioned and irreversible — are hidden from # the main list so admins don't have to filter past dead rows. # ``include_closed=true`` returns them (used by the "已下线" tab and # for back-compat with deep links). ``closed_only=true`` returns # ONLY closed rows (the "已下线" tab payload). The two flags are # mutually-exclusive shortcuts; passing both pins to closed_only. include_closed: bool = False, closed_only: bool = False, # Result ordering. Whitelisted (never a raw column name off the # wire) and deliberately tiny: the live list is browsed by name, # the "已下线" archive is browsed by WHEN — "which ones did I close # this morning" is unanswerable in a name-sorted list once the # archive has more than a page of rows. Omitting ``sort`` keeps the # historical default per tab: ``decommissioned_at`` for the archive, # ``display_name`` everywhere else, so no existing caller changes. # ``sort_dir`` defaults to None, NOT to a direction: the natural # direction differs per key (names A→Z, closures newest-first), and # a hardcoded default here would have silently flipped the live # list — which passes neither param — to name-DESCENDING. sort: str | None = Query( default=None, pattern="^(display_name|decommissioned_at)$", ), sort_dir: str | None = Query(default=None, pattern="^(asc|desc)$"), page: int = 1, # Cap matches ``/my-grants`` (le=500) because the employee # "Request cloud access" picker fetches every requestable scope in # one shot (``MyCloudGrantsPage`` → ``listScopes({page_size: 500, # requestable: true})``), and an org at 500-employee scale routinely # has 200+ Azure subscriptions / AWS accounts. The previous le=200 # silently 422'd the picker and rendered it empty. page_size: int = Query(50, ge=1, le=500), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """List cloud scopes. Default: admins see everything; employees are restricted to scopes they already own or hold a grant on (membership filter). ``requestable=true`` switches the caller into "I'm picking a scope to request access to" mode — returns every ``active`` scope that has an owner, regardless of the caller's current membership. This is what the employee "Request cloud access" modal needs; otherwise the picker is empty for anyone without existing grants. Only scope metadata (name, provider, owner) is exposed — no grants, no billing. """ if not can_view_cloud_resources(user.role_names): raise ForbiddenError() q = select(CmdbCloudScope).where(CmdbCloudScope.deleted_at.is_(None)) if integration_id: q = q.where(CmdbCloudScope.integration_id == integration_id) if scope_type: q = q.where(CmdbCloudScope.scope_type == scope_type) elif not include_legacy_billing: # Legacy billing units are cost-allocation rows imported from # decommissioned tenants, not cloud accounts — 37 of them were # rendering as live subscriptions in 云订阅 (see migration 466). # Hidden by default rather than deleted: the historical bills are # still meant to be imported against them. q = q.where( CmdbCloudScope.scope_type != SCOPE_TYPE_LEGACY_BILLING_UNIT, ) if status: q = q.where(CmdbCloudScope.status == status) elif closed_only: q = q.where(CmdbCloudScope.status == SCOPE_STATUS_CLOSED) elif not include_closed: # Default UX: the main list shows live scopes only. # Decommissioned scopes live behind the "已下线" tab, which # passes ``closed_only=true``. ``requestable=true`` already # restricts to ``status='active'`` further down so this guard # is a no-op for that path; redundant but explicit. q = q.where(CmdbCloudScope.status != SCOPE_STATUS_CLOSED) if owner_person_id: q = q.where(CmdbCloudScope.owner_person_id == owner_person_id) if unowned: q = q.where(CmdbCloudScope.owner_person_id.is_(None)) if mine_as_owner: q = q.where(CmdbCloudScope.owner_person_id == str(user.id)) if search: like = f"%{search}%" q = q.where(or_( CmdbCloudScope.display_name.ilike(like), CmdbCloudScope.external_id.ilike(like), )) if department: # Resolve the department to its owning person-id set (reusing the # admin-overview resolver so headcount / cost / scope filters all # agree on what a parent department contains) and narrow to scopes # those people own. Empty set (department with no members) returns # zero rows rather than silently dropping the filter — matching the # provider / environment "nothing matchable → nothing" rule above. from app.domains.cmdb.admin_overview_service import ( resolve_department_person_ids, ) dept_person_ids = await resolve_department_person_ids(db, department) if dept_person_ids: q = q.where(CmdbCloudScope.owner_person_id.in_(dept_person_ids)) else: q = q.where(sa.false()) if environment: # Whitelist defends against a malicious client poking unknown # values into the IN-clause. If the caller passes ONLY invalid # environment values, return zero rows (matching the provider # filter's "asked for nothing matchable → return nothing" # semantics) rather than silently dropping the filter — which # would surface dev/prod scopes the caller didn't ask for and # look like the chip is broken. valid_envs = [e for e in environment if e in {"dev", "staging", "prod"}] if valid_envs: q = q.where(CmdbCloudScope.environment.in_(valid_envs)) else: q = q.where(sa.false()) if provider: # Provider lives on ``integrations.provider``, not on the scope # row itself. Resolve to integration_ids first so the filter # composes with any other restriction. integ_ids = list((await db.execute( select(Integration.id).where( Integration.kind == "cloud", Integration.provider.in_(provider), ) )).scalars().all()) if integ_ids: q = q.where( CmdbCloudScope.integration_id.in_([str(i) for i in integ_ids]), ) else: # Caller asked for a provider with no matching integration # — return zero rows rather than silently dropping the # filter (which would surface scopes from OTHER providers # and look like the chip is broken). q = q.where(sa.false()) if requestable: # Always active + has-owner, regardless of caller's membership. # Also exclude scopes whose ``environment`` is in # ``PROD_ENVIRONMENTS`` — production cloud access must go # through the R&D security team (Kidd.li), not this self- # service flow. Filtering at the API level (rather than only # in the UI) prevents an employee from bypassing the picker # and POSTing ``/grants/request`` directly with a prod # scope_id; the service-layer guard # (``_refuse_prod_self_service``) is the second line of # defense. # # GCP billing-account scopes are intentionally OWNERLESS — a # billing account has no owner-equivalent IAM role to mint, so # the approval authority is the cloud-admin team, not a per-scope # owner (see ``submit_access_request`` + the ownerless-approval # notification fan-out). They must still surface in the picker — # this is the GCP-console billing-visibility 正路 — so admit them # even when ``owner_person_id IS NULL``. Approval still flows # through ``_require_scope_owner_or_admin`` which, with no owner, # only ``can_manage_cloud`` satisfies. # # Only ``TENANT_LEVEL_SCOPE_TYPES`` are grantable. Sub-tenant # rows (``azure_resource_group``, one per approved AI-API-key # request from the Azure OpenAI auto-grant flow) are active, # owned by the requester, and ``environment='dev'``, so they # satisfied every predicate above and surfaced in the employee # picker as ``helix-aoai-… · AI API · REQ-…`` — a dead-end # entry: its role dropdown is empty (roles_sync cannot # enumerate roles for an RG) and a grant on it could only fail # at dispatch with ARM ``400 InvalidSubscriptionId``. The # service-layer guard (``_refuse_ungrantable_scope_type``) is # the second line of defence, same pairing as the prod gate. q = q.where( CmdbCloudScope.status == "active", CmdbCloudScope.scope_type.in_(TENANT_LEVEL_SCOPE_TYPES), or_( CmdbCloudScope.owner_person_id.isnot(None), CmdbCloudScope.scope_type == SCOPE_TYPE_GCP_BILLING_ACCOUNT, ), CmdbCloudScope.environment.notin_(PROD_ENVIRONMENTS), ) elif not can_read_cloud(user.role_names) and not mine_as_owner: # Employees (no can_read_cloud) can only see scopes they own OR # are members of. member_scope_ids = list((await db.execute( select(CmdbCloudAccessGrant.scope_id).where( CmdbCloudAccessGrant.person_id == str(user.id), CmdbCloudAccessGrant.status.in_( (GRANT_STATUS_ACTIVE, GRANT_STATUS_PENDING), ), ) )).scalars().all()) q = q.where(or_( CmdbCloudScope.owner_person_id == str(user.id), CmdbCloudScope.id.in_([uuid.UUID(s) for s in member_scope_ids]) if member_scope_ids else False, )) count_q = select(func.count()).select_from(q.subquery()) total = (await db.execute(count_q)).scalar() or 0 q = (q.order_by(*_scope_order_by(sort, sort_dir, closed_only=closed_only)) .offset((page - 1) * page_size).limit(page_size)) rows = list((await db.execute(q)).scalars().all()) data = await _serialize_scopes(db, rows, viewer=user) return paginated_response( data=data, total=total, page=page, page_size=page_size, request=request, ) @router.get("/scopes/idle-candidates") async def list_scope_idle_candidates( request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), no_grants_for_days: int = 30, no_cost_for_days: int = 30, ): """Return scopes that look abandoned (0 active grants + $0 30d spend). Admin-only. Same logic as the weekly ``cmdb.scope_idle_sweep`` job but exposed for the on-demand admin UI banner / drill-down. Declared BEFORE ``/scopes/{scope_id}`` so the literal path wins over the parameterised one — FastAPI matches in declaration order. """ if not can_manage_cloud_scope_lifecycle(user.role_names): raise ForbiddenError() candidates = await list_idle_scope_candidates( db, no_grants_for_days=no_grants_for_days, no_cost_for_days=no_cost_for_days, ) return success_response( data=[ IdleScopeCandidateResponse( scope_id=c.scope_id, integration_id=c.integration_id, integration_name=c.integration_name, provider=c.provider, scope_type=c.scope_type, external_id=c.external_id, display_name=c.display_name, owner_person_id=c.owner_person_id, last_synced_at=c.last_synced_at, days_since_any_grant_activity=c.days_since_any_grant_activity, cost_30d_usd=c.cost_30d_usd, ).model_dump() for c in candidates ], request=request, ) @router.get("/scopes/{scope_id}") async def get_scope( scope_id: uuid.UUID, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): if not can_view_cloud_resources(user.role_names): raise ForbiddenError() scope = await db.get(CmdbCloudScope, scope_id) if scope is None or scope.deleted_at is not None: raise NotFoundError("CmdbCloudScope", str(scope_id)) # Privileged readers (can_read_cloud OR can_view_finance) see any # scope. Everyone else must be the owner or a (live-grant) member. # Finance promotion (2026-05-16): a finance viewer who reached # this page via /admin/cloud/overview row click needs to load # scope metadata + members + billing — same shape as auditor. if not ( can_read_cloud(user.role_names) or can_view_finance(user.role_names) ): is_owner = scope.owner_person_id == str(user.id) is_member = (await db.scalar( select(func.count(CmdbCloudAccessGrant.id)).where( CmdbCloudAccessGrant.scope_id == str(scope.id), CmdbCloudAccessGrant.person_id == str(user.id), CmdbCloudAccessGrant.status.in_( (GRANT_STATUS_ACTIVE, GRANT_STATUS_PENDING), ), ) )) or 0 if not is_owner and not is_member: raise ForbiddenError() [serialized] = await _serialize_scopes(db, [scope], viewer=user) return success_response(data=serialized, request=request) @router.patch("/scopes/{scope_id}") async def update_scope( scope_id: uuid.UUID, body: ScopeUpdateRequest, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): # Owner or admin. await _require_scope_owner_or_admin(db, user, str(scope_id)) # Prod cloud-account write lock (role-independent): editing config / # transferring owner on a production scope is blocked while the # global kill-switch is engaged. # # Exception 1: changing the ``environment`` label is an escape # hatch out of a *mis-marked* prod scope. It is a pure Helix-metadata # edit — it never reaches the cloud provider's IAM — and a prod scope # that is locked from every write could otherwise never be demoted to # staging/dev. So a request whose ONLY mutation is ``environment`` # bypasses the lock; the admin-only gate below still applies. # # Exception 2 (2026-06-12; widened 2026-08-04): owner transfer. Like # ``environment``, ``assign_scope_owner`` is pure Helix metadata # (DB rebind + audit + governance timeline — no provider IAM call; # the outgoing owner's grant revoke is separately prod-gated and # skipped on prod), and re-seating ownership is exactly the # governance action people need WHILE prod stays locked (e.g. handing # off a leaver's subscription). Per explicit product decision this # carve-out now covers the OWNER too, not just admins: a scope owner # handing prod ownership to a successor is a legitimate metadata # write, so owner transfer never trips the lock. # # Exception 3 (2026-08-04): the account-email override. Same rationale # as ``notes`` in every respect EXCEPT that it is deliberately NOT # lockable — it is contact metadata the owner must be able to correct # on any scope, prod included (product decision). ``cloud_account_email`` # itself stays provider-owned and is never written here. # # Everything else (budget / basis / status / ttl / auto-suspend / # notes) stays blocked on a locked prod scope exactly as before. mutates_lockable = any( v is not None for v in ( body.budget_usd_monthly, # A basis-only edit re-denominates every percentage rule's # threshold — a real alert-config mutation, not a label. body.budget_measure, body.auto_suspend_after_days, body.default_grant_ttl_days, body.notes, body.status, ) ) if mutates_lockable: await ensure_prod_scope_writable(db, await db.get(CmdbCloudScope, scope_id)) service = CloudLifecycleService(db) trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) if body.owner_person_id is not None: # Owner self-transfer is allowed: a scope owner is the right # party to hand off the account to a successor (peer-to-peer # transfer doesn't need IT as the middleman). The outer gate # (``_require_scope_owner_or_admin``) already proves the caller # is the current owner or a cloud-admin, so transfer can only # come from someone with legitimate management authority. # Service-layer ``assign_scope_owner`` continues to reject # leavers (offboarding invariant) and audits every rebind. # Privilege escalation is bounded — once owner transfers to B, # only B (or an admin) can transfer again, so A cannot reclaim. await service.assign_scope_owner( str(scope_id), str(body.owner_person_id), actor_user_id=user.id, trace_id=trace_id, revoke_previous_owner_grant=body.revoke_previous_owner_grant, actor_role_names=user.role_names, ) if body.environment is not None and not can_manage_cloud(user.role_names): # ``environment='prod'`` opts the scope OUT of the T+0 expiry # auto-revoke. A scope owner who could flip this themselves # would be able to retain access indefinitely past their TTL, # which defeats the point of the cleanup sweep. Restrict to # admins (matching the same posture as ``owner_person_id``). raise ForbiddenError() if body.status is not None and not can_manage_cloud(user.role_names): # ``status`` controls the active/closed lifecycle. Closing or # re-opening a scope is admin-only because ``closed`` is the # marker the expiry sweeper / billing summariser / owner-list # all key off, and the *real* close path goes through # ``/scopes/{id}/decommission`` which calls the provider's # CloseAccount API. Letting an owner flip this field directly # would skip the provider call (decommission becomes a lie) and # would also let them resurrect a closed scope — bypassing the # admin-only decommission gate above. Match the # ``environment`` / ``owner_person_id`` posture. raise ForbiddenError() scope = await service.update_scope_config( str(scope_id), budget_usd_monthly=body.budget_usd_monthly, budget_measure=body.budget_measure, auto_suspend_after_days=body.auto_suspend_after_days, default_grant_ttl_days=body.default_grant_ttl_days, notes=body.notes, status=body.status, environment=body.environment, account_email_override=body.account_email_override, actor_user_id=user.id, trace_id=trace_id, ) await db.commit() # Auto-provision the NEW owner's owner-equivalent grant so ownership and # membership stay in lock-step: a person set as owner must appear in the # member list with real cloud IAM, not just as a bare ``owner_person_id`` # column (the split-brain state synced/imported accounts landed in — # they never hit the create-scope path and the backfill cron excludes # them). Same inline, best-effort ``_post_provision_finalize`` the # create/approve paths use — it reads the just-committed owner in its # own session, resolves the per-provider owner role, and is idempotent # + prod/leaver-gated. Awaited (not a BackgroundTask) because prod # uvicorn silently drops those (2026-06-02 finding). if body.owner_person_id is not None: await _post_provision_finalize(str(scope_id), trace_id=trace_id) [serialized] = await _serialize_scopes(db, [scope], viewer=user) return success_response(data=serialized, request=request) @router.put("/scopes/{scope_id}/alerts") async def set_scope_alerts( scope_id: uuid.UUID, body: SetAlertRulesRequest, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Replace the alert-rule list for a scope. Scope owner or admin.""" await _require_scope_owner_or_admin(db, user, str(scope_id)) await ensure_prod_scope_writable(db, await db.get(CmdbCloudScope, scope_id)) service = CloudLifecycleService(db) trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) scope = await service.set_scope_alert_rules( str(scope_id), [r.model_dump() for r in body.rules], actor_user_id=user.id, # A brand-new absolute threshold is stamped with the basis its author # can actually READ (non-circle → list). Pass the request's role names # so ``X-View-Role`` narrowing counts: an admin authoring while # narrowed sees the list-basis surfaces the narrowed view serves. actor_role_names=list(user.role_names or []), trace_id=trace_id, ) await db.commit() [serialized] = await _serialize_scopes(db, [scope], viewer=user) return success_response(data=serialized, request=request) _SYNC_FAILURE_TEXT_KEYS = frozenset({ "error", "errors", "error_message", "reason", "scope_sync_error", }) _SYNC_FAILED_STATUSES = frozenset({"failed", "partial_failure"}) def _redact_sync_failure_fields( value: Any, *, field_name: str = "", failure_context: bool = False, ) -> Any: """Copy a sync summary while scrubbing diagnostic text recursively. Provider SDKs often put secrets below ``errors[].message`` or a nested ``error.detail`` object. Once a recognized failure field is entered, all descendant strings remain tainted even when their immediate key is a generic ``message``/``value`` name. """ failure_context = ( failure_context or field_name in _SYNC_FAILURE_TEXT_KEYS ) if isinstance(value, dict): return { key: _redact_sync_failure_fields( item, field_name=str(key).lower(), failure_context=failure_context, ) for key, item in value.items() } if isinstance(value, list): return [ _redact_sync_failure_fields( item, field_name=field_name, failure_context=failure_context, ) for item in value ] if isinstance(value, tuple): return tuple( _redact_sync_failure_fields( item, field_name=field_name, failure_context=failure_context, ) for item in value ) if isinstance(value, str) and failure_context: return redact_provider_error(value, max_len=400) return value def _sync_stage_failed(summary: dict[str, Any]) -> bool: return str(summary.get("status") or "").lower() in _SYNC_FAILED_STATUSES async def _run_scope_sync_background( *, integration_id: str, actor_user_id: uuid.UUID, scope_id: str, trace_id: str, ) -> None: """Run the three scope-sync sweeps in the background, holding a Redis advisory lock for the integration so an admin retry doesn't spawn a second concurrent sweep that races mark-and-sweep on the ``last_seen_sync_job_id`` column (would falsely close active scopes). """ from app.audit.service import AuditService as _AuditService from app.core.database import async_session_factory as _factory # ``LockBusy`` raises from ``async with``, not from the bare # ``distributed_lock(...)`` call (it's an @asynccontextmanager). try: async with distributed_lock( f"cloud_sync:{integration_id}", ttl=600, ): # Each stage consumes authority established by the previous one. # A failed scope snapshot means the DB only contains stale scope # membership; running role/grant mark-and-sweeps against it would # turn a fail-closed provider error back into downstream writes. # Convert raised guard failures to the same structured shape as # driver failures so every exit is audited rather than only logged. try: scopes_summary = await sync_scopes_for_integration(integration_id) except Exception as exc: # noqa: BLE001 scopes_summary = { "status": "failed", "error": redact_provider_error( format_exception_for_log(exc, max_len=400), max_len=400, ), } roles_summary: dict[str, Any] = { "status": "skipped", "reason": "scope sync failed", } grants_summary: dict[str, Any] = { "status": "skipped", "reason": "scope sync failed", } if not _sync_stage_failed(scopes_summary): try: roles_summary = await sync_roles_for_integration(integration_id) except Exception as exc: # noqa: BLE001 roles_summary = { "status": "failed", "error": redact_provider_error( format_exception_for_log(exc, max_len=400), max_len=400, ), } grants_summary = { "status": "skipped", "reason": "role sync failed", } if not _sync_stage_failed(roles_summary): try: grants_summary = await sync_grants_for_integration( integration_id, ) except Exception as exc: # noqa: BLE001 grants_summary = { "status": "failed", "error": redact_provider_error( format_exception_for_log(exc, max_len=400), max_len=400, ), } async with _factory() as session: audit = _AuditService(session) # Status reflects the INNER results (#2189-8). The cron # path already raises on partial failure; this admin # "立即同步" path buried the same failures inside a row # stamped ``success``, so the audit log said the sync # worked and the errors were only visible to whoever # expanded the detail JSON. summaries = { "scopes": scopes_summary, "roles": roles_summary, "grants": grants_summary, } audit_summaries = _redact_sync_failure_fields(summaries) failed_stages = [ stage for stage, summary in summaries.items() if _sync_stage_failed(summary) ] skipped_stages = [ stage for stage, summary in summaries.items() if summary.get("status") == "skipped" ] inner_errors = [ e for summary in audit_summaries.values() for e in (summary or {}).get("errors") or [] ] inner_errors.extend( str(audit_summaries[stage].get("error") or "stage failed") for stage in failed_stages ) audit_status = ( "failure" if "scopes" in failed_stages else "partial" if failed_stages or inner_errors else "skipped" if len(skipped_stages) == len(summaries) else "partial" if skipped_stages else "success" ) await audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="cloud.scope.sync_now", status=audit_status, resource_type="cmdb_cloud_scope", resource_id=scope_id, detail={ **audit_summaries, "error_count": len(inner_errors), "failed_stages": failed_stages, "skipped_stages": skipped_stages, }, ) await session.commit() except LockBusy: # Without an audit row here the admin sees a 202 "scheduled" # response, refreshes a few seconds later, and finds no new # data — *and* no trace in the audit log explaining why. Log # the skip distinctly so the scheduled-jobs history UI shows # "sync was busy; cron will catch up" rather than silently # appearing not to have run. (Mirrors the LockBusy branch in # ``run_grants_sync`` / ``run_scopes_sync`` / ``run_roles_sync`` # cron handlers, which return ``status=skipped`` for the same # reason.) logger.info( "skip background sync — already running for integration=%s", integration_id, ) try: async with _factory() as session: audit = _AuditService(session) await audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_user_id, action="cloud.scope.sync_now", status="skipped", resource_type="cmdb_cloud_scope", resource_id=scope_id, detail={ "reason": "another sync is already running", "integration_id": integration_id, }, ) await session.commit() except Exception: # noqa: BLE001 — best-effort audit logger.exception( "failed to audit-log LockBusy skip for integration=%s", integration_id, ) return except Exception: logger.exception( "background scope sync failed for integration=%s", integration_id, ) @router.post("/scopes/{scope_id}/sync-now", status_code=202) async def trigger_scope_sync( scope_id: uuid.UUID, request: Request, background: BackgroundTasks, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Force-refresh scope metadata, role catalog, and cloud-side grants. Admin-only. Returns 202 IMMEDIATELY and runs the three sweeps in the background under a per-integration Redis advisory lock so an admin retry (or cron + manual overlap) doesn't double-run and race the ``last_seen_sync_job_id`` mark-and-sweep — the previous synchronous behaviour would tie up the worker for 30-90s while holding no lock at all, with admin retries silently corrupting scope status. """ if not can_manage_cloud(user.role_names): raise ForbiddenError() scope = await db.get(CmdbCloudScope, scope_id) if scope is None or scope.deleted_at is not None: raise NotFoundError("CmdbCloudScope", str(scope_id)) # A force-sync reconciles cloud-side state and can revoke drifted # grants — a write against the prod account. Gate it too. await ensure_prod_scope_writable(db, scope) integration_id = str(scope.integration_id) trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) # Pre-flight: fail loud (409) if a sync is already in progress for # this integration so the admin sees feedback instead of a silent # "skipped" outcome from the background worker. from app.core.redis_client import get_redis as _get_redis redis = _get_redis() busy = await redis.get(f"helix:lock:cloud_sync:{integration_id}") if busy: raise AppException( code="SCOPE_SYNC_ALREADY_RUNNING", message=( "A sync is already running for this integration; " "wait for it to complete and retry." ), status_code=409, ) background.add_task( _run_scope_sync_background, integration_id=integration_id, actor_user_id=user.id, scope_id=str(scope_id), trace_id=trace_id, ) return success_response( data={"status": "scheduled", "integration_id": integration_id}, request=request, ) @router.get("/scopes/{scope_id}/last-sync-summary") async def get_scope_last_sync_summary( scope_id: uuid.UUID, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Return the grants summary from the most recent completed sync for this scope, specifically the per-scope ``skipped_unresolved`` count. Used by the scope detail page to explain why cloud-side principals that don't match a Helix identity are absent from the members list. Returns ``null`` when no sync has ever run. Two sources, newest wins: * the admin's manual **sync now** (``cloud.scope.sync_now`` audit row); * the nightly ``cmdb.grants_sync`` cron for this scope's integration. The cron source is not a nice-to-have. It is the ONLY source on almost every scope: manual sync-now is rare, so a page that read audit rows alone showed nothing at all — the prod Tencent scope had zero ``cloud.scope.sync_now`` rows in its lifetime, so the unresolved banner (the page's whole answer to "why are there fewer members here than in the cloud console?") had never once rendered. The job's health travels with the summary for the same reason: ``cmdb.grants.tencent-primary`` failed on every run for 10 days straight while this endpoint kept serving a cheerful ``null`` and the members table kept showing a frozen 2026-08-01 snapshot as if it were current. Silent staleness on an access-control surface reads as fact. """ if not can_read_cloud(user.role_names): raise ForbiddenError() scope = await db.get(CmdbCloudScope, scope_id) if scope is None or scope.deleted_at is not None: raise NotFoundError("CmdbCloudScope", str(scope_id)) from app.audit.models import AuditLog as _AuditLog from app.scheduled_jobs.models import ScheduledJob as _ScheduledJob row = (await db.execute( select(_AuditLog) .where( _AuditLog.resource_id == str(scope_id), _AuditLog.action == "cloud.scope.sync_now", _AuditLog.status == "success", ) .order_by(_AuditLog.created_at.desc()) .limit(1) )).scalars().first() # The grants cron for THIS integration. ``config`` is a small JSON # blob ({"integration_id": "..."}), matched in Python rather than with # a JSON path predicate: there is a handful of these rows, and a # portable JSON extract across MySQL/TiDB buys nothing here. job_rows = (await db.execute( select(_ScheduledJob).where( _ScheduledJob.handler_key == "cmdb.grants_sync", ) )).scalars().all() job = next( ( j for j in job_rows if str((j.config or {}).get("integration_id") or "") == str(scope.integration_id) ), None, ) if row is None and job is None: return success_response(data=None, request=request) def _naive(value: datetime) -> datetime: # Audit rows come back naive from MySQL/TiDB DATETIME while the # scheduler's columns are declared tz-aware; comparing the two # raw raises TypeError. return value.replace(tzinfo=None) if value.tzinfo else value job_summary: dict[str, Any] = {} job_last_run: datetime | None = None if job is not None: job_summary = job.last_result_summary or {} job_last_run = job.last_run_at # Newest successful run wins as the *content* source. A failed cron # contributes health only — its summary is empty or partial, and # overwriting a good manual summary with it would replace real numbers # with zeros. use_job = ( job is not None and job_last_run is not None and job.last_status == "ok" and (row is None or _naive(job_last_run) > _naive(row.created_at)) ) if use_job and job_last_run is not None: grants_detail = job_summary synced_at = job_last_run.isoformat() elif row is not None: grants_detail = (row.detail or {}).get("grants") or {} synced_at = row.created_at.isoformat() else: # A cron exists but has never completed successfully. Report the # health block with no content rather than pretending nothing runs. grants_detail = {} synced_at = None per_scope = grants_detail.get("skipped_per_scope") or {} scope_skipped = per_scope.get(str(scope_id), 0) return success_response( data={ "synced_at": synced_at, "skipped_unresolved": scope_skipped, # WHO could not be mapped, not just how many. These # principals hold live access cloud-side that Helix has no row # for: absent from every members table and from every sweep's # reach. The banner used to show only a count, so the one # actionable fact — which colleague to look up — was missing. # Tenant-wide (the sync records the sample per run, not per # scope), so the banner says "including" rather than implying # the list is exactly this scope's. "unresolved_principals": ( grants_detail.get("unresolved_principals") or [] ), "upserted": grants_detail.get("upserted", 0), "touched": grants_detail.get("touched", 0), "errors": grants_detail.get("errors") or [], "source": "scheduled" if use_job else "manual", # Cron health. ``job_status`` is None when no grants cron is # configured for this integration — which is itself worth # showing, since it means nothing refreshes this members list. "job_name": job.name if job is not None else None, "job_status": job.last_status if job is not None else None, "job_last_run_at": ( job_last_run.isoformat() if job_last_run is not None else None ), "job_last_error": ( (job.last_error or "")[:500] if job is not None else None ), "job_enabled": bool(job.enabled) if job is not None else None, }, request=request, ) @router.post( "/integrations/{integration_id}/grants-sync/force-drift-sweep", status_code=200, ) async def force_grants_drift_sweep_endpoint( integration_id: uuid.UUID, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), _gate: None = Depends(require_browser_session), ): """Admin-confirmed override of the 50% drift-sweep safety floor. Why this exists: the floor refuses to close stale ``discovered`` grants when ``matched / existing < 50%``, defending against driver pagination bugs silently mass-revoking legitimate access. When the admin has intentionally cleaned up many users in the cloud console (e.g. closed 28 of 37 IAM users), that delta is real, the floor blocks it forever, and only this explicit override can unstick grants_sync. Without an escape valve every subsequent sync stays locked at the same ratio. Behaviour: * Requires ``can_manage_cloud``. 403 otherwise. * Holds the same ``cloud_sync:`` distributed lock the cron uses, so this cannot race a regular sync. * Runs ``sync_grants_for_integration`` synchronously with ``force_drift_sweep=True`` and returns the summary including ``drift_closed_grant_ids`` so the UI can show exactly what it closed. * Writes a ``cloud.grant.drift_sweep_force_override`` audit row attributing the decision to the admin user. The inner sweep also writes its usual ``cloud.grant.drift_removed_sweep`` with ``force_override: True`` in the detail. """ if not can_manage_cloud(user.role_names): raise ForbiddenError() integ = await db.get(Integration, integration_id) if integ is None or integ.kind != "cloud": raise NotFoundError("Integration", str(integration_id)) trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) integ_id_str = str(integ.id) try: async with distributed_lock( f"cloud_sync:{integ_id_str}", ttl=600, ): summary = await sync_grants_for_integration( integ_id_str, force_drift_sweep=True, ) except LockBusy: raise AppException( code="SCOPE_SYNC_ALREADY_RUNNING", message=( "A sync is already running for this integration; " "wait for it to complete and retry." ), status_code=409, ) from None audit = AuditService(db) await audit.log( trace_id=trace_id, actor_type="user", actor_user_id=user.id, action="cloud.grant.drift_sweep_force_override", status="success", resource_type="integration", resource_id=integ_id_str, detail={ "integration_id": integ_id_str, "provider": integ.provider, "drift_closed": summary.get("drift_closed", 0), "drift_closed_grant_ids": summary.get( "drift_closed_grant_ids", [], ), }, ) await db.commit() return success_response(data=summary, request=request) @router.post( "/integrations/{integration_id}/roles-sync/force-sweep", status_code=200, ) async def force_roles_sweep_endpoint( integration_id: uuid.UUID, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), _gate: None = Depends(require_browser_session), ): """Admin-confirmed override of the roles_sync safety floor. Why this exists: ``sync_roles_for_integration`` refuses to soft- delete role rows when (a) every scope's ``list_scope_roles`` failed (errors-full, upserted=0) or (b) the upserted count drops below the configured tolerance (default 20%) of the active baseline. The floor defends against a partial driver outage silently emptying the request-form dropdown across the whole org. A *legitimate* one-shot mass drop (driver-side filter introduced, catalog format migrated, tenant truly shrunk) trips the same guard and the cron stays red forever — ack only silences the alarm without resolving the underlying state. This endpoint is the explicit escape valve: re-run roles_sync with ``force_sweep=True`` so the orphan rows are soft-deleted and the next cron returns to green. Concrete incident motivating this (2026-05-29): PR #565 added the ``AliyunReserved*`` filter to ``AliyunCloudDriver.list_scope_roles`` — the 10 674-row baseline (mostly RAM ``AliyunReservedSSO-*`` mirror policies that ``AttachPolicyToUser`` 400's on) legitimately collapsed to 288 real CloudSSO permission sets. The 20% drop guard correctly refused the sweep; without this endpoint the admin had no way to confirm-and- proceed and the ``cmdb.roles_sync`` cron was stuck red. Behaviour: * Requires ``can_manage_cloud``. 403 otherwise. * Holds the same ``cloud_sync:`` distributed lock the cron and other force-sweep endpoints use, so this cannot race a regular sync. * Runs ``sync_roles_for_integration`` synchronously with ``force_sweep=True`` and returns the summary including the ``forced_floor_bypass`` block (which guard tripped, prior baseline / upserted counts) so the UI can show what was bypassed. * Writes a ``cmdb.roles_sync.force_sweep_override`` audit row attributing the decision to the admin user at request time — the decision is made now regardless of how the sync ultimately completes. Soft-deletes themselves are reversible (``deleted_at=NULL`` on re-discovery); the audit trail is the forever record of WHO chose to bypass. """ if not can_manage_cloud(user.role_names): raise ForbiddenError() integ = await db.get(Integration, integration_id) if integ is None or integ.kind != "cloud": raise NotFoundError("Integration", str(integration_id)) trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) integ_id_str = str(integ.id) try: async with distributed_lock( f"cloud_sync:{integ_id_str}", ttl=600, ): summary = await sync_roles_for_integration( integ_id_str, force_sweep=True, ) except LockBusy: raise AppException( code="CLOUD_SYNC_ALREADY_RUNNING", message=( "A sync is already running for this integration; " "wait for it to complete and retry." ), status_code=409, ) from None audit = AuditService(db) await audit.log( trace_id=trace_id, actor_type="user", actor_user_id=user.id, action="cmdb.roles_sync.force_sweep_override", status="success", resource_type="integration", resource_id=integ_id_str, detail={ "integration_id": integ_id_str, "provider": integ.provider, "upserted": summary.get("upserted", 0), "deleted": summary.get("deleted", 0), "forced_floor_bypass": summary.get("forced_floor_bypass"), "errors": summary.get("errors", []), }, ) await db.commit() return success_response(data=summary, request=request) # ── Scope decommission (admin manual takedown) ───────────── @router.get("/scopes/{scope_id}/decommission-preflight") async def decommission_preflight_endpoint( scope_id: uuid.UUID, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """What would a decommission of this scope destroy? Read-only. Powers the 下线 dialog's "会一并删除以下内容" list. Why the dialog needs it: a cloud refuses to close a scope that still holds resources (Azure ``ResourcesExistingOnSubscription``, GCP ``liens_exist``), and until 2026-09-07 the admin met that fact as a 502 AFTER confirming an irreversible action. Worse, they met it having already decided: the production case that motivated this (``it-team-us``) was submitted with the reason 从未使用过 — true of its $0.00 spend, false of its contents, and no surface in Helix disagreed until Azure did. Response fields: ``supported`` — this provider/scope_type has a preflight at all. ``false`` means "we did not look"; the dialog must not claim the scope is empty. ``checked`` — the provider was actually reachable. ``false`` with ``supported=true`` means the same thing, one level narrower. ``cleanup_supported`` — Helix can delete what it found, so the dialog can offer 「一并清理并下线」 rather than instructions. ``blockers[]`` — ``key`` (what the confirm call echoes back), ``kind``, ``name``, ``detail``, ``children`` / ``child_count`` (what is INSIDE a container — a resource group's name alone is not enough to consent to deleting it), ``removable``, ``blocked_reason``. RBAC mirrors the decommission button itself (``can_manage_cloud_scope_lifecycle``) — this read describes an irreversible admin action, and its blocker list is a partial inventory of the scope's contents. """ if not can_manage_cloud_scope_lifecycle(user.role_names): raise ForbiddenError() service = CloudLifecycleService(db) scope, integration, preflight = await service.preflight_decommission_scope( str(scope_id), ) return success_response( data={ "scope_id": str(scope.id), "provider": integration.provider, "scope_type": scope.scope_type, "external_id": scope.external_id, "supported": preflight.supported, "checked": preflight.checked, "cleanup_supported": preflight.cleanup_supported, "error_code": preflight.error_code, "error_message": preflight.error_message, "blockers": [ { "key": b.key, "kind": b.kind, "name": b.name, "detail": b.detail, "children": b.children, "child_count": b.child_count, "removable": b.removable, "blocked_reason": b.blocked_reason, } for b in preflight.blockers ], # Mid-flight takedown, if any — the dialog and the detail # page both render "下线中·正在清理资源" from this. "pending": { "since": scope.decommission_pending_since, "state": scope.decommission_cleanup_state, "attempts": scope.decommission_cleanup_attempts, "error": scope.decommission_cleanup_error, } if scope.decommission_pending_since else None, }, request=request, ) @router.post("/scopes/{scope_id}/decommission", status_code=200) # IRREVERSIBLE + provider-side: a replay must never fire a second # decommission fan-out. @idempotent( "cloud_scope.decommission", key_params=("scope_id", "user.id"), takeover_minutes=None, ) async def decommission_scope_endpoint( scope_id: uuid.UUID, body: DecommissionScopeRequest, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), _gate: None = Depends(require_browser_session), ): """Admin closes a sub-account / subscription / project / RAM scope. The service calls the **provider's** close API first (AWS Organizations CloseAccount, Azure subscription cancel, GCP projects.delete, Aliyun ResourceDirectory UpdateAccount → CheckAccountDelete → DeleteAccount, or IAM-mode evacuation of every Helix-tracked IAM user). Only when the provider acknowledges the closure is Helix's row flipped to ``status='closed'`` and outstanding grants disabled. Returns: ``scope`` — refreshed scope row with decommission bookkeeping ``disabled_grants`` — Helix-side grants disabled in this call ``provider_state`` — terminal cloud-side state (``pending_closure`` / ``cancelled`` / ``delete_requested`` / ``deleted`` / ``deletion_silence_period`` / ``users_evacuated`` / ``not_in_org``) ``revoked_users`` — count of cloud-side IAM users actually deleted (only populated for IAM-mode close paths) ``already_closed`` — true when the provider reported the scope was already closed at the start of the call (idempotent) Failures bubble out as 502 (provider rejected) or 501 (driver does not support real closure for this scope_type) — Helix's row is left unchanged so the admin can retry after fixing the underlying cause. A 409 ``SCOPE_CLOSE_PRECHECK_RUNNING`` is not a failure: Aliyun's member-deletion pre-check takes 3–5 minutes and the close is resumable, so the admin simply clicks again shortly. The decommission action is irreversible: there is no reactivate. """ if not can_manage_cloud_scope_lifecycle(user.role_names): raise ForbiddenError() # Decommission is the highest-blast-radius write (irreversible # provider close). Gate it behind the prod-write lock too. await ensure_prod_scope_writable(db, await db.get(CmdbCloudScope, scope_id)) service = CloudLifecycleService(db) trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) scope, disabled_grants, close_result = await service.decommission_scope( str(scope_id), reason=body.reason, actor_user_id=user.id, trace_id=trace_id, cleanup_blockers=body.cleanup_blockers, confirmed_blocker_keys=body.confirmed_blocker_keys, ) await db.commit() [serialized] = await _serialize_scopes(db, [scope], viewer=user) # ``cleanup_in_progress`` is the one non-``ok`` close result that # reaches here instead of raising: the provider refused because the # scope still held confirmed blockers, Helix issued their deletion, # and the takedown is now in flight (see # ``CloudLifecycleService._begin_blocker_cleanup``). A 200 with # ``cleanup_in_progress=true`` is the honest answer — the admin's # action was accepted and nothing more is required of them — and the # pending-state writes must NOT be rolled back by an exception. return success_response( data={ "scope": serialized, "disabled_grants": disabled_grants, "provider_state": close_result.provider_state, "revoked_users": close_result.revoked_users, "already_closed": close_result.already_closed, "cleanup_in_progress": ( close_result.error_code == "cleanup_in_progress" ), "cleanup_message": ( close_result.error_message if close_result.error_code == "cleanup_in_progress" else None ), "cleanup_detail": ( close_result.detail if close_result.error_code == "cleanup_in_progress" else None ), }, request=request, ) # ── Scope idle-sweep snooze (owner self-service) ───────── @router.post("/scopes/{scope_id}/snooze-idle", status_code=200) async def snooze_idle_scope_endpoint( scope_id: uuid.UUID, body: SnoozeIdleScopeRequest, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), _gate: None = Depends(require_browser_session), ): """Suppress the weekly idle-sweep notification for this scope. Sets ``idle_snoozed_until = now + body.days`` (default 30). The weekly ``cmdb.scope_idle_sweep`` cron skips any scope still inside its snooze window OR within the 14-day automatic dedup window after a previous notification — see :func:`app.domains.cmdb.lifecycle_service.list_idle_scope_candidates`. RBAC: scope OWNER OR ``can_manage_cloud`` (cloud_admin / it_admin). Returns the refreshed scope row (same shape as GET /scopes/{id}) so the UI can re-render the "已忽略至 …" pill without a second fetch. Idempotent on re-clicks: a second snooze inside the existing window just pushes the deadline further out; audit row records both the previous and new ``snoozed_until`` so the timeline is honest. ``require_browser_session`` matches the CSRF posture of every other state-changing cmdb route (``decommission_scope_endpoint``, ``force_grants_drift_sweep_endpoint``, etc.) — defence-in-depth on top of the SameSite=Strict refresh-token cookie. MCP / programmatic clients reach the same service method via ``snooze_cloud_scope_idle_reminder``, which is gated by the MCP server's own per-tool auth. """ # No RBAC check at the router gate — the service-layer check # already covers "owner OR cloud-admin", which is the right shape # here. ``get_current_user`` ensures the request is authenticated. await ensure_prod_scope_writable(db, await db.get(CmdbCloudScope, scope_id)) service = CloudLifecycleService(db) trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) scope = await service.snooze_idle_scope( str(scope_id), days=body.days, actor_user_id=user.id, actor_role_names=list(user.role_names or []), trace_id=trace_id, ) await db.commit() [serialized] = await _serialize_scopes(db, [scope], viewer=user) return success_response(data=serialized, request=request) # ── Scope roles (request form support) ──────────────────── @router.get("/scopes/{scope_id}/roles") async def list_scope_roles( scope_id: uuid.UUID, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), promoted_only: bool = True, include_admin_roles: bool = False, ): """Roles available for granting on this scope. Defaults to ``promoted_only=True`` — i.e. only the admin-curated subset (Owner/Contributor/Reader-style short list) plus tenant custom roles. Without this filter the dropdown surfaced the full vendor catalog (Azure has ~380 built-ins) and made the "添加成员" form unusable. Callers that need the entire catalog (the role- catalog admin page lives at /admin/cloud/roles) can opt out with ``?promoted_only=false`` — the data is the same vendor metadata Azure / AWS / etc. publish in their public docs, so we don't gate the toggle behind an admin role. Governance-tier visibility (classified live from the driver — see ``app/integrations/cloud_role_governance.py``): ``ungrantable`` roles (can't be bound at this scope) are stripped for everyone; ``restricted`` roles (owner-equivalent / full-IAM-admin) are stripped for non-cloud-admin callers — regular employees must not self-service into Owner / "User Access Administrator" / "*Admin" power roles. The cloud-admin UI passes ``include_admin_roles=true`` to keep the full catalogue when managing other admins. ``elevated`` roles stay visible (the request modal badges them "needs admin approval"). Pairs with the same gate in :class:`CloudLifecycleService.submit_access_request` / ``owner_add_member`` so the rule holds even if a caller bypasses the dropdown. Each row also carries ``self_grant_status`` — ``'active'`` / ``'pending'`` when the CALLER already holds or has already applied for that role on this scope, else ``None``. The request form greys those rows out; ``submit_access_request`` 409s on them anyway. """ if not can_view_cloud_resources(user.role_names): raise ForbiddenError() scope = await db.get(CmdbCloudScope, scope_id) if scope is None or scope.deleted_at is not None: raise NotFoundError("CmdbCloudScope", str(scope_id)) where_clauses = [ CmdbCloudRole.integration_id == str(scope.integration_id), or_( CmdbCloudRole.scope_id == str(scope.id), CmdbCloudRole.scope_id.is_(None), ), CmdbCloudRole.deleted_at.is_(None), ] if promoted_only: where_clauses.append(CmdbCloudRole.is_promoted.is_(True)) # Manual rank first (lower = earlier); ``0`` means "unranked" and # must sort AFTER any explicitly-ranked role — mirrors the admin # catalog's ``sort_order or 10**9`` key so the employee dropdown and # the catalog table agree on order. rank = case( (CmdbCloudRole.sort_order == 0, 2_000_000_000), else_=CmdbCloudRole.sort_order, ) rows = (await db.execute( select(CmdbCloudRole).where(*where_clauses) .order_by( rank.asc(), CmdbCloudRole.is_custom.asc(), CmdbCloudRole.display_name.asc(), ) )).scalars().all() # Restricted-role visibility: only honor ``include_admin_roles=true`` # for callers that already hold cloud-admin — employees can ask but # can't flip the switch. show_admin = include_admin_roles and can_manage_cloud(user.role_names) from app.integrations.cloud_role_governance import ( TIER_NORMAL, TIER_RESTRICTED, TIER_UNGRANTABLE, ) from app.integrations.registry import get_driver_class # Classify live from the driver class so visibility never depends on # whether a role sync has run (the persisted ``governance_tier`` is a # convenience for the admin catalog, not the source of truth here). # An admin override on the row wins over the live classification. integration = await db.get( Integration, uuid.UUID(str(scope.integration_id)), ) driver_cls = ( get_driver_class("cloud", integration.provider) if integration else None ) def _tier_of(r: CmdbCloudRole) -> str: # Mirror _classify_scope_role: live driver classification is the # base; an admin override is honored EXCEPT it can't loosen an # ungrantable (provider fact). Keeps the dropdown and the # request/approve gates in lock-step. live = ( driver_cls.classify_role( r.role_key, r.display_name, scope.scope_type, ).tier if driver_cls is not None else (r.governance_tier or TIER_NORMAL) ) if live == TIER_UNGRANTABLE: return TIER_UNGRANTABLE if r.governance_overridden_by_admin and r.governance_tier: return r.governance_tier return live # The caller's OWN live grants on this scope, keyed by role_key. # ``submit_access_request`` 409s on a re-request for a role the # caller already holds ``active`` / ``pending``, so the request form # must be able to grey those rows out instead of letting the # employee pick a role that can only fail. Caller-scoped by # definition — never exposes anyone else's grants through this # endpoint (which non-members can read). self_grant_status: dict[str, str] = { role_key: status for role_key, status in (await db.execute( select( CmdbCloudAccessGrant.role_key, CmdbCloudAccessGrant.status, ).where( CmdbCloudAccessGrant.scope_id == str(scope.id), CmdbCloudAccessGrant.person_id == str(user.id), CmdbCloudAccessGrant.status.in_( (GRANT_STATUS_ACTIVE, GRANT_STATUS_PENDING), ), ) )).all() } data = [] for r in rows: tier = _tier_of(r) if tier == TIER_UNGRANTABLE: continue # can't be bound at this scope — hide from everyone if tier == TIER_RESTRICTED and not show_admin: continue # owner-equivalent / full-IAM-admin — cloud-admin only data.append({ "id": str(r.id), "role_key": r.role_key, "display_name": r.display_name, # Effective description: admin override wins over synced text. "description": r.description_override or r.description, "is_custom": r.is_custom, "source": r.source, "is_promoted": r.is_promoted, "sort_order": r.sort_order, "governance_tier": tier, "governance_reason": r.governance_reason, "governance_overridden_by_admin": r.governance_overridden_by_admin, # ``'active'`` / ``'pending'`` when the CALLER already holds # (or has already applied for) this role on this scope; # ``None`` when the role is freely requestable by them. "self_grant_status": self_grant_status.get(r.role_key), }) return success_response(data=data, request=request) @router.get("/scopes/{scope_id}/pickable-people") async def list_pickable_people_for_scope( scope_id: uuid.UUID, request: Request, search: str | None = None, page_size: int = Query(50, ge=1, le=200), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Active-person picker for the "add member" form. The admin user-listing endpoint at ``GET /users`` is gated behind ``require_hr_admin`` (GLOBAL_ADMIN ∪ HR) and returns sensitive PII (mobile_phone, personal_email, role assignments). A scope owner who is a regular employee can't call it — but they DO need to pick a teammate to grant access to. This endpoint returns the minimum identifying fields needed to populate the picker (id, display_name, work_email, primary_department) and is gated on ``_require_scope_owner_or_admin`` so it inherits the same authorisation posture as the ``owner-add`` action it feeds. No phones, no personal emails, no role assignments — those would re-introduce the disclosure problem this endpoint is dodging. """ await _require_scope_owner_or_admin(db, user, str(scope_id)) if page_size < 1 or page_size > 500: page_size = 50 where = [ IdentityPerson.deleted_at.is_(None), # Exclude already-departed people from the picker — granting # cloud access to a leaver is what the offboarding flow is # specifically designed to prevent. IdentityPerson.employment_status.in_(["active", "pre_hire"]), ] if search and search.strip(): like = f"%{search.strip()}%" where.append(or_( IdentityPerson.display_name.ilike(like), IdentityPerson.work_email.ilike(like), )) rows = (await db.execute( select(IdentityPerson) .where(*where) .order_by(IdentityPerson.display_name.asc()) .limit(page_size) )).scalars().all() data = [ { "id": str(p.id), "display_name": p.display_name, "work_email": p.work_email, "department": p.department, } for p in rows ] return success_response(data=data, request=request) # ── Grants ──────────────────────────────────────────────── @router.get("/grants") async def list_grants( request: Request, scope_id: str | None = None, person_id: str | None = None, status: str | None = None, # Multi-status filter — repeatable query param so the admin "已结束" # tab on the scope detail page can pull # ``?status_in=revoked&status_in=expired&status_in=disabled`` in # one round-trip. Cheaper than 3 paged calls and keeps the table's # "ended_at desc" sort coherent across all three statuses. Single # ``status`` still works for legacy callers. status_in: list[str] | None = Query(default=None), # Optional revoke_reason filter (e.g. "orphan_principal_swept") # so the admin can isolate one cleanup batch in the history tab. revoke_reason: str | None = None, mine: bool = False, page: int = 1, # Cap matches ``/my-grants`` (le=500) because the employee scope # detail page (``MyCloudScopeDetailPage``) fetches every grant on # the scope in one shot to render the members table. With Azure RBAC # discovery a single subscription routinely accumulates 100+ rows # (active + revoked + expired across N members × M roles), and the # previous le=200 silently 422'd the page on every detail-page load # — surfacing only as a "加载失败" toast and an empty members tab. page_size: int = Query(50, ge=1, le=500), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """List grants. ``mine=true`` scopes to the caller (works for any authenticated user). Admin-wide listing requires can_read_cloud. Scope owners can list all grants on scopes they own. Regular employees can list grants on scopes they're a *member* of when a specific ``scope_id`` is supplied — that's how the employee scope detail page renders "who else has access". """ q = select(CmdbCloudAccessGrant) if mine: q = q.where(CmdbCloudAccessGrant.person_id == str(user.id)) else: # ``can_read_cloud`` (global_admin / auditor / cloud_admin / # management) — full read. Finance (``can_view_finance``) also # gets full read since the FinOps / cost-allocation flows # depend on seeing who's spending on each scope; this matches # the same can_view_finance bypass used by ``get_scope`` and # ``list_billing_daily`` for the detail-page data load. if can_read_cloud(user.role_names) or can_view_finance(user.role_names): pass elif scope_id is not None: # Employee viewing a specific scope: allow if they hold a # grant on that scope OR own it. This is the read path that # powers /my/cloud/scopes/:scopeId's member list. try: scope_uuid = uuid.UUID(scope_id) except ValueError: raise NotFoundError("CmdbCloudScope", scope_id) from None scope = await db.get(CmdbCloudScope, scope_uuid) if scope is None or scope.deleted_at is not None: raise NotFoundError("CmdbCloudScope", scope_id) is_owner = scope.owner_person_id == str(user.id) is_member = (await db.scalar( select(func.count(CmdbCloudAccessGrant.id)).where( CmdbCloudAccessGrant.scope_id == str(scope.id), CmdbCloudAccessGrant.person_id == str(user.id), CmdbCloudAccessGrant.status.in_( (GRANT_STATUS_ACTIVE, GRANT_STATUS_PENDING), ), ) )) or 0 if not is_owner and not is_member: raise ForbiddenError() else: # No scope_id, not admin: fall back to owner-only listing # (preserves prior behavior for the owner console). owned_ids = list((await db.execute( select(CmdbCloudScope.id).where( CmdbCloudScope.owner_person_id == str(user.id), CmdbCloudScope.deleted_at.is_(None), ) )).scalars().all()) if not owned_ids: return paginated_response( data=[], total=0, page=page, page_size=page_size, request=request, ) q = q.where(CmdbCloudAccessGrant.scope_id.in_( [str(s) for s in owned_ids], )) if scope_id: q = q.where(CmdbCloudAccessGrant.scope_id == scope_id) if person_id: q = q.where(CmdbCloudAccessGrant.person_id == person_id) if status: q = q.where(CmdbCloudAccessGrant.status == status) if status_in: # Skip empty / whitespace-only entries so a stray ``?status_in=`` # in the URL doesn't degenerate to "WHERE status IN ('')". cleaned = [s.strip() for s in status_in if s and s.strip()] if cleaned: q = q.where(CmdbCloudAccessGrant.status.in_(cleaned)) if revoke_reason: q = q.where(CmdbCloudAccessGrant.revoke_reason == revoke_reason) count_q = select(func.count()).select_from(q.subquery()) total = (await db.execute(count_q)).scalar() or 0 q = (q.order_by(CmdbCloudAccessGrant.created_at.desc()) .offset((page - 1) * page_size).limit(page_size)) rows = list((await db.execute(q)).scalars().all()) data = await _serialize_grants(db, rows, viewer=user) return paginated_response( data=data, total=total, page=page, page_size=page_size, request=request, ) @router.post("/grants/request", status_code=201) async def request_grant( body: GrantRequestBody, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Employee-initiated request → pending until scope owner approves. Role-governance gating (ungrantable refused, restricted = cloud-admin only) is enforced centrally in :meth:`CloudLifecycleService.submit_access_request` so HTTP and MCP share one chokepoint — see ``app/integrations/cloud_role_governance.py``. """ await ensure_prod_scope_writable( db, await db.get(CmdbCloudScope, body.scope_id), ) service = CloudLifecycleService(db) trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) grant = await service.submit_access_request( scope_id=str(body.scope_id), person_id=str(user.id), role_key=body.role_key, role_display_name=body.role_display_name, reason=body.reason, ttl_days=body.ttl_days, requester_id=user.id, trace_id=trace_id, extra_role_keys=body.extra_role_keys, actor_role_names=list(user.role_names), ) await db.commit() [data] = await _serialize_grants(db, [grant], viewer=user) return success_response(data=data, request=request) @router.post("/grants/owner-add", status_code=201) async def owner_add_grant( request: Request, body: OwnerAddMemberBody, scope_id: str = Query(...), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Scope owner directly adds a member (no approval). RBAC: caller must own ``scope_id`` or be ``global_admin``. Elevated / restricted / ungrantable roles are refused for non-cloud- admin callers (and ungrantable for everyone) centrally in :meth:`CloudLifecycleService.owner_add_member` — a regular-employee scope owner must not elevate a colleague through this no-approval bypass. See ``app/integrations/cloud_role_governance.py``. """ await _require_scope_owner_or_admin(db, user, scope_id) await ensure_prod_scope_writable( db, await db.get(CmdbCloudScope, uuid.UUID(scope_id)), ) service = CloudLifecycleService(db) trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) outcome = await service.owner_add_member( scope_id=scope_id, person_id=str(body.person_id), role_key=body.role_key, role_display_name=body.role_display_name, ttl_days=body.ttl_days, actor_user_id=user.id, trace_id=trace_id, actor_role_names=list(user.role_names), ) await db.commit() grant = await db.get(CmdbCloudAccessGrant, uuid.UUID(outcome.grant_id)) [data] = await _serialize_grants(db, [grant], viewer=user) data["warning"] = outcome.warning return success_response(data=data, request=request) @router.post("/grants/{grant_id}/approve") # Provider dispatch is NOT internally idempotent — a retried approve # fires a second IAM/SCIM write. Scoped per grant so one client key # cannot bleed across grants. @idempotent("cloud_grant.approve", key_params=("grant_id", "user.id")) async def approve_grant( grant_id: uuid.UUID, body: ApprovalBody, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): grant = await db.get(CmdbCloudAccessGrant, grant_id) if grant is None: raise NotFoundError("CmdbCloudAccessGrant", str(grant_id)) await _require_scope_owner_or_admin(db, user, str(grant.scope_id)) await ensure_prod_scope_writable( db, await db.get(CmdbCloudScope, uuid.UUID(grant.scope_id)), ) service = CloudLifecycleService(db) trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) outcome = await service.approve_grant( str(grant_id), approver_id=user.id, ttl_days=body.ttl_days, trace_id=trace_id, approver_role_names=list(user.role_names), ) await db.commit() grant = await db.get(CmdbCloudAccessGrant, grant_id) [data] = await _serialize_grants(db, [grant], viewer=user) data["warning"] = outcome.warning return success_response(data=data, request=request) @router.post("/grants/{grant_id}/reject") @idempotent("cloud_grant.reject", key_params=("grant_id", "user.id")) async def reject_grant( grant_id: uuid.UUID, body: RejectBody, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): grant = await db.get(CmdbCloudAccessGrant, grant_id) if grant is None: raise NotFoundError("CmdbCloudAccessGrant", str(grant_id)) await _require_scope_owner_or_admin(db, user, str(grant.scope_id)) await ensure_prod_scope_writable( db, await db.get(CmdbCloudScope, uuid.UUID(grant.scope_id)), ) service = CloudLifecycleService(db) trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) await service.reject_grant( str(grant_id), rejector_id=user.id, reason=body.reason, trace_id=trace_id, ) await db.commit() grant = await db.get(CmdbCloudAccessGrant, grant_id) [data] = await _serialize_grants(db, [grant], viewer=user) return success_response(data=data, request=request) @router.post("/grants/{grant_id}/renew") # Renew bumps expires_at by N days FROM NOW, so a replay would extend # the grant twice. @idempotent("cloud_grant.renew", key_params=("grant_id", "user.id")) async def renew_grant( grant_id: uuid.UUID, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Cloud-admin break-glass renewal — NOT the employee path. Employee renewal goes through ``POST /scopes/{id}/renewal-requests`` and needs the scope owner's approval (2026-07-29: renewal used to be pure self-service, so "90-day expiry recycling" was a control nobody ever exercised — 763 renewals in three months, every one executed by the holder alone). This route survives for the case the approval flow cannot serve: an on-call engineer locked out of a scope mid-incident, where waiting for an owner to wake up is the wrong trade. It is therefore restricted to ``can_manage_cloud`` and is not reachable from the employee UI. A holder hitting it gets 403 with a pointer to the request flow. """ grant = await db.get(CmdbCloudAccessGrant, grant_id) if grant is None: raise NotFoundError("CmdbCloudAccessGrant", str(grant_id)) if not can_manage_cloud(user.role_names): raise AppException( code="CLOUD_RENEWAL_NEEDS_APPROVAL", message=( "续期需要该订阅 Owner 审批,不能自助直接续期。" "请在订阅详情页提交「申请续期」。" ), status_code=403, ) await ensure_prod_scope_writable( db, await db.get(CmdbCloudScope, uuid.UUID(grant.scope_id)), ) service = CloudLifecycleService(db) trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) await service.renew_grant( str(grant_id), actor_user_id=user.id, trace_id=trace_id, actor_role_names=list(user.role_names), ) # Break-glass is a bypass of the four-eyes rule, so it gets its own # audit row on top of the service's ``cloud.grant.renew`` — an # auditor asking "which renewals skipped approval?" must be able to # answer with one query rather than by diffing renew rows against # approved requests. await AuditService(db).log( trace_id=trace_id, actor_type="user", actor_user_id=user.id, action="cloud.grant.renew_admin_override", status="success", resource_type="cmdb_cloud_access_grant", resource_id=str(grant_id), detail={ "holder_person_id": ( str(grant.person_id) if grant.person_id else None ), "scope_id": str(grant.scope_id), "bypassed": "owner_approval", }, ) await db.commit() grant = await db.get(CmdbCloudAccessGrant, grant_id) [data] = await _serialize_grants(db, [grant], viewer=user) return success_response(data=data, request=request) # ── Renewal requests (owner-approved) ────────────────────── class RenewalRequestSubmitBody(BaseModel): """Employee-submitted renewal covering a whole subscription. ``grant_ids`` omitted (the normal case) means "every renewable grant I hold on this scope", which is the unit the owner then decides on. The pre-change per-grant flow made a holder with 8 roles click 8 times and put them in the owner's queue 8 times. """ grant_ids: list[uuid.UUID] | None = None # Bounded here as well as in the service so a malformed body is # rejected before touching the DB. ``le`` is the hard cap — the scope # owner's approval is sufficient for anything at or under it. ttl_days: int = Field( default=GRANT_TTL_DEFAULT_DAYS, ge=1, le=GRANT_TTL_HARD_CAP_DAYS, ) reason: str = Field(min_length=1, max_length=1000) class RenewalDecisionBody(BaseModel): approve: bool note: str | None = Field(default=None, max_length=1000) @router.post("/scopes/{scope_id}/renewal-requests") @idempotent( "cloud_grant.renewal_request", key_params=("scope_id", "user.id"), ) async def submit_renewal_request( scope_id: uuid.UUID, body: RenewalRequestSubmitBody, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Ask the scope owner to extend (or revive) your access.""" from app.domains.cmdb.renewal_service import CloudRenewalService trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) service = CloudRenewalService(db) result = await service.submit_renewal_request( scope_id=str(scope_id), person_id=str(user.id), grant_ids=[str(g) for g in (body.grant_ids or [])] or None, ttl_days=body.ttl_days, reason=body.reason, requested_by_person_id=user.id, trace_id=trace_id, actor_role_names=list(user.role_names), ) return success_response(data=result, request=request) @router.get("/scopes/{scope_id}/renewal-requests") async def list_scope_renewal_requests( scope_id: uuid.UUID, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Open renewal submissions on this scope — the owner's queue. Read gate is the scope-READER superset (owner / cloud admin / finance read-only), matching the other scope-detail panels so a viewer who reached the page does not hit a 403 on one tab. """ from app.domains.cmdb.renewal_service import CloudRenewalService await _require_scope_reader(db, user, str(scope_id)) rows = await CloudRenewalService(db).list_open_for_scope( str(scope_id), # Same 工号 privacy rule as ``_serialize_grants``: a scope owner is # frequently a plain employee, and this read gate also admits # finance-class viewers. privileged_viewer=can_read_cloud(user.role_names), ) return success_response(data=rows, request=request) @router.post("/renewal-requests/{request_group_id}/decide") @idempotent( "cloud_grant.renewal_decide", key_params=("request_group_id", "user.id"), ) async def decide_renewal_request( request_group_id: uuid.UUID, body: RenewalDecisionBody, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Approve or reject a renewal submission. Authority is verified inside the service (scope owner or cloud admin, never the holder themselves) — the gate is not duplicated here because the MCP tool shares the same service call and the two must not drift. The decision is final at every allowed TTL; the cloud-admin second signature above 90 days was removed on 2026-08-26. """ from app.domains.cmdb.renewal_service import CloudRenewalService trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) result = await CloudRenewalService(db).decide_renewal_group( request_group_id=str(request_group_id), approve=body.approve, approver_id=user.id, approver_role_names=list(user.role_names), trace_id=trace_id, note=body.note, ) return success_response(data=result, request=request) @router.post("/renewal-requests/{request_group_id}/cancel") async def cancel_renewal_request( request_group_id: uuid.UUID, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Withdraw your own open renewal submission.""" from app.domains.cmdb.renewal_service import CloudRenewalService trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) result = await CloudRenewalService(db).cancel_renewal_group( request_group_id=str(request_group_id), actor_id=user.id, actor_role_names=list(user.role_names), trace_id=trace_id, ) return success_response(data=result, request=request) @router.get("/my-pending-renewal-approvals") async def list_my_pending_renewal_approvals( request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Renewal submissions awaiting MY decision (owner or cloud admin). Mirrors the mobile action-inbox provider so the to-do badge and this list can never disagree. Powers the mobile decision page, whose deep link carries only the submission id — the Feishu approver card lands on a phone, so without this the card would be a dead end. """ from app.domains.cmdb.renewal_service import CloudRenewalService rows = await CloudRenewalService(db).list_pending_for_approver( str(user.id), is_cloud_admin=can_manage_cloud(user.role_names), privileged_viewer=can_read_cloud(user.role_names), ) return success_response(data=rows, request=request) @router.get("/me/renewal-requests") async def list_my_renewal_requests( request: Request, limit: int = Query(default=100, ge=1, le=500), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """The caller's own renewal history, newest first.""" from app.domains.cmdb.renewal_service import CloudRenewalService rows = await CloudRenewalService(db).list_for_person( str(user.id), limit=limit, ) return success_response(data=rows, request=request) @router.post("/grants/renew-by-token") async def renew_grant_via_token( request: Request, token: str = Query(...), db: AsyncSession = Depends(get_db), ): """One-click no-login renewal REQUEST from the expiry reminder card. The signed action token carries the grant holder + the grant id it authorises. Mirrors the cert/domain ``/expiry/ack`` magic-link pattern: the token IS the credential, so the link works in any webview (Feishu mobile included) without SSO. What changed on 2026-07-29: this used to APPLY the renewal on click. It now files a renewal request for the scope owner to decide. The token still only authorises the holder to act on their own access — which is exactly what "ask for an extension" is — so the security properties are unchanged while the four-eyes rule is gained. Both token purposes are accepted. Cards minted before the change carry ``cloud_grant_renew`` and have TTLs up to 44 days, so a live card in somebody's Feishu must keep working; it now does the new thing, which is strictly weaker than what it promised. """ from app.core.security import ( consume_action_jti, decode_action_token, release_action_jti, ) payload = None for purpose in ("cloud_grant_renewal_request", "cloud_grant_renew"): try: payload = decode_action_token(token, expected_purpose=purpose) break except Exception: # noqa: BLE001,S112 — try the legacy purpose continue if payload is None: raise ForbiddenError("Invalid or expired renew link.") try: person_uuid = uuid.UUID(str(payload["sub"])) grant_uuid = uuid.UUID(str(payload.get("res") or "")) except (ValueError, KeyError, TypeError) as exc: raise NotFoundError("CmdbCloudAccessGrant", str(payload.get("res"))) from exc grant = await db.get(CmdbCloudAccessGrant, grant_uuid) if grant is None: raise NotFoundError("CmdbCloudAccessGrant", str(grant_uuid)) if str(grant.person_id) != str(person_uuid): raise ForbiddenError("This renew link belongs to a different grant holder.") person = await db.get(IdentityPerson, person_uuid) if person is None or person.employment_status != "active": raise ForbiddenError("Grant holder is no longer an active employee.") # Force-logout parity: an active-but-force-logged-out holder (e.g. a # credential-compromise lockout) must also be refused, matching every # other credential surface. The active check above is stricter on the # departed axis; this adds the force_logout_at binding the local check # lacked. (enforce_* re-runs a departed check too — a no-op here since the # holder is already known active.) try: await enforce_action_token_subject(db, payload) except ValueError as exc: raise ForbiddenError("Grant holder session was terminated.") from exc scope = await db.get(CmdbCloudScope, uuid.UUID(str(grant.scope_id))) # Production scopes are not managed by Helix's lifecycle sweep — they # are never auto-recycled, and the prod cloud-write lock is # default-LOCKED so a renew would always trip ``ensure_prod_scope_ # writable`` and surface as the scary "link invalid or expired" page. # A reminder card should no longer be sent for prod (see # ``sweep_expiring_grants``), but cards delivered BEFORE that fix are # still sitting in holders' Feishu. Short-circuit to a reassuring # "your prod access is retained, nothing to renew" result so an old # card is not a dead-end. The grant keeps its access regardless. if scope is not None and scope.environment in PROD_ENVIRONMENTS: return success_response( data={ "already_renewed": True, "prod_no_renew": True, "expires_at": grant.expires_at, "scope_display_name": ( scope.display_name or scope.external_id ), "role": grant.role_display_name or grant.role_key, }, request=request, ) # Single-use gate: the FIRST click renews; every later click of the # same magic link (double-tap, webview re-open, back-then-reclick) # sees the jti already consumed and returns an idempotent "already # renewed" response instead of re-pinning expiry to today+90 or # racing the per-grant lock into a 409 that surfaced as the scary # "link invalid or expired" page. The claim is taken AFTER the # holder/lifecycle checks (a leaver's replay still errors) and is # released if the renew itself fails, so transient failures stay # retryable. claimed = await consume_action_jti(payload) if not claimed: return success_response( data={ "already_renewed": True, "expires_at": grant.expires_at, "scope_display_name": ( scope.display_name or scope.external_id if scope else None ), "role": grant.role_display_name or grant.role_key, }, request=request, ) from app.domains.cmdb.renewal_service import CloudRenewalService trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) try: await ensure_prod_scope_writable(db, scope) # The token authorises this holder to act on this grant, so the # request is filed in their own name — ``person_id`` and # ``requested_by_person_id`` are both the token subject, and no # role is inferred from the token. # # ``grant_ids`` is pinned to the one grant the token names rather # than defaulting to "every renewable grant on the scope": a card # is minted per grant, and silently widening its blast radius to # roles the card never mentioned would be a scope escalation. result = await CloudRenewalService(db).submit_renewal_request( scope_id=str(grant.scope_id), person_id=str(person_uuid), grant_ids=[str(grant_uuid)], ttl_days=None, reason="通过到期提醒卡片一键申请续期", requested_by_person_id=person_uuid, trace_id=trace_id, actor_role_names=[], ) await AuditService(db).log( trace_id=trace_id, actor_type="user", actor_user_id=person_uuid, action="cloud.grant.renewal_requested_via_token", status="success", resource_type="cmdb_cloud_access_grant", resource_id=str(grant_uuid), detail={ "via": "action_token", "request_group_id": result.get("request_group_id"), }, ) await db.commit() except Exception: # Failed after the jti was claimed — un-claim so the user's # one-click link isn't permanently burned by a transient error. await release_action_jti(payload) raise grant = await db.get(CmdbCloudAccessGrant, grant_uuid) return success_response( data={ # Kept for wire compatibility with the card's result page, # but a click no longer renews on its own. "already_renewed": False, "renewal_requested": True, "request_group_id": result.get("request_group_id"), "expires_at": grant.expires_at, "scope_display_name": ( scope.display_name or scope.external_id if scope else None ), "role": grant.role_display_name or grant.role_key, }, request=request, ) @router.post("/grants/{grant_id}/revoke") @idempotent("cloud_grant.revoke", key_params=("grant_id", "user.id")) async def revoke_grant_endpoint( grant_id: uuid.UUID, body: RevokeBody, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Revoke a grant. Allowed by: (a) scope owner, (b) admin, (c) the grant holder (self-release).""" grant = await db.get(CmdbCloudAccessGrant, grant_id) if grant is None: raise NotFoundError("CmdbCloudAccessGrant", str(grant_id)) is_self = grant.person_id == str(user.id) if not is_self: await _require_scope_owner_or_admin(db, user, str(grant.scope_id)) # Prod cloud-account write lock — blocks ALL revoke paths on a prod # scope, INCLUDING a member's own self-release. The incident that # prompted this lock was exactly a self-release on a production # account, so ``is_self`` does NOT exempt the caller here. await ensure_prod_scope_writable( db, await db.get(CmdbCloudScope, uuid.UUID(grant.scope_id)), ) service = CloudLifecycleService(db) trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) # Distinguish self-release from admin removal in audit. reason = ( "user_released" if is_self else body.reason or "admin_removed" ) # ``suppress_driver_error=False`` (round 16 audit, 2026-05-09): if # the cloud provider genuinely rejects the revoke (5xx, IAM lock, # quota), surface a 502 to the user instead of silently flipping # the Helix-side status to REVOKED while the cloud-side grant is # still active. The previous suppress=True default produced # "ghost-revoke" rows where Helix's UI claimed the access was # gone but the user still had it on AWS / Azure — a real # compliance + offboarding-trust hazard. Idempotent already-absent # responses (drivers return ``already_absent=True``) are still # treated as success at the service layer, so this won't break # the "revoke twice" / sweeper-then-admin-click case. # Offboarding / system sweeps still pass suppress=True at their # own call-sites (``lifecycle_service.py`` expiry + offboarding # sweeps, ``hr_lifecycle/advance_service.py`` Stage 4 leaver — # grep ``suppress_driver_error=True`` to enumerate) where bulk # recovery matters more than per-row strictness. await service.revoke_grant( str(grant_id), reason=reason, actor_user_id=user.id, trace_id=trace_id, suppress_driver_error=False, actor_role_names=user.role_names, ) await db.commit() grant = await db.get(CmdbCloudAccessGrant, grant_id) [data] = await _serialize_grants(db, [grant], viewer=user) return success_response(data=data, request=request) @router.post("/grants/{grant_id}/retry-cloud-revoke") async def retry_cloud_revoke_endpoint( grant_id: uuid.UUID, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Reset + immediately retry a stuck cloud-side revoke (``detail.requires_manual_cloud_revoke=True`` rows on the 已结束 tab). The resume path after an escalation page: clears the retry-storm counter + escalation stamp (re-arming the hourly sweep), runs one driver attempt under the grant lock, returns ``{cleared, last_error}``. Gated to ``can_manage_cloud`` — this drives a cloud-side IAM mutation, the same authority bar as scope decommission prep work. Prod write-lock applies like every other human-initiated cloud write; the *automated* sweep is unaffected by the lock. """ if not can_manage_cloud(user.role_names): raise ForbiddenError("cloud admin role required") grant = await db.get(CmdbCloudAccessGrant, grant_id) if grant is None: raise NotFoundError("CmdbCloudAccessGrant", str(grant_id)) await ensure_prod_scope_writable( db, await db.get(CmdbCloudScope, uuid.UUID(grant.scope_id)), ) service = CloudLifecycleService(db) trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) result = await service.admin_retry_cloud_revoke( str(grant_id), actor_user_id=user.id, trace_id=trace_id, ) await db.commit() return success_response(data=result, request=request) # ── Employee self-service ──────────────────────────────── @router.get("/my-grants") async def list_my_grants( request: Request, status: str | None = None, page: int = 1, page_size: int = Query(200, ge=1, le=500), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """All grants held by the caller across all clouds. Default ``page_size=200`` because Azure RBAC discovery routinely seeds 50–100+ active grants per user (one per subscription × role) and the frontend page renders one paged list — anything below that silently truncates the user's view of their own access. """ q = select(CmdbCloudAccessGrant).where( CmdbCloudAccessGrant.person_id == str(user.id), ) if status: q = q.where(CmdbCloudAccessGrant.status == status) count_q = select(func.count()).select_from(q.subquery()) total = (await db.execute(count_q)).scalar() or 0 q = (q.order_by(CmdbCloudAccessGrant.created_at.desc()) .offset((page - 1) * page_size).limit(page_size)) rows = list((await db.execute(q)).scalars().all()) data = await _serialize_grants(db, rows, viewer=user) return paginated_response( data=data, total=total, page=page, page_size=page_size, request=request, ) @router.get("/my-owner-status") async def my_owner_status( request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Used by the frontend to decide whether to show the owner console.""" is_owner = await person_owns_any_cloud_scope(db, str(user.id)) return success_response( data={"owns_any_scope": is_owner}, request=request, ) @router.get("/my-pending-grant-approvals") async def list_my_pending_grant_approvals( request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Pending grant requests the caller is responsible for approving. Powers two surfaces: * "我的工作" / ``MyTodosPage`` cloud-approvals tab — gives every scope owner a single inbox across all scopes they own, instead of relying on per-scope Feishu DMs that only fire once at submission time. * Optional dashboard badge — total pending count for the bell / header chip. Visibility rules: * ``global_admin`` (``can_manage_cloud``) sees every pending grant across the org — same surface as the admin "all approvals" console, scoped here to grant approvals only. * Plain scope owners see pending grants ONLY on scopes they own. Self-requested grants (``person_id == user.id``) are filtered out — a scope owner cannot four-eyes-approve their own grant, so surfacing them in the inbox just creates a phantom 409. The ``owner-add-member`` path is the legitimate "owner gives self a role" surface; this endpoint stays a peer-review queue. * Everyone else gets an empty list — no 403 because the frontend polls this in the background to render the inbox badge for every authenticated user, and a 403 there flips the global error toast on every page load. Returned rows reuse ``_serialize_grants`` so each row carries the full scope label / role / requester block the inbox renders. """ q = ( select(CmdbCloudAccessGrant) .where(CmdbCloudAccessGrant.status == GRANT_STATUS_PENDING) .order_by(CmdbCloudAccessGrant.created_at.desc()) ) if can_manage_cloud(user.role_names): rows = list((await db.execute(q)).scalars().all()) else: # Restrict to scopes the caller owns. A LEFT JOIN here keeps # the query a single round-trip — no separate "fetch scope ids # the caller owns" hop. q = q.join( CmdbCloudScope, CmdbCloudScope.id == CmdbCloudAccessGrant.scope_id, ).where( CmdbCloudScope.owner_person_id == str(user.id), # Block self-approval surface here too (defence in depth — # service-layer also raises ConflictError on self-approve). CmdbCloudAccessGrant.person_id != str(user.id), ) rows = list((await db.execute(q)).scalars().all()) data = await _serialize_grants(db, rows, viewer=user) return success_response( data={"items": data, "total": len(data)}, request=request, ) async def _requestable_directory_rows( db: AsyncSession, *, exclude_scope_ids: set[str], ) -> list[dict[str, Any]]: """Directory rows for scopes the caller is not a member of. Active production scopes are discoverable here so employees can identify them as production, but they are *not* self-service requestable. The separate ``requestable`` flag is intentionally distinct from ``accessible``: both kinds of directory row hide billing, members, and detail access; only non-production rows offer the quick-request action. The grant-request picker and service-layer production guard remain the authorization boundary. Scope-type predicate mirrors the ``requestable=true`` branch of ``list_scopes`` — ``TENANT_LEVEL_SCOPE_TYPES`` only, minus ``standalone_iam`` (Tencent / UCloud / AWS-SG follow the IAM-user mental model and live on their own page, see the exclusion note in ``list_my_subscriptions``). Anything outside that set is not a unit of cloud access and ``_refuse_ungrantable_scope_type`` rejects a grant on it, so surfacing one here hands the employee a 快速申请 button that 500s/422s *after* they tap it: * ``legacy_billing_unit`` — cost-allocation rows from decommissioned tenants (migration 466). 28 active rows across 12 owners in prod, and the ADMIN 云子账号/订阅 list has hidden them by default since 466 — which is how the same owner filter returned 7 rows for a global_admin and 12 for an employee (5 legacy units + the closed one the directory's ``status='active'`` already drops). * ``azure_resource_group`` — one row minted per approved Azure OpenAI API-key request; its ``external_id`` is an RG name, not a subscription GUID, so a grant could only fail at dispatch with ARM ``400 InvalidSubscriptionId``. The picker learned this filter already; this directory never did. """ q = select(CmdbCloudScope).where( *employee_directory_scope_criteria(), or_( CmdbCloudScope.owner_person_id.isnot(None), CmdbCloudScope.scope_type == SCOPE_TYPE_GCP_BILLING_ACCOUNT, ), ) rows = [ r for r in (await db.execute(q)).scalars().all() if str(r.id) not in exclude_scope_ids ] if not rows: return [] integ_ids = {str(r.integration_id) for r in rows} integ_map: dict[str, Integration] = {} if integ_ids: integ_rows = (await db.execute( select(Integration).where( Integration.id.in_([uuid.UUID(i) for i in integ_ids]), ) )).scalars().all() integ_map = {str(i.id): i for i in integ_rows} owner_ids = {r.owner_person_id for r in rows if r.owner_person_id} person_map: dict[str, IdentityPerson] = {} if owner_ids: person_rows = (await db.execute( select(IdentityPerson).where( IdentityPerson.id.in_([uuid.UUID(p) for p in owner_ids]), ) )).scalars().all() person_map = {str(p.id): p for p in person_rows} out: list[dict[str, Any]] = [] for scope in rows: integ = integ_map.get(str(scope.integration_id)) owner = ( person_map.get(scope.owner_person_id) if scope.owner_person_id else None ) out.append({ "scope_id": str(scope.id), "integration_id": str(scope.integration_id), "integration_name": integ.name if integ else None, "provider": integ.provider if integ else None, "scope_type": scope.scope_type, "external_id": scope.external_id, "display_name": scope.display_name, "owner_person_id": scope.owner_person_id, "owner_label": ( f"{owner.display_name} · {owner.work_email}" if owner else None ), "owner_employee_no": owner.employee_no if owner else None, "owner_employment_status": owner.employment_status if owner else None, "owner_department_chain": owner.department_chain if owner else [], "budget_usd_monthly": None, "default_grant_ttl_days": scope.default_grant_ttl_days, "status": scope.status, "environment": scope.environment, "write_locked": scope.environment in PROD_ENVIRONMENTS, "last_synced_at": scope.last_synced_at, "my_grants": [], "earliest_expires_at": None, "earliest_days_until_expiry": None, "my_latest_login_at": None, "has_active": False, "has_pending": False, "members": [], "member_count": 0, "cost_30d_usd": 0.0, "cost_unconverted": False, "cost_month_usd": 0.0, # The caller has no grant on this scope — it's a directory # entry they can request, not a subscription they're in. "accessible": False, # Production entries are discoverable but must never gain a # self-service request path. "requestable": scope.environment not in PROD_ENVIRONMENTS, }) return out @router.get("/my-subscriptions") async def list_my_subscriptions( request: Request, month: str | None = Query( None, description=( "Optional billing month ``YYYY-MM``. When set, the ``cost_month_usd`` " "field on each row is the total spend for that month (clamped to today " "for the current month). When omitted, ``cost_month_usd`` is the " "current month-to-date — the most natural default for the landing view." ), pattern=r"^\d{4}-(0[1-9]|1[0-2])$", ), include_requestable: bool = Query( False, description=( "When true, append a lightweight DIRECTORY of every cloud " "scope the caller could request access to but isn't a member " "of (``accessible=false`` rows — owner only, no billing / no " "members). Powers the employee 我的云订阅 'see everything, " "quick-apply for what I don't have' view. Active production " "scopes are included as discoverable, non-requestable rows so " "employees can identify them without gaining an access path." ), ), measure: str | None = Query( default=None, description=( "成本口径:``list``(默认,业务折前成本/研发口径)或 " "``invoice``(实付,仅 finance / cloud_admin / global_admin)。" ), ), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Subscription-centric view of every cloud scope the caller is a member of. Powers ``/my/cloud/grants`` ("我的云订阅" — My cloud subscriptions): one row per (scope) with the caller's role(s) + expiry, OTHER members on the scope, and per-month cost. Built as a single endpoint so the page renders in one round-trip instead of N+1 fetches per scope. 口径路由（2026-07 保密红线补口，第三轮补）：本端点**没有任何角色 闸**（自限于调用者自己的 grant），受众因此宽到普通员工，而两个成本 字段此前都是裸实付列。同一个 scope-member 从 ``/cmdb/scopes/{id}/billing-detail``（月 × scope）和 ``/cmdb/billing/daily``（日 × scope）拿到的是 **list** —— 同人同月 同粒度两个口径一相除就是折扣率。两个字段一律按 ``resolve_measure`` 路由，默认 list，``invoice`` 仅保密圈（无权限 → 403）。 Cost field semantics: * ``cost_month_usd`` — total spend for the requested month (current MTD when ``month`` is omitted). Pulled from ``cmdb_cost_facts`` via ``fact_amount_expr(measure)`` so the number agrees with the FinOps dashboard and the per-scope billing-detail drawer to the cent — **same measure on both**. * ``cost_30d_usd`` — legacy trailing-30-day total from ``cmdb_cloud_billing_daily`` via ``daily_amount_expr(measure)``. Kept on the wire to avoid breaking MCP/script callers that aren't aware of the month rework; the web UI now reads ``cost_month_usd``. * ``cost_unconverted`` — true when at least one of the scope's trailing-30-day rows had no resolvable FX rate, so ``cost_30d_usd`` is incomplete. Without this flag the row is a fabricated "$0 / cheap" claim (the driver/ops contract forbids a missing rate becoming a 0). """ from app.domains.cmdb.cost_view import ( daily_amount_expr, fact_amount_expr, resolve_measure, ) from app.domains.cmdb.models import CmdbCloudBillingDaily try: resolved_measure = resolve_measure(measure, user.role_names) except ValueError as exc: raise AppException( code="VALIDATION_ERROR", message=str(exc), status_code=422, field="measure", ) from exc if resolved_measure is None: raise ForbiddenError( "measure=invoice (实付) 仅限 finance / cloud_admin / " "global_admin。" ) from app.domains.cmdb.scope_billing_detail import ( _month_bounds, parse_month_param, ) async def _finish( rows: list[dict[str, Any]], accessible_scope_ids: set[str], ): """Common tail: optionally append the requestable directory, sort the merged list, and wrap in the standard envelope. Called from the early-exit paths (caller has no grants) AND the full path so the directory view works for brand-new employees too.""" if include_requestable and can_view_cloud_resources(user.role_names): rows = rows + await _requestable_directory_rows( db, exclude_scope_ids=accessible_scope_ids, ) rows.sort(key=lambda r: ( r["provider"] or "", (r["display_name"] or r["external_id"]).lower(), )) m_year, m_num = parse_month_param(month) _, _, m_partial = _month_bounds(m_year, m_num) return success_response( data=rows, meta={ "month": f"{m_year:04d}-{m_num:02d}", "month_is_partial": m_partial, # 口径标识 —— 与 ``/billing/daily`` / # ``/scopes/{id}/billing-detail`` 同框同口径的凭据。 "measure": resolved_measure, }, request=request, ) my_grants = list((await db.execute( select(CmdbCloudAccessGrant).where( CmdbCloudAccessGrant.person_id == str(user.id), CmdbCloudAccessGrant.status.in_( (GRANT_STATUS_ACTIVE, GRANT_STATUS_PENDING), ), ) )).scalars().all()) if not my_grants: return await _finish([], set()) scope_ids = list({str(g.scope_id) for g in my_grants}) # "我的云订阅" is the SUBSCRIPTION view: AWS/Azure/GCP/AliCloud accounts # the caller has SSO role assignments on. ``standalone_iam`` scopes # (Tencent, UCloud, AWS-Singapore for Bedrock) follow a different # mental model — the user gets an IAM account / API key rather than # role-on-shared-tenant — so they live on the IAM-users / API-key # pages and are filtered out here. Without this exclusion, the user's # AWS-SG IAM grants surfaced as "subscriptions" alongside the real # SSO accounts and confused the page. scope_rows = list((await db.execute( select(CmdbCloudScope).where( CmdbCloudScope.id.in_([uuid.UUID(s) for s in scope_ids]), CmdbCloudScope.deleted_at.is_(None), CmdbCloudScope.scope_type != SCOPE_TYPE_STANDALONE_IAM, ) )).scalars().all()) scope_by_id = {str(s.id): s for s in scope_rows} # Tighten the downstream IN-clauses to scopes that actually survived # the standalone_iam filter — otherwise the OTHER-members / # billing roll-ups query rows we're about to throw away. scope_ids = list(scope_by_id.keys()) if not scope_ids: return await _finish([], set()) integ_ids = {str(s.integration_id) for s in scope_rows} integ_map: dict[str, Integration] = {} if integ_ids: integ_rows = (await db.execute( select(Integration).where( Integration.id.in_([uuid.UUID(i) for i in integ_ids]), ) )).scalars().all() integ_map = {str(i.id): i for i in integ_rows} all_grants = list((await db.execute( select(CmdbCloudAccessGrant).where( CmdbCloudAccessGrant.scope_id.in_(scope_ids), CmdbCloudAccessGrant.status.in_( (GRANT_STATUS_ACTIVE, GRANT_STATUS_PENDING), ), ) )).scalars().all()) person_ids = {g.person_id for g in all_grants} | { s.owner_person_id for s in scope_rows if s.owner_person_id } person_map: dict[str, IdentityPerson] = {} if person_ids: person_rows = (await db.execute( select(IdentityPerson).where( IdentityPerson.id.in_([uuid.UUID(p) for p in person_ids]), ) )).scalars().all() person_map = {str(p.id): p for p in person_rows} end_date = datetime.now(UTC).date() start_date = end_date - timedelta(days=30) # ``total_cost_usd`` is misnamed: CN-billed tenants (国内阿里云 / 部分 # 腾讯云) write the NATIVE CNY amount into it with ``currency='CNY'`` # (conversion is deferred to read time). A bare SUM of the column # therefore folds ~7×-inflated CNY-as-USD into the same total as # genuine USD, and disagrees with the month figure on the same row # (which reads cost_facts.amount_usd already in USD). Pull the # per-row (scope, amount, currency, integration) tuples and fold each # through CloudCostUsdConverter before summing in Python so this 30-day # figure reconciles to one USD number. The converter caches the FX # rate + integration config per integration, so this is one lookup # per integration — not an N+1 over rows. from app.core.fx_rate import CloudCostUsdConverter # noqa: PLC0415 cost_conv = CloudCostUsdConverter(db) billing_rows = (await db.execute( select( CmdbCloudBillingDaily.scope_id, daily_amount_expr(resolved_measure).label("amount"), CmdbCloudBillingDaily.currency, CmdbCloudBillingDaily.integration_id, ).where( CmdbCloudBillingDaily.scope_id.in_(scope_ids), CmdbCloudBillingDaily.date >= start_date, CmdbCloudBillingDaily.date <= end_date, ) )).all() cost_by_scope: dict[str, float] = {} # scope 里有任何一行折不出汇率 → 该行的 30 天数字不完整,打旗标。 unconverted_scope_ids: set[str] = set() for scope_id, raw_cost, currency, int_id in billing_rows: usd, converted = await cost_conv.to_usd_checked( raw_cost, currency, int_id, ) if not converted: unconverted_scope_ids.add(str(scope_id)) continue cost_by_scope[str(scope_id)] = cost_by_scope.get(str(scope_id), 0.0) + usd if cost_conv.unconverted_by_currency: # A missing/stale FX rate would make an un-converted CN-billed # (CNY) row contribute 0.0, so ``cost_30d_usd`` would silently # under-report as if those scopes were free. Those rows are # excluded from the sum and the scope carries # ``cost_unconverted: true`` so the number is never read as a # complete "$0 / cheap" claim. logger.warning( "list_my_subscriptions: %d billing row currenc(ies) had no FX " "rate; those rows were excluded from cost_30d_usd and the " "affected scopes flagged cost_unconverted: %s", len(cost_conv.unconverted_by_currency), cost_conv.unconverted_by_currency, ) # Month-anchored cost — pulled from ``cmdb_cost_facts`` (the # ground-truth fact table). Walking this in addition to the # 30-day legacy sum gives the UI a number that matches the # billing-detail drawer the user opens from the same row; before # this rework the two surfaces could disagree because the list # read billing_daily aggregates while the drawer reads facts. # # Tenant-level scopes filter by ``scope_id``; RG-typed scopes # filter by ``(integration_id, resource_group)`` since cost_facts # for an RG carry the parent subscription's scope_id, not the # RG-scope's own. Same routing decision lifecycle_router already # makes for ``/billing/daily``. month_year, month_num = parse_month_param(month) month_start, month_end, month_is_partial = _month_bounds( month_year, month_num, ) tenant_scope_ids = [ s for s in scope_ids if scope_by_id[s].scope_type not in _RG_SCOPE_TYPES ] cost_month_by_scope: dict[str, float] = {sid: 0.0 for sid in scope_ids} if tenant_scope_ids: fact_rows = (await db.execute( select( CmdbCostFact.scope_id, func.coalesce( func.sum(fact_amount_expr(resolved_measure)), 0, ).label("usd"), ).where( CmdbCostFact.scope_id.in_(tenant_scope_ids), CmdbCostFact.date >= month_start, CmdbCostFact.date <= month_end, ).group_by(CmdbCostFact.scope_id) )).all() for r in fact_rows: cost_month_by_scope[str(r.scope_id)] = float(r.usd or 0) # RG-typed scopes — one composite query keyed by (integration, rg) # for ALL RG scopes the caller is on, then map back to scope_id. # ``cost_facts.scope_id`` is the PARENT subscription's id for RG- # owned facts (not the RG scope's own), so we can't use the tenant # query branch above. Pre-fix this was an N+1 (one query per RG # scope); a power user with many Azure OpenAI AKs would pay N # round-trips on every page load. Single query with OR-of-AND # tuples keeps the cost flat. rg_pairs = [ (str(s.integration_id), s.external_id, sid) for sid in scope_ids for s in [scope_by_id[sid]] if s.scope_type in _RG_SCOPE_TYPES and s.external_id ] if rg_pairs: rg_filter = or_( *[ and_( CmdbCostFact.integration_id == int_id, CmdbCostFact.resource_group == rg_name, ) for int_id, rg_name, _sid in rg_pairs ] ) rg_rows = (await db.execute( select( CmdbCostFact.integration_id, CmdbCostFact.resource_group, func.coalesce( func.sum(fact_amount_expr(resolved_measure)), 0, ).label("usd"), ).where( rg_filter, CmdbCostFact.date >= month_start, CmdbCostFact.date <= month_end, ).group_by( CmdbCostFact.integration_id, CmdbCostFact.resource_group, ) )).all() # Map (integration_id, rg_name) → scope_id. rg_to_sid = {(int_id, rg): sid for int_id, rg, sid in rg_pairs} for r in rg_rows: sid = rg_to_sid.get((str(r.integration_id), r.resource_group)) if sid is not None: cost_month_by_scope[sid] = float(r.usd or 0) now = _utc_now_naive() # Prod cloud-account write lock — one read for the whole list so the # "我的云订阅" page can disable the per-scope actions (申请角色 / # 续期 / 释放) on production scopes and show the lock explainer. prod_locked = await prod_cloud_writes_locked(db) out: list[dict[str, Any]] = [] # Prefetched once for the whole page (not per grant): the perpetual # role set and which of the caller's grants already have an open # renewal request. Both feed ``check_renewal_eligibility`` below. from app.domains.cmdb.lifecycle_service import load_perpetual_person_ids from app.domains.cmdb.models import ( RENEWAL_STATUSES_OPEN, CmdbCloudGrantRenewalRequest, ) from app.domains.cmdb.renewal_service import check_renewal_eligibility sub_perpetual_ids = await load_perpetual_person_ids(db) sub_open_renewals = { str(g) for g in (await db.execute( select(CmdbCloudGrantRenewalRequest.grant_id).where( CmdbCloudGrantRenewalRequest.person_id == str(user.id), CmdbCloudGrantRenewalRequest.status.in_( RENEWAL_STATUSES_OPEN, ), ) )).scalars().all() } for sid in scope_ids: scope = scope_by_id.get(sid) if scope is None: continue integ = integ_map.get(str(scope.integration_id)) owner = ( person_map.get(scope.owner_person_id) if scope.owner_person_id else None ) my_grant_dicts: list[dict[str, Any]] = [] earliest_expires: datetime | None = None for g in my_grants: if str(g.scope_id) != sid: continue days_until: int | None = None if g.expires_at: # Coerce to naive UTC before any arithmetic / comparison # — same tz-safety guard as ``_serialize_grants`` (a # just-written grant carries an offset-aware expiry # under ``expire_on_commit=False``). Without this the # subtraction OR the ``< earliest_expires`` compare can # raise the naive/aware TypeError. exp = g.expires_at if exp.tzinfo is not None: exp = exp.astimezone(UTC).replace(tzinfo=None) days_until = (exp - now).days if earliest_expires is None or exp < earliest_expires: earliest_expires = exp # ``last_login_at`` was emitted historically by the # retired ``cmdb.scope_inactivity_sweep`` cron (see # migration 138). The signal could never be populated # reliably for SSO-class logins, so the field is no # longer surfaced — every value would be NULL or stale. my_grant_dicts.append({ "id": str(g.id), "role_key": g.role_key, "role_display_name": g.role_display_name, "status": g.status, "granted_at": g.granted_at, "expires_at": g.expires_at, "days_until_expiry": days_until, # Server-decided renewal eligibility, so the # subscriptions list can offer 「申请续期」 without a # second fetch and can never enable it on a row the # backend would refuse. Same authority the scope-detail # serializer uses — one answer, every surface. "renewal_eligibility": ( await check_renewal_eligibility( db, grant=g, scope=scope, holder=person_map.get(str(g.person_id)), perpetual_person_ids=sub_perpetual_ids, open_request_grant_ids=sub_open_renewals, ) ).as_dict, }) member_buckets: dict[str, dict[str, Any]] = {} for g in all_grants: if str(g.scope_id) != sid: continue if g.person_id == str(user.id): continue mb = member_buckets.setdefault(g.person_id, { "person_id": g.person_id, "person_label": None, "person_employee_no": None, "person_employment_status": None, "person_role_names": [], "roles": [], "active_count": 0, "pending_count": 0, "earliest_expires_at": None, }) # Same disclosure gate as ``person_role_names`` below — the # cloud-side roles list (``Owner`` / ``Reader`` / # ``AdminAccess`` …) of OTHER members is the same family of # signal as their Helix role_names: privileged readers see # it, plain employees do not. Without this, a plain employee # who shares one scope with a colleague can enumerate # whether that colleague is the scope's Owner, just an # AdminAccess holder, etc. Same bug family as PRs 122→130. if can_read_cloud(user.role_names): mb["roles"].append(g.role_display_name or g.role_key) if g.status == GRANT_STATUS_ACTIVE: mb["active_count"] += 1 elif g.status == GRANT_STATUS_PENDING: mb["pending_count"] += 1 if g.expires_at: ee = mb["earliest_expires_at"] if ee is None or g.expires_at < ee: mb["earliest_expires_at"] = g.expires_at person = person_map.get(g.person_id) if person: if not mb["person_label"]: # Same disclosure gate as ``_serialize_grants`` # (Round-44 audit): non-privileged co-members must # not see other members' full ``work_email`` — the # entire member list of every shared scope was # otherwise a phishing-target enumeration primitive # for any employee with one grant. Privileged # readers (admin / auditor / cloud_admin / # finance / management) keep the full label. if can_read_cloud(user.role_names): mb["person_label"] = ( f"{person.display_name} · {person.work_email}" ) else: mb["person_label"] = ( person.display_name or "(team member)" ) # Leaver visibility: always mark a departed member; 工号 # follows the same privacy gate as the full label. mb["person_employment_status"] = person.employment_status if can_read_cloud(user.role_names): mb["person_employee_no"] = person.employee_no # Same disclosure gate as ``_serialize_grants``: a plain # employee viewing their own /my-subscriptions list # must not learn which other members are auditors / # cloud admins. Privileged readers see the full role # set; everyone else gets ``[]`` and the frontend falls # back to the standard expiry chip on those rows. # # NOTE: this lives at the ``if person:`` level (NOT # nested under ``not person_label``) so a future split # of the label-population step can't silently regress # role_names to ``[]`` and re-introduce the perpetual- # chip stale-grant bug fixed in PR 130. Same family of # silent-fallback bugs as PRs 122→127→128→129; keep # role_names assignment unconditional once we have a # person row. if can_read_cloud(user.role_names): mb["person_role_names"] = list(person.role_names) members = list(member_buckets.values()) members.sort(key=lambda m: ( -m["active_count"], m["person_label"] or "", )) days_until_earliest = ( (earliest_expires - now).days if earliest_expires else None ) out.append({ "scope_id": sid, "integration_id": str(scope.integration_id), "integration_name": integ.name if integ else None, "provider": integ.provider if integ else None, "scope_type": scope.scope_type, "external_id": scope.external_id, "display_name": scope.display_name, "owner_person_id": scope.owner_person_id, "owner_label": ( f"{owner.display_name} · {owner.work_email}" if owner else None ), "owner_employee_no": owner.employee_no if owner else None, "owner_employment_status": owner.employment_status if owner else None, "owner_department_chain": owner.department_chain if owner else [], "budget_usd_monthly": ( float(scope.budget_usd_monthly) if scope.budget_usd_monthly is not None else None ), "budget_measure": scope_budget_measure(scope), "default_grant_ttl_days": scope.default_grant_ttl_days, "status": scope.status, "environment": scope.environment, # True when this is a prod scope AND the global prod-write # lock is engaged — frontend disables 申请角色 / 续期 / 释放 # on the row and renders the lock banner. Backend enforces # independently at the mutating endpoints. "write_locked": ( prod_locked and scope.environment in PROD_ENVIRONMENTS ), "last_synced_at": scope.last_synced_at, "my_grants": my_grant_dicts, "earliest_expires_at": earliest_expires, "earliest_days_until_expiry": days_until_earliest, # ``my_latest_login_at`` was historically max(grants.last_login_at) # for the caller on this scope; retired alongside migration 138. # Always null going forward; frontends already treat it as # nullable so dropping the value doesn't break rendering. "my_latest_login_at": None, "has_active": any( g["status"] == GRANT_STATUS_ACTIVE for g in my_grant_dicts ), "has_pending": any( g["status"] == GRANT_STATUS_PENDING for g in my_grant_dicts ), "members": members, "member_count": len(members), "cost_30d_usd": cost_by_scope.get(sid, 0.0), # 汇率缺失时 ``cost_30d_usd`` 是不完整的(相关行已剔除)—— # 打旗标,不让 0 被读成「这个订阅不花钱」。 "cost_unconverted": sid in unconverted_scope_ids, "cost_month_usd": cost_month_by_scope.get(sid, 0.0), # The caller holds (or has pending) a grant on this scope — # full billing + member roster + detail click-through are # unlocked. Directory rows appended by ``_finish`` carry # ``accessible=False``. "accessible": True, }) return await _finish(out, set(scope_ids)) # ── Daily billing drill-down ───────────────────────────── @router.get("/billing/daily") async def list_billing_daily( request: Request, scope_id: str | None = None, integration_id: str | None = None, start: date | None = None, end: date | None = None, measure: str | None = Query( default=None, description=( "成本口径:``list``(默认,业务折前成本/研发口径)或 " "``invoice``(实付,仅 finance / cloud_admin / global_admin)。" ), ), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Daily billing rows for a scope (or integration). 口径路由（2026-07 保密红线补口）：本端点的受众是 **scope 成员，含 普通员工**（订阅列表的 14 天 sparkline / ``/my/cloud/grants``）， 而同一天粒度的 ``/billing/overview`` 对重叠受众已按研发口径出数。 继续裸返回 ``total_cost_usd``（实付）就是同粒度两个口径并存 → 日/周粒度折扣率可导出。默认改为 ``list``；``invoice`` 只有保密圈 （finance / cloud_admin / global_admin）能请求，无权限 → 403，绝不 静默降级。 Two paths depending on the requested scope: * **Tenant-level scope** (Azure subscription, AWS account, GCP project, Aliyun account) — reads ``cmdb_cloud_billing_daily``, which the per-provider daily cost cron writes. * **Resource-group scope** (Azure RG, ``parent_scope_id`` set, ``scope_type='azure_resource_group'``) — reads ``cmdb_cost_facts`` filtered to ``integration_id`` + ``resource_group=``, then aggregates per day. Rationale: providers don't return daily cost at RG granularity (Cost Management groups RG only on per-resource fact queries), so ``cmdb_cloud_billing_daily`` is structurally empty for RG scope rows. Pre-PR-251 the endpoint returned $0/empty for every RG scope detail page even when the underlying RG had real spend (verified 2026-05-11 prod: NexusCRM RG shown as $0 while cost_facts.resource_group='NexusCRM' had the actual usage). Members of the scope can read their scope's billing; admins / finance see all. The RG branch inherits the parent subscription's membership check — an RG owner can read their RG's billing even if they're not on the parent subscription's grants. """ if not can_view_cloud_resources(user.role_names): raise ForbiddenError() # Tenant-wide billing dump (no scope_id) leaks the full daily-cost # picture across every cloud account — competitive intel + layoff # forecasting + social-engineering goldmine. Limit unscoped reads # to admins / cloud_admin / auditor / finance. Scoped queries are # still gated by membership below. if not scope_id and not ( can_read_cloud(user.role_names) or can_view_finance(user.role_names) ): raise ForbiddenError( "Tenant-wide billing query requires admin / finance role; " "pass ?scope_id= for a per-scope view." ) # 口径闸:默认 list;invoice 仅保密圈,无权限 → 403(绝不静默降级)。 from app.domains.cmdb.cost_view import resolve_measure try: resolved_measure = resolve_measure(measure, user.role_names) except ValueError as exc: raise AppException( code="VALIDATION_ERROR", message=str(exc), status_code=422, field="measure", ) from exc if resolved_measure is None: raise ForbiddenError( "measure=invoice (实付) 仅限 finance / cloud_admin / " "global_admin。" ) end_dt = end or (datetime.now(UTC).date()) start_dt = start or (end_dt - timedelta(days=30)) # Hard-cap the span so a tenant-wide ``?start=1970-01-01&end=2099-12-31`` # can't materialize the entire (forever-growing) cmdb_cloud_billing_daily # table into memory and OOM the worker. Mirrors the sibling CSV export's # ``_resolve_daily_range`` guard (router.py) — that path already 422s on # oversized windows; this JSON path was the one surface that skipped it. _BILLING_DAILY_MAX_DAYS = 18 * 31 # 558 days, ~18 months if start_dt > end_dt: raise AppException( code="VALIDATION_ERROR", message="`start` must be on or before `end`.", status_code=422, ) if (end_dt - start_dt).days + 1 > _BILLING_DAILY_MAX_DAYS: raise AppException( code="VALIDATION_ERROR", message=( f"Range too large: {(end_dt - start_dt).days + 1} days " f"requested, max is {_BILLING_DAILY_MAX_DAYS} (~18 months). " "Narrow the window." ), status_code=422, ) # Resolve the scope row up front so the auth check, the RG branch # decision, and the cost_facts integration lookup all share the # same view. Admins still get the scope load (one DB hit) so the # RG-vs-tenant routing decision is uniform. scope: CmdbCloudScope | None = None if scope_id: try: scope_uuid = uuid.UUID(scope_id) except ValueError as exc: raise NotFoundError("CmdbCloudScope", scope_id) from exc scope = await db.get(CmdbCloudScope, scope_uuid) if scope is None: raise NotFoundError("CmdbCloudScope", scope_id) # Scope-member read permission for non-admins: the user must # be on this scope OR be its owner. For sub-tenant scopes # (Azure RG), parent-subscription ownership/membership also # grants read on the child — pre-PR-251 the parent and child # were the same row and one auth check served both; after the # split a parent-sub member who deep-links into an RG would # get 403 unless we inherit. Walks at most one hop (RG → sub); # tenant-level scopes have ``parent_scope_id IS NULL`` so the # inner branch never fires for them. # # The parent is included in the auth check ONLY when its row # is active (``status != 'closed'``). ``CmdbCloudScope`` uses # ``status='closed'`` as its tombstone — there is no # ``deleted_at`` column on this table. A decommissioned parent # must not grant residual access to a surviving child; # filtering ``status != closed`` covers both admin-driven # status flips and the sweeper that closes scopes on tenant # deprovisioning. Logic-review pass post-fix flagged the # missing tombstone filter as P1 (closed parent still # inheriting auth = privilege creep). # Privileged readers (admin / cloud_admin / auditor / # management / finance) bypass the membership check — finance # specifically needs scope-level billing to do per-scope cost # allocation. Symmetric with the get_scope + list_grants # bypasses above so the detail-page data load is consistent. if not ( can_read_cloud(user.role_names) or can_view_finance(user.role_names) ): scope_ids_for_perm = {str(scope.id)} owner_ids: set[str] = set() if scope.owner_person_id: owner_ids.add(scope.owner_person_id) if scope.parent_scope_id: try: parent_uuid = uuid.UUID(scope.parent_scope_id) except (ValueError, TypeError): parent_uuid = None if parent_uuid is not None: parent = await db.get(CmdbCloudScope, parent_uuid) parent_is_active = ( parent is not None and parent.status != SCOPE_STATUS_CLOSED ) if parent_is_active: scope_ids_for_perm.add(str(parent.id)) if parent.owner_person_id: owner_ids.add(parent.owner_person_id) is_owner = str(user.id) in owner_ids is_member = await db.scalar( select(func.count(CmdbCloudAccessGrant.id)).where( CmdbCloudAccessGrant.scope_id.in_(scope_ids_for_perm), CmdbCloudAccessGrant.person_id == str(user.id), CmdbCloudAccessGrant.status.in_( (GRANT_STATUS_ACTIVE, GRANT_STATUS_PENDING), ), ) ) or 0 if not is_owner and not is_member: raise ForbiddenError() # ── RG-scope branch: route through cost_facts.resource_group ── if scope is not None and scope.scope_type in _RG_SCOPE_TYPES: # RG branch pins the integration to the scope's own; allowing # the caller-supplied ``integration_id`` to override would let # a scope-member of RG ``A`` (on integration X) pass # ``?integration_id=Y`` and sum ``cost_facts`` keyed on # ``(integration=Y, resource_group=A.external_id)`` — leaking # cost from a tenant they have no grant on whenever RG names # collide across integrations ("default", "rg-prod", ...). The # auth gate above only validated membership on scope A, not on # integration Y. (P0 from code-review pass 1.) if integration_id and integration_id != str(scope.integration_id): raise AppException( code="RG_BILLING_INTEGRATION_MISMATCH", message=( "integration_id query param must match the RG " "scope's integration when both are provided " "(RG scopes are integration-pinned; cross-" "integration RG billing is not a supported view)." ), status_code=400, ) return await _list_billing_daily_for_rg_scope( db, scope=scope, start_dt=start_dt, end_dt=end_dt, request=request, measure=resolved_measure, ) # ── Tenant-level branch: existing billing_daily path ───────── from app.domains.cmdb.cost_view import daily_amount_expr q = select( CmdbCloudBillingDaily.date, CmdbCloudBillingDaily.scope_id, CmdbCloudBillingDaily.integration_id, daily_amount_expr(resolved_measure).label("total_cost_usd"), CmdbCloudBillingDaily.currency, CmdbCloudBillingDaily.cost_by_service, ).where( CmdbCloudBillingDaily.date >= start_dt, CmdbCloudBillingDaily.date <= end_dt, ) if scope_id: q = q.where(CmdbCloudBillingDaily.scope_id == scope_id) if integration_id: q = q.where(CmdbCloudBillingDaily.integration_id == integration_id) q = q.order_by(CmdbCloudBillingDaily.date.asc()) rows = list((await db.execute(q)).all()) # ``total_cost_usd`` 列名骗人:CN 云(国内阿里云 / 腾讯 / UCloud)按 # 契约把原生 CNY 写进该列并标 ``currency='CNY'``,写侧不 rebase。 # 本端点的消费方(``summarizeSparkline`` 的契约就是「USD total per # day」、``CostSparkline`` 的 $ 头条)按行裸加总,CN scope 的 14 天 # 花费因此 ~7× 虚高,且与同页 ``cost_30d_usd``(已折算,见 # list_my_subscriptions)和 ``/billing/overview`` 对不上。折算放在 # 服务端:同一个 ``CloudCostUsdConverter``(同解析顺序、同缓存), # 行的 ``currency`` 随之标 USD —— 数值与单位标签必须一致。 # 折不出汇率时**整行不发**并在 ``meta.unconverted_by_currency`` + # 日志披露:既不把未折算的 CNY 数字当 USD 发出去,也不发 $0 行 # (sparkline 的消费方不读 meta,$0 会被读成「这天没花钱」)。 from app.core.fx_rate import CloudCostUsdConverter # noqa: PLC0415 from app.domains.cmdb.cost_view import COST_MEASURE_INVOICE cost_conv = CloudCostUsdConverter(db) wants_invoice = resolved_measure == COST_MEASURE_INVOICE data = [] for r in rows: raw = float(r.total_cost_usd or 0) usd, converted = await cost_conv.to_usd_checked( r.total_cost_usd, r.currency, str(r.integration_id), ) if not converted: # 折不出汇率 → 这一行没有 USD 值,整行不发。此前它发 # ``total_cost_usd: 0.0 / currency: "USD"``,于是一天真实的 # ¥5000 花费在 sparkline 上读成「这天没花钱」,披露只在 # ``meta`` 里(而 sparkline 的消费方根本不读 meta)。 continue # ``cost_by_service`` 是日表的 JSON 列,**恒为实付口径拆分** # (日表没有对应的 list 列)。list 口径下不发:一个 list 合计 # 旁边挂实付服务拆分,(服务 × 日) 粒度的折扣率就能相除导出 # ——实测 total 1000.0 / {S3:100, EC2:300} → 0.4。本端点 # ``can_view_cloud_resources`` 受众含**普通员工**(scope 成员), # 是这条红线上受众最宽的面。 # # 不发而不是「从 cost_facts 重算一份 list 拆分」:grep 过全部 # 消费方——``CostSparkline``/``summarizeSparkline``、 # ``CloudScopesPage``、``MyCloudGrantsPage`` 只读 ``date`` + # ``total_cost_usd``,没有任何面渲染这个 map;而这两个列表页是 # **逐 scope 扇出**调用本端点(20 行 × 14 天),再加一次 # cost_facts GROUP BY 是纯成本。需要 list 口径的服务拆分时走 # ``/cmdb/scopes/{id}/billing-detail``(facts 直算,带 measure) # 或本端点的 RG 分支——所以没有死胡同面板。 scale = (usd / raw) if raw else 1.0 data.append({ "date": r.date.isoformat(), "scope_id": r.scope_id, "integration_id": r.integration_id, "total_cost_usd": usd, "currency": "USD", "cost_by_service": { k: float(v) * scale for k, v in (r.cost_by_service or {}).items() } if wants_invoice else {}, }) meta: dict[str, Any] = {"measure": resolved_measure} if cost_conv.unconverted_by_currency: logger.warning( "list_billing_daily: %d currenc(ies) had no FX rate; those rows " "were dropped from the response: %s", len(cost_conv.unconverted_by_currency), cost_conv.unconverted_by_currency, ) meta["unconverted_by_currency"] = cost_conv.unconverted_by_currency return success_response(data=data, request=request, meta=meta) async def _list_billing_daily_for_rg_scope( db: AsyncSession, *, scope: CmdbCloudScope, start_dt: date, end_dt: date, request: Request, measure: str, ): """Aggregate ``cmdb_cost_facts`` per day for an RG-typed scope. Pre-condition: ``scope.scope_type in _RG_SCOPE_TYPES`` AND the caller has already enforced read permission AND ``measure``已过 :func:`resolve_measure` 的保密圈闸（本函数不再重复判权）。 The integration is pinned to ``scope.integration_id`` — the caller-supplied ``?integration_id=`` query param is rejected by the outer handler when it disagrees (see the ``RG_BILLING_INTEGRATION_MISMATCH`` raise in ``list_billing_daily``). RG names collide across cloud integrations (``default``, ``rg-prod``, …); accepting a caller-supplied override here would let any scope-member of RG A on integration X read RG-A-named facts on integration Y instead. SQL contract: groups by date, sums the measure's amount column via ``fact_amount_expr`` (in Decimal — fact rows are Numeric(14, 6) and dollar-precise reconciliation against vendor invoices runs to 4 dp; a Python float accumulator would drift on 30-day sums), and collapses per-service breakdown into the same ``cost_by_service`` dict shape ``cmdb_cloud_billing_daily`` exposes (so frontends consume a uniform payload regardless of branch). 口径:``fact_amount_expr`` —— list 是 ``COALESCE(amount_usd_list, amount_usd)``,invoice 是 ``amount_usd``。两列都已经是 USD,所以这 条分支不需要（也没有）汇率折算,与 tenant 分支的 CNY rebase 不同。 When ``scope.external_id`` is empty (defensive — should never happen post-migration-141 since the autogrant always sets it on RG-typed rows), returns an empty list rather than over-fetching rows with NULL resource_group from another scope. """ from app.domains.cmdb.cost_view import fact_amount_expr integration_id = str(scope.integration_id) rg_name = scope.external_id if not rg_name: return success_response( data=[], request=request, meta={"measure": measure}, ) # Per-day aggregate (total + service map) in one query. The # service breakdown is built by grouping (date, service) then # folding rows into the per-day dict on the Python side — TiDB # has no native JSON_OBJECTAGG that round-trips cleanly across # the asyncmy driver, and the row count per day per RG is small # (typically <30 service rows / day) so the in-Python fold is # cheap. breakdown_stmt = ( select( CmdbCostFact.date, CmdbCostFact.service, func.coalesce( func.sum(fact_amount_expr(measure)), 0, ).label("usd"), ) .where( CmdbCostFact.integration_id == integration_id, CmdbCostFact.resource_group == rg_name, CmdbCostFact.date >= start_dt, CmdbCostFact.date <= end_dt, ) .group_by(CmdbCostFact.date, CmdbCostFact.service) .order_by(CmdbCostFact.date.asc()) ) # Accumulate in Decimal — float sums drift across many service-rows # per day and the column is Decimal-typed on the source side. per_day: dict[date, dict[str, Any]] = {} for row in (await db.execute(breakdown_stmt)).all(): day = row.date bucket = per_day.setdefault( day, { "date": day.isoformat(), "scope_id": str(scope.id), "integration_id": integration_id, "_total": Decimal("0"), "currency": "USD", "_service": {}, }, ) usd = Decimal(str(row.usd or 0)) bucket["_total"] += usd if row.service: bucket["_service"][row.service] = ( bucket["_service"].get(row.service, Decimal("0")) + usd ) # Decimal → float at the response boundary, rounded to 4 dp to # match ``CmdbCloudBillingDaily.total_cost_usd`` column precision. data = [] for day in sorted(per_day.keys()): bucket = per_day[day] bucket["total_cost_usd"] = float( bucket.pop("_total").quantize(_DP_4) ) bucket["cost_by_service"] = { k: float(v.quantize(_DP_4)) for k, v in bucket.pop("_service").items() } data.append(bucket) return success_response( data=data, request=request, meta={"measure": measure}, ) # Decimal quantizer reused across RG-billing aggregation. Defined at # module level so the per-day fold doesn't allocate a fresh Decimal # every call. _DP_4 = Decimal("0.0001") # ── Per-scope month-anchored billing detail (drawer payload) ───── @router.get("/scopes/{scope_id}/billing-detail") async def get_scope_billing_detail_endpoint( request: Request, scope_id: str, month: str | None = Query( None, description=( "Billing month in ``YYYY-MM`` format. Defaults to the " "current UTC month. Strict — single-digit month is " "rejected so MCP/curl callers get a clear 400 instead " "of a silently-accepted query." ), pattern=r"^\d{4}-(0[1-9]|1[0-2])$", ), measure: str | None = Query( None, description=( "成本口径:``list``(默认,业务折前成本/研发口径)或 " "``invoice``(实付,仅 finance / cloud_admin / global_admin)。" ), ), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Rich month-anchored billing payload for one cloud scope. Powers the shared ``SubscriptionBillingDrawer`` on both the employee-facing ``/my/cloud/grants`` page AND the admin ``/admin/cloud/overview`` Subscriptions tab. Single round-trip; bundles identity + summary + daily trend + 6-month history + cost-by-service + top resources + members so the drawer renders in one fetch. Authorisation mirrors :func:`list_billing_daily` exactly — membership / ownership / admin role / finance — including the parent-subscription auth inheritance for RG-typed scopes so an RG owner who deep-links into a child can read its billing even if they're not on the parent subscription's grants. Decommissioned parents (``status='closed'``) do NOT grant inherited read on surviving children; the tombstone filter is reproduced verbatim so a redesign here can't quietly drop the safeguard. Member-roster role-name redaction is enforced for plain employees (no ``can_read_cloud``): a co-member viewing this drawer sees the other members' names but not their cloud privilege levels, same disclosure gate as ``/my-subscriptions``. Without this, the drawer would be a side-channel for the role-info that PRs 122–130 locked down on the list view. """ from app.domains.cmdb.scope_billing_detail import get_scope_billing_detail # Reject obviously-malformed UUIDs at the boundary so the response # is a clean 404 rather than a 500 from ``db.get`` raising. try: scope_uuid = uuid.UUID(scope_id) except (ValueError, AttributeError) as exc: raise NotFoundError("CmdbCloudScope", scope_id) from exc scope = await db.get(CmdbCloudScope, scope_uuid) if scope is None: raise NotFoundError("CmdbCloudScope", scope_id) # First — basic "can read any cloud-resource" gate. Employees with # zero cloud grants still pass this (the role check is broad); the # per-scope membership check below is the real authorisation. if not can_view_cloud_resources(user.role_names): raise ForbiddenError() # Per-scope membership / ownership check for non-admins. Admins # (``can_read_cloud``) and finance readers (``can_view_finance``) # see all scopes; everyone else must be on the scope or own it. # Parent-subscription inheritance for RG-typed scopes — copied # from list_billing_daily so the two surfaces agree. privileged = ( can_read_cloud(user.role_names) or can_view_finance(user.role_names) ) if not privileged: scope_ids_for_perm = {str(scope.id)} owner_ids: set[str] = set() if scope.owner_person_id: owner_ids.add(scope.owner_person_id) if scope.parent_scope_id: try: parent_uuid = uuid.UUID(scope.parent_scope_id) except (ValueError, TypeError): parent_uuid = None if parent_uuid is not None: parent = await db.get(CmdbCloudScope, parent_uuid) # Tombstone filter on the parent: a closed parent must # not grant residual read on a surviving child. Same # safeguard list_billing_daily applies. if ( parent is not None and parent.status != SCOPE_STATUS_CLOSED ): scope_ids_for_perm.add(str(parent.id)) if parent.owner_person_id: owner_ids.add(parent.owner_person_id) is_owner = str(user.id) in owner_ids is_member = await db.scalar( select(func.count(CmdbCloudAccessGrant.id)).where( CmdbCloudAccessGrant.scope_id.in_(scope_ids_for_perm), CmdbCloudAccessGrant.person_id == str(user.id), CmdbCloudAccessGrant.status.in_( (GRANT_STATUS_ACTIVE, GRANT_STATUS_PENDING), ), ), ) or 0 if not is_owner and not is_member: raise ForbiddenError() redact_member_roles = not can_read_cloud(user.role_names) # 口径路由:默认业务折前成本(研发口径);``invoice`` 仅保密圈 # (finance / cloud_admin / global_admin),无权限 → 403,绝不 # 静默降级(保密圈用户误把研发口径当实付做决策更危险)。 from app.domains.cmdb.cost_view import resolve_measure try: resolved_measure = resolve_measure(measure, user.role_names) except ValueError as exc: raise ValidationError(str(exc)) from exc if resolved_measure is None: raise ForbiddenError() data = await get_scope_billing_detail( db, scope=scope, month=month, redact_member_roles=redact_member_roles, measure=resolved_measure, ) return success_response(data=data, request=request) # ── Custom-dimension billing query (admin-only) ────────────────── @router.get("/billing/by-dimension") async def list_billing_by_dimension( request: Request, integration_id: str = Query(..., description="UUID of a cloud integration"), group_by: str = Query( ..., description=( "Comma-separated dimension list. Cloud-specific aliases " "from the driver's supported_billing_dimensions(). " "``tag:KEY`` requests the provider's tag with key ``KEY``. " "AWS hard-caps at 2 dimensions per query; Azure has no cap." ), examples=["service", "service,tag:Project", "subscription,service"], ), start: date | None = None, end: date | None = None, scope_id: str | None = Query( None, description=( "Optional scope filter. AWS: pins LINKED_ACCOUNT to the " "scope's external_id (real 12-digit AWS account). Azure: " "required — Cost Management queries are subscription-scoped. " "Aliyun/Tencent/UCloud: not yet implemented." ), ), filters: str | None = Query( None, description=( "JSON object mapping dimension alias → list[str] of allowed " "values, e.g. ``{\"service\":[\"EC2 - Other\"]," "\"tag:Project\":[\"tidb\"]}``. AND-ed with the scope filter." ), ), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Pull a per-day cost breakdown grouped by one or more cloud-native dimensions, computed live against the provider API. Read access: only ``can_manage_cloud`` (IT admin / cloud admin / finance — anyone with the privilege to look at unaggregated cost data) — the response can include subscription/service granular cost figures we don't pre-aggregate or filter for non-admin users. Window: capped at 90 days end - start to keep CE / Cost Management requests within their hard size limits and to bound per-request cost. Drivers that can't honor a dimension raise 400; drivers that don't yet ship dimension support raise 501 (UI surfaces a "billing dimension query not available for this cloud" hint). """ if not can_manage_cloud(user.role_names): raise ForbiddenError() end_dt = end or (datetime.now(UTC).date() - timedelta(days=1)) start_dt = start or (end_dt - timedelta(days=29)) if end_dt < start_dt: raise AppException( code="BAD_DATE_RANGE", message="end < start", status_code=400, ) if (end_dt - start_dt).days > 90: raise AppException( code="BILLING_RANGE_TOO_LARGE", message=( "max 90-day window per request — split into multiple " "queries if you need a longer span" ), status_code=400, ) parsed_group_by = [g.strip() for g in group_by.split(",") if g.strip()] if not parsed_group_by: raise AppException( code="MISSING_GROUP_BY", message="group_by required", status_code=400, ) parsed_filters: dict[str, list[str]] = {} if filters: try: raw = _json.loads(filters) except (ValueError, TypeError) as e: raise AppException( code="BAD_FILTERS_JSON", message=f"filters must be valid JSON: {e}", status_code=400, ) from e if not isinstance(raw, dict): raise AppException( code="BAD_FILTERS_SHAPE", message="filters must be an object mapping str → list[str]", status_code=400, ) for k, v in raw.items(): if not isinstance(v, list) or not all(isinstance(x, str) for x in v): raise AppException( code="BAD_FILTERS_SHAPE", message=f"filters[{k!r}] must be list[str]", status_code=400, ) parsed_filters[str(k)] = list(v) try: integration_uuid = uuid.UUID(integration_id) except ValueError as exc: raise NotFoundError("Integration", integration_id) from exc integration = await db.get(Integration, integration_uuid) if integration is None or integration.kind != "cloud": raise NotFoundError("Integration", integration_id) scope_ref: CloudScopeRef | None = None if scope_id: try: scope_uuid = uuid.UUID(scope_id) except ValueError as exc: raise NotFoundError("CmdbCloudScope", scope_id) from exc scope = await db.get(CmdbCloudScope, scope_uuid) if scope is None or str(scope.integration_id) != str(integration.id): raise NotFoundError("CmdbCloudScope", scope_id) scope_ref = CloudScopeRef( scope_type=scope.scope_type, external_id=scope.external_id, display_name=scope.display_name, ) driver = await _load_cloud_driver(db, integration) # Surface the supported-dimension contract back to the caller so # the UI can validate before the round-trip — but we still # validate server-side too (don't trust the client). supported = driver.supported_billing_dimensions() for d in parsed_group_by: base = d.split(":", 1)[0].lower() if base not in supported and d.lower() not in supported: try: await driver.close() except Exception: # noqa: BLE001 pass raise AppException( code="DIMENSION_NOT_SUPPORTED", message=( f"{integration.provider!r} does not support dimension " f"{d!r} — supported: {sorted(supported)}" ), status_code=400, ) # Resolve BQ billing source for AWS-on-BQ integrations. Other # providers / AWS-on-CE get None and the driver short-circuits. # Misconfiguration surfaces as 500 with a specific message rather # than a generic "AccessDenied" pile. from app.domains.cmdb.lifecycle_service import _resolve_aws_bq_billing try: bq_billing = await _resolve_aws_bq_billing(db, integration) except RuntimeError as exc: try: await driver.close() except Exception: # noqa: BLE001 pass raise AppException( code="BILLING_BQ_CONFIG_INVALID", message=str(exc), status_code=500, ) from exc try: rows = await driver.list_billing_by_dimension( start_dt, end_dt, group_by=parsed_group_by, scope=scope_ref, filters=parsed_filters or None, bq_billing=bq_billing, ) except NotImplementedError as e: raise AppException( code="BILLING_DIMENSION_UNSUPPORTED", message=( f"cloud driver {integration.provider!r} hasn't shipped " "list_billing_by_dimension yet" ), status_code=501, ) from e except ValueError as e: # Driver-level validation (e.g. AWS's 2-dimension cap). raise AppException( code="BAD_DIMENSION_REQUEST", message=str(e), status_code=400, ) from e finally: try: await driver.close() except Exception: # noqa: BLE001 pass data = [ { "date": r.date.isoformat(), "dimensions": r.dimensions, "cost": r.cost, "currency": r.currency, } for r in rows ] return success_response( data={ "integration_id": str(integration.id), "provider": integration.provider, "start": start_dt.isoformat(), "end": end_dt.isoformat(), "group_by": parsed_group_by, "filters": parsed_filters, "rows": data, }, request=request, ) @router.get("/billing/dimensions") async def list_billing_supported_dimensions( request: Request, integration_id: str = Query(..., description="UUID of a cloud integration"), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Surface which dimensions :func:`list_billing_by_dimension` accepts for this integration. UI uses it to populate the group-by dropdown without baking provider-specific lists into the frontend. """ if not can_manage_cloud(user.role_names): raise ForbiddenError() try: integration_uuid = uuid.UUID(integration_id) except ValueError as exc: raise NotFoundError("Integration", integration_id) from exc integration = await db.get(Integration, integration_uuid) if integration is None or integration.kind != "cloud": raise NotFoundError("Integration", integration_id) driver = await _load_cloud_driver(db, integration) try: supported = sorted(driver.supported_billing_dimensions()) finally: try: await driver.close() except Exception: # noqa: BLE001 pass return success_response( data={ "integration_id": str(integration.id), "provider": integration.provider, "dimensions": supported, }, request=request, ) # ── Scope-creation request ───────────────────────────────────────── class ScopeRequestSubmitBody(BaseModel): integration_id: str # 50, not 255: becomes the cloud-side DisplayName (Aliyun RD + AWS # Organizations both cap at 50). Character-level rules + the # at-least-one-alphanumeric guard live in ``_validate_name``. name: str = Field(min_length=1, max_length=50) description: str | None = Field(default=None, max_length=2000) monthly_budget_usd: float = Field(gt=0, le=10_000_000) # Free-form short string the requester picks when filing the # request — typically "production" / "test" / "POC" / "personal" # but the Feishu template uses a free-text widget so we mirror # that and let the requester write whatever fits. Audit reviewers # see this verbatim in the Feishu approval card. cloud_account_type: str = Field(min_length=1, max_length=64) class ScopeRequestRejectBody(BaseModel): reason: str | None = Field(default=None, max_length=500) def _serialize_scope_request( req: CmdbCloudScopeRequest, requester: IdentityPerson | None, approver: IdentityPerson | None, ) -> dict[str, Any]: return { "id": req.id, "requester_person_id": req.requester_person_id, "requester_label": ( (requester.display_name or requester.work_email) if requester is not None else None ), "integration_id": req.integration_id, "provider": req.provider, "name": req.name, "description": req.description, "monthly_budget_usd": float(req.monthly_budget_usd), "status": req.status, "approver_person_id": req.approver_person_id, "approver_label": ( (approver.display_name or approver.work_email) if approver is not None else None ), "approved_at": req.approved_at.isoformat() if req.approved_at else None, "rejected_at": req.rejected_at.isoformat() if req.rejected_at else None, "rejection_reason": req.rejection_reason, "created_scope_id": req.created_scope_id, "provisioned_at": ( req.provisioned_at.isoformat() if req.provisioned_at else None ), "failure_reason": req.failure_reason, "feishu_instance_code": req.feishu_instance_code, "created_at": req.created_at.isoformat(), "updated_at": req.updated_at.isoformat(), } async def _hydrate_scope_requests( db: AsyncSession, rows: list[CmdbCloudScopeRequest], ) -> list[dict[str, Any]]: if not rows: return [] # Un-expire before serializing (2026-08-17 prod incident, trace # 79b802e7-…). ``updated_at`` carries a SQL-side # ``onupdate=func.now()``, so every flush that UPDATEs a # scope-request row leaves that attribute EXPIRED — only the DB # knows the new value. ``_serialize_scope_request`` then reads it # *synchronously*, and a sync lazy-load on an AsyncSession raises # ``MissingGreenlet`` instead of quietly awaiting. Every mutation # endpoint here commits BEFORE hydrating, so the failure landed # after the point of no return: 「回填云端 ID」 created the scope, # the budget, the alert rules, pre-warmed the role catalog and # notified the requester — then answered 500 and the admin saw a # failure toast for work that had fully succeeded. Refreshing # inside the ``await`` makes the read a loaded one. for r in rows: state = sa.inspect(r) if state.expired or state.expired_attributes: await db.refresh(r) person_ids = { pid for r in rows for pid in (r.requester_person_id, r.approver_person_id) if pid } persons: dict[str, IdentityPerson] = {} if person_ids: rows_p = (await db.execute( select(IdentityPerson).where( IdentityPerson.id.in_([uuid.UUID(p) for p in person_ids]), ) )).scalars().all() persons = {str(p.id): p for p in rows_p} return [ _serialize_scope_request( r, persons.get(r.requester_person_id), persons.get(r.approver_person_id) if r.approver_person_id else None, ) for r in rows ] @router.post("/scope-requests", status_code=201) async def create_scope_request( body: ScopeRequestSubmitBody, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Employee submits a request to create a new sub-account / subscription. IT admin sees a pending row + Feishu/email notification; on approval Helix calls the driver's ``create_scope`` and binds the requester as owner. """ service = CloudScopeRequestService(db) trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) req = await service.submit( requester=user, integration_id=body.integration_id, name=body.name, description=body.description, monthly_budget_usd=body.monthly_budget_usd, cloud_account_type=body.cloud_account_type, trace_id=trace_id, ) await db.commit() [data] = await _hydrate_scope_requests(db, [req]) return success_response(data=data, request=request) @router.get("/scope-requests") async def list_scope_requests( request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), status: str | None = Query(None), provider: list[str] | None = Query( None, description=( "Repeated as ``?provider=aws&provider=gcp``. Narrows to " "requests raised against those cloud providers." ), ), search: str | None = Query( None, description=( "Case-insensitive substring over the request's name and " "description." ), ), page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), mine_only: bool = Query( False, description=( "Set true to scope to the current user's own requests. " "Admins (can_manage_cloud) get the full tenant view by " "default." ), ), ): q = select(CmdbCloudScopeRequest) if mine_only or not can_manage_cloud(user.role_names): q = q.where( CmdbCloudScopeRequest.requester_person_id == str(user.id), ) if status: q = q.where(CmdbCloudScopeRequest.status == status) # ``provider`` / ``search`` mirror the filter bar the admin queue # already offers on the sibling 订阅列表 tab. Without them the # pending queue was status-only, so "which AWS requests are still # open" meant paging by eye. if provider: q = q.where(CmdbCloudScopeRequest.provider.in_(provider)) if search and search.strip(): needle = f"%{search.strip()}%" q = q.where(or_( CmdbCloudScopeRequest.name.ilike(needle), CmdbCloudScopeRequest.description.ilike(needle), )) count_q = select(func.count()).select_from(q.subquery()) total = (await db.execute(count_q)).scalar() or 0 q = (q.order_by(CmdbCloudScopeRequest.created_at.desc()) .offset((page - 1) * page_size).limit(page_size)) rows = list((await db.execute(q)).scalars().all()) data = await _hydrate_scope_requests(db, rows) return paginated_response( data=data, total=total, page=page, page_size=page_size, request=request, ) @router.post("/scope-requests/{req_id}/withdraw") async def withdraw_scope_request( req_id: str, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): service = CloudScopeRequestService(db) trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) req = await service.withdraw( req_id=req_id, actor=user, trace_id=trace_id, ) await db.commit() [data] = await _hydrate_scope_requests(db, [req]) # False ⇒ Helix withdrew but the parallel Feishu instance is still # pending (e.g. template withdraw disabled) — frontend warns. data["feishu_cancel_synced"] = service.last_feishu_cancel_synced return success_response(data=data, request=request) # Per-provider default owner-equivalent role used by the # post-provision auto-grant flow (2026-05-13). Each entry is the # ``(role_key, role_display_name)`` we hand to the driver's # ``grant_access`` so the requester lands with the same authority # they had pre-Helix (root / Owner / equivalent) on the # freshly-minted scope. # # Only providers whose ``driver.create_scope`` is actually # implemented appear here: # * GCP — ``roles/owner`` is the predefined Owner role; binds via # ``cloudresourcemanager.projects.setIamPolicy``. # * AWS — by-name resolution of an Identity Center permission set # (default ``DBaaS-DevLeader-Role``, overridable via # ``platform_settings`` key # ``cloud.aws.scope_request_default_owner_role_name``). The # resolved ARN is then bound via ``sso-admin.create_account_assignment`` # in :meth:`AwsCloudDriver._grant_sso_sync`. AWS-specific # resolution lives in :func:`_resolve_default_owner_role` # because the ARN is not known until runtime — see that helper # for the dynamic path. # An entry is required as soon as a provider can reach ``provisioned`` # AT ALL — NOT only when its driver grows ``create_scope``. That older # framing (the line this comment replaces said "future providers slot in # here when their drivers grow create_scope support") is what left Azure # out for three months: Azure has no ``create_scope``, so every Azure # sub-account arrives through the ``manual-provision`` rescue path, which # runs the very same ``_post_provision_finalize`` — and with no map entry # ``_resolve_default_owner_role`` returns ``None``, the finalize # audit-logs ``auto_owner_grant status=skipped`` and stops. Result: # requester is Helix-side owner with ZERO cloud-side IAM and an empty # member list — exactly the split-brain PR #279 existed to kill. Seen in # prod on ``azure-ai-prod`` (scope 7845b0f7, 2026-08-17), and before that # ``e7babe7b`` (2026-08-14) / ``ae0486fc`` (2026-08-11) / tencent # ``7f4642bb`` (2026-08-12). Manual-provision providers need this map # MORE than auto-create ones, not less. # # Each value is the role_key the provider's own ``grant_access`` # consumes, cross-checked against that driver's ``owner_role_keys`` # (what the grants-sync sweep recognises as owner-equivalent) so the # grant this map mints is the same role the sweep will later adopt. # Note the two are NOT always spellable the same — see the ``aliyun`` # entry below — so this map stays explicit rather than deriving itself # from ``owner_role_keys``. ``tests/test_cmdb/ # test_auto_owner_role_map_coverage_2026_08_17.py`` holds both ends. _DEFAULT_OWNER_ROLE_BY_PROVIDER: dict[str, tuple[str, str]] = { "gcp": ("roles/owner", "Owner"), # AWS entry uses a sentinel that signals "resolve at runtime via # _resolve_default_owner_role"; the resolved (ARN, display_name) # replaces these values before the grant is dispatched. We keep # the entry in the map so the provider-handled list is centrally # discoverable, and we sanity-check the presence of an entry # before doing the dynamic resolve. "aws": ("__resolve_at_runtime__", "DBaaS-DevLeader-Role"), # Aliyun (domestic) and AliCloud 海外 consume DIFFERENT role_key shapes # because their ``grant_access`` impls differ — they must NOT share a # single ``("admin", "admin")`` entry: # • Domestic ``aliyun`` (AliyunCloudDriver) has NO grant_access # override; it uses the base RAM impl, which rejects any role_key # without a ``Type:Name`` colon ("Aliyun role_key must be # System:PolicyName or Custom:PolicyName"). So owner = the RAM # system policy ``System:AdministratorAccess`` — it splits into # PolicyType=System / PolicyName=AdministratorAccess and matches the # domestic driver's ``owner_role_keys = {"AdministratorAccess"}``. # • Intl ``alicloud_intl`` (AlicloudIntlCloudDriver) OVERRIDES # grant_access and, for an ``aliyun_account`` scope with a CloudSSO # directory, maps the bare AccessConfiguration name ``admin`` → # CreateAccessAssignment. So owner = ``admin``. # Added 2026-06-03 (corrected 2026-06-04): without these entries the # post-provision finalize was a no-op for Aliyun, so a self-service # sub-account requester landed as Helix-side owner with ZERO cloud-side # IAM (prod: dev-offline-afterworld-aliyun / scope 32e44d04). The first # cut shipped ``("admin", "admin")`` for BOTH providers, which silently # broke the DOMESTIC path: ``admin`` has no colon, so the RAM # grant_access returned ok=False and the grant stranded PENDING — the # exact zero-IAM bug this map was meant to close. "aliyun": ("System:AdministratorAccess", "AdministratorAccess"), "alicloud_intl": ("admin", "admin"), # ── Manual-provision-only providers (no ``create_scope``) ───────── # Added 2026-08-17 after the ``azure-ai-prod`` incident above. # # Azure: the built-in **Owner** role, whose definition GUID is stable # across every tenant. ``AzureCloudDriver.grant_access`` passes # ``ctx.role_key`` straight through as ARM's ``roleDefinitionId``, and # ARM accepts the tenant-level ``/providers/…/roleDefinitions/{guid}`` # form for built-ins on a subscription-scoped assignment. Verified # against prod rather than inferred: this exact string is the # ``role_key`` on 57 ACTIVE discovered Azure Owner grants and on the # ``Owner`` row of the scope role catalog, and it is the sole member of # ``AzureCloudDriver.owner_role_keys`` (the set grants_sync uses to # auto-adopt scope owners — so the row we mint here is the row that # sweep recognises). Owner is ``restricted`` in Helix's role # governance; that is deliberate and matches GCP's ``roles/owner`` # already in this map — the *initial* owner of a sub-account they # themselves requested is the one case where owner-equivalent # self-grant is the intended outcome (see # ``provision_initial_owner_grant``'s docstring). "azure": ( "/providers/Microsoft.Authorization/roleDefinitions/" "8e3af657-a8ff-443c-a75c-2fe8c4bcb635", "Owner", ), # Tencent: CAM policy NAME. ``TencentCloudDriver._grant_sync`` takes # ``role_key`` as a PolicyName, resolves it to a PolicyId via # ``_resolve_policy_id`` and attaches it to the (idempotently created) # CAM sub-user — so the bare name is the right shape, no ``Type:Name`` # prefix (that convention is domestic Aliyun's RAM impl, above). # ``AdministratorAccess`` is the sole member of the driver's # ``owner_role_keys`` and is present in the prod Tencent role catalog. "tencent": ("AdministratorAccess", "AdministratorAccess"), # UCloud: same shape — ``UCloudDriver.grant_access`` reads ``role_key`` # as an IAM policy name, and ``AdminPolicy`` is both the driver's sole # ``owner_role_keys`` member and present in the prod UCloud catalog. # No UCloud scope has come through self-service yet, so this entry is # inert today; it is here so the next one doesn't repeat the Azure # hole. Worst case if the connector is mid-drift: ``grant_access`` # returns ok=False and the grant row stays PENDING with # ``detail.last_error`` — a visible failure the backfill cron retries, # strictly better than today's silent skip. "ucloud": ("AdminPolicy", "AdminPolicy"), # Deliberately absent: ``ksyun`` declares neither ``owner_role_keys`` # nor a ``grant_access`` override, so there is no owner-equivalent # role to mint (its one integration row is disabled). Adding an entry # would only make the backfill cron burn ticks on a guaranteed no-op. } # Platform-setting key for the AWS default owner-role NAME (the # Identity Center permission-set Name, not the ARN — ARNs vary per # instance and aren't user-meaningful). Admins can override the # default ``DBaaS-DevLeader-Role`` here without a code deploy if the # org renames or replaces the permission set. _AWS_DEFAULT_OWNER_ROLE_NAME_SETTING = ( "cloud.aws.scope_request_default_owner_role_name" ) _AWS_DEFAULT_OWNER_ROLE_NAME_FALLBACK = "DBaaS-DevLeader-Role" # How long the ``cmdb.auto_owner_grant_backfill`` cron skips a scope # after it stamps ``auto_owner_grant_unresolvable_at``. The owner-role # permission-set is deterministically unresolvable (the configured name # has no match in the provider role catalog), so re-running the paid # Identity Center ``list_scope_roles`` call every 10-min tick is pure # waste — but we DO want eventual self-healing once an admin renames the # permission set (in Identity Center or via the # ``cloud.aws.scope_request_default_owner_role_name`` platform setting), # so the scope re-qualifies after this window. Short enough that # recovery is same-day, long enough that the paid call fires a few times # a day, not ~144. Manual owner re-assignment and a successful # resolve+dispatch clear the marker immediately (no need to wait this # out). Migration 244. _OWNER_ROLE_UNRESOLVABLE_RETRY_HOURS = 6 class _OwnerRoleUnresolvable(Exception): """Raised by :func:`_resolve_default_owner_role` when the configured auto-Owner permission-set is DETERMINISTICALLY unresolvable — the provider role catalog was listed successfully but contains no entry matching the configured name. Distinct from a transient failure (driver crash, Identity Center throttle, LockBusy): those propagate as ordinary exceptions and must NOT stamp the backfill backoff marker — the next tick should retry immediately. This signal tells ``_post_provision_finalize`` to stamp ``cmdb_cloud_scopes.auto_owner_grant_unresolvable_at`` so the cron stops re-running the paid ``list_scope_roles`` call every 10 min. """ def __init__(self, provider: str, reason: str) -> None: self.provider = provider self.reason = reason super().__init__(f"{provider}: {reason}") async def _resolve_default_owner_role( provider: str, *, integration_id: str, session: AsyncSession, ) -> tuple[str, str] | None: """Return ``(role_key, role_display_name)`` for the auto-Owner grant on this provider, resolving dynamic entries. GCP-style static entries return their tuple directly. AWS reads a permission-set NAME (from platform_settings or the hardcoded ``DBaaS-DevLeader-Role`` default), then queries Identity Center via ``driver.list_scope_roles`` to map the name → ARN at call time. Returns ``None`` when there is no auto-grant policy to apply or the attempt should be retried as-is next tick: * the provider isn't in :data:`_DEFAULT_OWNER_ROLE_BY_PROVIDER` (no auto-grant policy for this provider), or * the integration row vanished mid-resolve (transient). Raises :class:`_OwnerRoleUnresolvable` when the role catalog WAS listed successfully but contains no permission set matching the configured name — a deterministic config miss that will keep failing until an admin fixes the name, so the caller stamps the backfill backoff marker instead of retrying every tick. Transient driver errors (Identity Center throttle, etc.) propagate as their original exception type — neither ``None`` nor ``_OwnerRoleUnresolvable`` — so the next tick retries immediately. """ entry = _DEFAULT_OWNER_ROLE_BY_PROVIDER.get(provider) if not entry: return None role_key, display_name = entry if role_key != "__resolve_at_runtime__": return (role_key, display_name) # Provider-specific dynamic resolution. if provider == "aws": from app.integrations.driver_base import CloudScopeRef from app.integrations.drivers.aws_cloud import AwsCloudDriver from app.integrations.models import Integration as _Integration from app.integrations.service import ( _resolve as _resolve_integration, ) from app.integrations.service import ( get_platform_setting, ) # Read overridable role name; fall back to the hardcoded # default so a brand-new install works out of the box. # NB: `get_platform_setting` signature is # ``(key, default=None, *, session=None)`` — session is a # keyword-only arg. The original (pre-2026-06-02) call passed # `session` positionally which silently aliased to `key`, then # also passed `default=` as kwarg → TypeError "got multiple # values for argument 'default'". The TypeError was masked for # weeks because PR #503's `_post_provision_finalize` was # scheduled via FastAPI BackgroundTasks (silently dropped in # prod uvicorn). PR #671 made the finalize inline-awaited, # which surfaced this latent crash on every AWS scope # approval. Pin the keyword-only contract so a future signature # tweak can't re-introduce the same shape of bug silently. configured_name = await get_platform_setting( _AWS_DEFAULT_OWNER_ROLE_NAME_SETTING, default=_AWS_DEFAULT_OWNER_ROLE_NAME_FALLBACK, session=session, ) role_name = str(configured_name or "").strip() or \ _AWS_DEFAULT_OWNER_ROLE_NAME_FALLBACK integration = await session.get( _Integration, uuid.UUID(integration_id), ) if integration is None: logger.warning( "AWS auto-owner-grant resolve: integration %s missing", integration_id, ) return None resolved = await _resolve_integration(session, integration) driver = AwsCloudDriver( integration_id=str(integration.id), config=resolved.config, ) try: # list_scope_roles needs SOME CloudScopeRef but only # uses .scope (for back-reference) when building # CloudRoleInfo — we don't care about that field here. permission_sets = await driver.list_scope_roles( CloudScopeRef(scope_type="aws_account", external_id="*"), ) finally: try: await driver.close() except Exception: # noqa: BLE001 pass for role_info in permission_sets: if (role_info.display_name or "").strip() == role_name: return (role_info.role_key, role_info.display_name or role_name) logger.warning( "AWS Identity Center has no permission set Name=%r — " "auto-owner-grant will be skipped (configurable via " "platform_settings key %r)", role_name, _AWS_DEFAULT_OWNER_ROLE_NAME_SETTING, ) # Deterministic config miss — the catalog listed fine, the name # just isn't in it. Raise so the caller stamps the backfill # backoff marker rather than re-listing every 10-min tick. raise _OwnerRoleUnresolvable( "aws", f"no Identity Center permission set named {role_name!r} " f"(configurable via platform_settings " f"{_AWS_DEFAULT_OWNER_ROLE_NAME_SETTING!r})", ) # Unknown dynamic-resolve sentinel for a provider we don't # know how to resolve — treat as no auto-grant. return None async def _mark_owner_role_unresolvable( scope_id: str, *, provider: str, reason: str, trace_id: str, retry_hours: int = _OWNER_ROLE_UNRESOLVABLE_RETRY_HOURS, reason_code: str = "owner_role_unresolvable", ) -> None: """Stamp ``cmdb_cloud_scopes.auto_owner_grant_unresolvable_at`` so the backfill cron stops re-attempting a deterministically unresolvable auto-Owner grant every 10-min tick. Best-effort: opens its own session, never raises (the caller is the best-effort finalize path). Audit-logs the skip AT MOST ONCE per re-attempt window — only when the marker was previously NULL or has aged past ``retry_hours`` (a stale re-scan). ``retry_hours`` MUST be the SAME window the cron used to re-surface this scope (``config.unresolvable_retry_hours`` → threaded through ``_post_provision_finalize``); deriving it from the bare constant here while the cron used an admin override desyncs the audit short-circuit from the candidate-skip window (a re-surfacing with retry_hours<6 would suppress its own audit row). Defaults to the constant for the inline/manual approve path, which never overrides it. A scope that is already marked-and-fresh won't normally reach here (the cron candidate scan excludes it), but the audit short-circuit keeps the inline approve path from re-logging on a manual retry. """ from app.audit.service import AuditService as _AuditService from app.core.database import async_session_factory as _factory from app.domains.cmdb.models import CmdbCloudScope as _CmdbCloudScope try: async with _factory() as session: scope = await session.get(_CmdbCloudScope, uuid.UUID(scope_id)) if scope is None or scope.deleted_at is not None: return now = datetime.now(UTC) prev = scope.auto_owner_grant_unresolvable_at # First-stamp or stale re-stamp → this is a fresh surfacing, # worth an audit row. Already-fresh marker → re-stamp the # clock forward silently (no duplicate audit noise). stale_cutoff = now - timedelta( hours=max(1, retry_hours) ) is_fresh_surfacing = prev is None or _as_aware(prev) < stale_cutoff scope.auto_owner_grant_unresolvable_at = now scope.updated_at = now if is_fresh_surfacing: audit = _AuditService(session) await audit.log( trace_id=trace_id, actor_type="system", actor_user_id=None, action="cloud.scope_request.auto_owner_grant", status="skipped", resource_type="cmdb_cloud_scope", resource_id=scope_id, detail={ "reason": reason_code, "provider": provider, "detail": reason[:300], }, ) await session.commit() except Exception: # noqa: BLE001 — best-effort marker write logger.exception( "failed to stamp auto_owner_grant_unresolvable_at for " "scope=%s", scope_id, ) def _as_aware(dt: datetime) -> datetime: """Normalise a possibly-naive DATETIME (MySQL/TiDB strip tzinfo on a fresh load — see the project-wide naive-datetime invariant) to UTC so the staleness comparison never raises on aware-vs-naive.""" return dt if dt.tzinfo is not None else dt.replace(tzinfo=UTC) async def _post_provision_finalize( scope_id: str, trace_id: str | None = None, *, is_backfill: bool = False, unresolvable_retry_hours: int | None = None, ) -> None: """Best-effort background work after a scope is provisioned. Two steps, both running under a single acquisition of the ``cloud_sync:{integration_id}`` Redis advisory lock so neither races the cron handlers (``run_scopes_sync`` / ``run_roles_sync`` / ``run_grants_sync``) nor the admin "sync now" sweep, which all perform mark-and-sweep on the ``last_seen_sync_job_id`` columns. 1. **Pre-warm the role catalog** for the new scope so the "添加成员" dropdown is non-empty by the time the requester opens it. Without this, the catalog is populated only by the daily ``cmdb.roles_sync`` cron — up to 24h late. 2. **Auto-grant the requester the owner-equivalent role** (per-provider map at :data:`_DEFAULT_OWNER_ROLE_BY_PROVIDER`). Goes through ``CloudLifecycleService.provision_initial_owner_grant`` which writes a ``CmdbCloudAccessGrant`` row AND invokes the driver's ``grant_access`` to bind cloud-side IAM — so the requester appears in the member list AND can actually use the new project. Replaces the older "requester must self-add via owner-add" UX that confused users (the Helix-side ``owner_person_id`` made them appear to be owner, but they had no cloud-side IAM and an empty member list — a misleading split-brain state). Failures in either step are logged and swallowed: the scope itself is already provisioned, and the cron remains the ultimate backstop for the catalog. If the auto-grant fails, the requester can re-bind via the standard owner-add flow. """ from app.audit.service import AuditService as _AuditService from app.core.database import async_session_factory as _factory from app.domains.cmdb.lifecycle_service import ( CloudLifecycleService as _CloudLifecycleService, ) from app.domains.cmdb.lifecycle_service import ( GrantPrincipalNotFoundError as _GrantPrincipalNotFoundError, ) from app.domains.cmdb.models import CmdbCloudScope as _CmdbCloudScope # Resolve integration_id + provider in a quick read so we can scope # the lock + look up the per-provider owner role. Done outside the # lock so a slow scope-row fetch doesn't widen the critical # section unnecessarily. try: async with _factory() as session: scope = await session.get(_CmdbCloudScope, uuid.UUID(scope_id)) if scope is None or scope.deleted_at is not None: logger.info( "post-provision finalize skipped — scope %s gone", scope_id, ) return integration_id = str(scope.integration_id) # Provider lives on the integration row, not the scope — # the scope_type encodes provider-specific shape # (``aws_account`` / ``gcp_project`` / …) but not the # canonical provider key the driver registry uses. from app.integrations.models import Integration as _Integration integration = await session.get( _Integration, uuid.UUID(integration_id), ) # Admin-disabled integrations are off-limits — driver calls # may opaque-error from missing creds (e.g. AWS Identity Center # AccessDenied) and the inline approve path / backfill cron # would log a failure-audit on every retry. Mirror the # ``integration.enabled`` gate the lifecycle-service helpers # at lifecycle_service.py:5873 / 6249 / 6662 / 7623 already # apply on similar provider-touching flows. if integration is None or not integration.enabled: logger.info( "post-provision finalize skipped — scope=%s " "integration=%s missing or disabled", scope_id, integration_id, ) return provider = str(integration.provider) except Exception: logger.exception( "post-provision finalize for scope=%s could not " "resolve integration_id", scope_id, ) return # Prefer the caller's request trace_id so all audit rows under one # approve roundtrip share a correlation key (per CLAUDE.md rule #6 # "Audit by design: every key action traceable with trace_id"). # Fallback to the synthetic key for the backfill cron / Feishu # callback paths that have their own trace_id discipline. if not trace_id: trace_id = f"post-provision-finalize:{scope_id}" try: async with distributed_lock( f"cloud_sync:{integration_id}", ttl=600, ): # Step 1 — role catalog pre-warm. try: role_sync_result = await sync_roles_for_scope(scope_id) errors = role_sync_result.get("errors") or [] if errors: logger.warning( "post-provision role sync for scope=%s " "reported errors: %s", scope_id, errors, ) elif role_sync_result.get("status") == "skipped": logger.info( "post-provision role sync for scope=%s " "skipped: %s", scope_id, role_sync_result.get("reason"), ) except Exception: logger.exception( "post-provision role sync for scope=%s crashed " "(continuing to owner-grant step)", scope_id, ) # Step 2 — auto-Owner grant, for any provider with an entry # in :data:`_DEFAULT_OWNER_ROLE_BY_PROVIDER` — which includes # the manual-provision-only ones, since they reach this same # finalize through the rescue path (the older "only providers # with create_scope" reading of this comment is what left # Azure out; see the map's header). AWS resolves the # permission-set ARN at runtime via # _resolve_default_owner_role (Identity Center by-name # lookup) — see that helper for the dynamic path. try: # SAVEPOINT (2026-08-06 review): scope this iteration's writes so a # DB-level failure rolls back only its own work. ``savepoint`` # re-raises everything, so the handler chain below keeps its exact # semantics — it only stops a failure from poisoning the session # for the remaining items and the run's trailing commit. async with savepoint(session, label="owner role resolve"): async with _factory() as grant_session: owner_role = await _resolve_default_owner_role( provider, integration_id=integration_id, session=grant_session, ) if not owner_role: logger.info( "post-provision auto-owner-grant skipped — " "no default owner role resolved for provider=%r " "(scope=%s)", provider, scope_id, ) # Audit-log the skip so an admin can spot a # missing permission-set / unhandled provider # without trawling logs. audit = _AuditService(grant_session) await audit.log( trace_id=trace_id, actor_type="system", actor_user_id=None, action="cloud.scope_request.auto_owner_grant", status="skipped", resource_type="cmdb_cloud_scope", resource_id=scope_id, detail={ "reason": "no default owner role resolved", "provider": provider, }, ) await grant_session.commit() return role_key, role_display_name = owner_role service = _CloudLifecycleService(grant_session) outcome = await service.provision_initial_owner_grant( scope_id=scope_id, role_key=role_key, role_display_name=role_display_name, trace_id=trace_id, # ``is_backfill=True`` means the auto-Owner cron # is filling a gap retroactively — the owner has # had cloud-side access since before Helix had # this code path. Suppress the "[已开通]" member # notification and write the grant row with # ``expires_at=NULL`` so the renewal-nag sweeper # leaves it alone. PR #671 backfill incident # (2026-06-02) — 14 owners got a misleading # "云访问已生效" card before this guard landed. is_backfill=is_backfill, ) if outcome is None: logger.info( "post-provision auto-owner-grant skipped " "for scope=%s — scope/owner missing or " "inactive", scope_id, ) return if outcome.warning: logger.warning( "post-provision auto-owner-grant for " "scope=%s landed with warning: %s", scope_id, outcome.warning, ) except _OwnerRoleUnresolvable as exc: # Deterministic config miss — the configured auto-Owner # permission-set isn't in the provider role catalog. No # grant row gets written, so without a marker the # backfill cron's ``~sub_helix_owner_grant`` predicate # keeps this scope a candidate and re-runs the paid # ``list_scope_roles`` call every 10-min tick forever. # Stamp ``auto_owner_grant_unresolvable_at`` so the cron # backs off (and surfaces the scope to an admin once via # the failure-notifier on its next tick). The marker # clears automatically once the perm-set name is fixed # (the cron re-attempts after the staleness window and # ``provision_initial_owner_grant`` clears it on a # successful resolve+dispatch). logger.warning( "post-provision auto-owner-grant for scope=%s " "deterministically unresolvable — marking for " "backfill-cron backoff: %s", scope_id, exc, ) await _mark_owner_role_unresolvable( scope_id, provider=exc.provider, reason=exc.reason, trace_id=trace_id, retry_hours=( unresolvable_retry_hours if unresolvable_retry_hours is not None else _OWNER_ROLE_UNRESOLVABLE_RETRY_HOURS ), ) except _GrantPrincipalNotFoundError as exc: # Deterministic principal miss — the owner's account is # gone at the provider IdP (e.g. a deleted Google # account: setIamPolicy 400 + ProjectIamPolicyError). # Same shape as _OwnerRoleUnresolvable: no ACTIVE grant # lands, so without a marker the backfill cron re-fires # the doomed IAM call every 10-min tick forever (prod # 2026-07: 31+ failure audits on one scope). Stamp the # backoff marker; the grant row stays PENDING so a # restored account self-heals on the first post-window # retry (``provision_initial_owner_grant`` clears the # marker on a successful dispatch). logger.warning( "post-provision auto-owner-grant for scope=%s: " "owner principal not found at provider — marking " "for backfill-cron backoff: %s", scope_id, exc, ) await _mark_owner_role_unresolvable( scope_id, provider=provider, reason=str(exc), trace_id=trace_id, retry_hours=( unresolvable_retry_hours if unresolvable_retry_hours is not None else _OWNER_ROLE_UNRESOLVABLE_RETRY_HOURS ), reason_code="principal_not_found", ) except ConflictError as exc: # Idempotent re-run case — an active / pending grant # already covers this (scope, owner, role) triple. # Treat as success; we wouldn't want to re-trigger the # driver IAM call either (the cloud-side binding is # also already in place). Audit the skip: a # source='discovered' conflict row is invisible to the # backfill cron's helix-source predicate, and this # branch looping silently (log-only) is exactly how 45 # imported scopes churned scanned=20/fixed=0 with zero # audit trail for weeks (prod 2026-07-03) before the # candidate scan grew its owner-equivalent-any-source # exclusion. The audit row makes any residual # recurrence greppable. logger.info( "post-provision auto-owner-grant for scope=%s " "is a no-op: %s", scope_id, exc, ) try: async with _factory() as audit_session: audit = _AuditService(audit_session) await audit.log( trace_id=trace_id, actor_type="system", actor_user_id=None, action=( "cloud.scope_request.auto_owner_grant" ), status="skipped", resource_type="cmdb_cloud_scope", resource_id=scope_id, detail={ "reason": ( "owner already holds an active/" "pending grant for the owner role" ), "conflict": str(exc)[:300], "is_backfill": is_backfill, }, ) await audit_session.commit() except Exception: # noqa: BLE001 — best-effort audit logger.exception( "failed to audit-log ConflictError skip for " "scope=%s", scope_id, ) except Exception: logger.exception( "post-provision auto-owner-grant for scope=%s " "crashed — requester can re-bind via the standard " "owner-add flow", scope_id, ) except LockBusy: # Cron or admin sync-now is mid-flight. Both eventually iterate # the new scope and refresh its role catalog; neither will # bind the auto-Owner grant though — the auto-grant lives # only on this code path. Schedule an audit row so an # operator can later replay it manually if needed; we cannot # safely retry inside the request handler. The # ``cmdb.auto_owner_grant_backfill`` cron will pick this scope # up on its next pass and complete the grant. logger.info( "post-provision finalize for scope=%s skipped — another " "sync is running for integration=%s; auto-Owner grant " "deferred (backfill cron will retry)", scope_id, integration_id, ) try: async with _factory() as session: audit = _AuditService(session) await audit.log( trace_id=trace_id, actor_type="system", actor_user_id=None, action="cloud.scope_request.auto_owner_grant", status="skipped", resource_type="cmdb_cloud_scope", resource_id=scope_id, detail={ "reason": "another sync was running for integration", "integration_id": integration_id, }, ) await session.commit() except Exception: # noqa: BLE001 — best-effort audit logger.exception( "failed to audit-log LockBusy skip for scope=%s", scope_id, ) except Exception: # noqa: BLE001 # Inline-safety net (2026-06-02): this function is now called # awaited inline from approve_scope_request / manual_provision # (was previously a FastAPI BackgroundTask — silently lost on # every prod approve since PR #279 landed (2026-05-13), audit # table had zero ``cloud.scope_request.auto_owner_grant`` rows # over the ~20-day silent-loss window). Any # unhandled exception from the inner try block (Redis # connection loss, driver crash, _resolve_default_owner_role # raise) must NOT propagate up the request handler — the # scope itself is already provisioned + audited at # ``cloud.scope_request.provision success``; the owner-grant # is a follow-up. The backfill cron will retry on the next # tick. Log loudly so ops can spot recurrent failures. logger.exception( "post-provision finalize for scope=%s crashed — " "auto-Owner grant deferred to backfill cron", scope_id, ) try: async with _factory() as session: audit = _AuditService(session) await audit.log( trace_id=trace_id, actor_type="system", actor_user_id=None, action="cloud.scope_request.auto_owner_grant", status="failure", resource_type="cmdb_cloud_scope", resource_id=scope_id, detail={ "reason": "post_provision_finalize raised", "integration_id": integration_id, }, ) await session.commit() except Exception: # noqa: BLE001 — best-effort audit logger.exception( "failed to audit-log finalize-crash skip for scope=%s", scope_id, ) @router.post("/scope-requests/{req_id}/approve") async def approve_scope_request( req_id: str, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), # Round-49: approving a scope request triggers AWS Organizations # CreateAccount + multi-$/mo budget binding — same catastrophic- # if-stolen-PAT class as the Round-46-gated paths (iam-users/owner, # iam-users/cleanup, dns-record-requests/bypass-execute, scopes/ # decommission, grants/force-sweep). Closes the last cloud_admin- # write PAT lane. _gate: None = Depends(require_browser_session), ): if not can_manage_cloud(user.role_names): raise ForbiddenError() service = CloudScopeRequestService(db) trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) req = await service.approve( req_id=req_id, approver=user, trace_id=trace_id, ) await db.commit() # Post-provision finalize: pre-warm role catalog + auto-grant the # requester an owner-equivalent role on the new scope. See # ``_post_provision_finalize`` for the full contract (both steps # run under a single per-integration Redis lock; failures are # best-effort and the ``cmdb.auto_owner_grant_backfill`` cron is # the backstop). # # Run INLINE (awaited) rather than via FastAPI BackgroundTasks # (2026-06-02 fix). Investigation of scope b521cafb-… (and every # AWS scope created since PR #279 landed 2026-05-13) showed zero # ``cloud.scope_request.auto_owner_grant`` audit rows ever — the # bg-task pattern silently lost every single firing in prod # uvicorn. Adding ~5-15s to the approve response is acceptable on # top of the ~19s AWS CreateAccount poll the admin already waits # for, and the auto-Owner grant is the user-visible payoff of the # whole flow. ``_post_provision_finalize`` wraps its work in # try/except (LockBusy + Exception) so inline-mode cannot fail # the approve response. if ( req.status == SCOPE_REQUEST_STATUS_PROVISIONED and req.created_scope_id ): # Pass the request trace_id so the finalize audit rows # (cloud.scope_request.auto_owner_grant + provision) correlate # with the approve audit row that triggered them. await _post_provision_finalize( req.created_scope_id, trace_id=trace_id, ) [data] = await _hydrate_scope_requests(db, [req]) return success_response(data=data, request=request) @router.post("/scope-requests/{req_id}/reject") async def reject_scope_request( req_id: str, body: ScopeRequestRejectBody, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): if not can_manage_cloud(user.role_names): raise ForbiddenError() service = CloudScopeRequestService(db) trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) req = await service.reject( req_id=req_id, approver=user, reason=body.reason, trace_id=trace_id, ) await db.commit() [data] = await _hydrate_scope_requests(db, [req]) return success_response(data=data, request=request) @router.get("/scope-requests/eligible-integrations") async def list_scope_request_eligible_integrations( request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Every active cloud integration the requester can submit against. All cloud integrations are eligible — both auto-create (driver implements ``create_scope``: AWS Organizations today) and manual fulfillment (admin creates the sub-account in the cloud console after approval, then calls ``manual-provision`` to bind it). The ``auto_create_supported`` flag lets the UI render an "auto" vs "需 IT 手动开通" hint so the requester knows what to expect. """ from app.integrations.registry import get_driver_class rows = list((await db.execute( select(Integration).where( Integration.kind == "cloud", Integration.enabled == True, # noqa: E712 ).order_by(Integration.name.asc()) )).scalars().all()) out: list[dict[str, Any]] = [] for integ in rows: driver_cls = get_driver_class("cloud", integ.provider) if driver_cls is None: # No driver registered at all → can't even tell which # cloud this is. Skip. continue auto_create = ( getattr(driver_cls, "create_scope", None) is not CloudDriver.create_scope ) out.append({ "id": str(integ.id), "name": integ.name, "provider": integ.provider, "slug": integ.slug, "auto_create_supported": auto_create, }) return success_response(data=out, request=request) class ScopeRequestManualProvisionBody(BaseModel): external_id: str = Field(min_length=1, max_length=512) scope_type: str | None = Field(default=None, max_length=32) @router.post("/scope-requests/{req_id}/manual-provision") async def manual_provision_scope_request( req_id: str, body: ScopeRequestManualProvisionBody, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), # Round-49: binds a fresh CmdbCloudScope row + requester ownership; # same catastrophic-if-stolen-PAT posture as ``approve_scope_request``. _gate: None = Depends(require_browser_session), ): """Admin completes a manual fulfillment leg — they created the sub-account in the cloud console themselves; this binds the resulting external_id back to the scope-request and inserts a matching CmdbCloudScope row. Only valid when the request is in ``feishu_approved`` (approve stopped at the driver-capability gate) or ``failed`` (auto path errored, admin is rescuing).""" if not can_manage_cloud(user.role_names): raise ForbiddenError() service = CloudScopeRequestService(db) trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) req = await service.manual_provision( req_id=req_id, external_id=body.external_id, scope_type=body.scope_type, actor=user, trace_id=trace_id, ) await db.commit() # Same finalize as the auto-approve path — see _post_provision_finalize. # Runs INLINE (awaited); see the matching comment in # ``approve_scope_request`` above for why we no longer use # ``background.add_task`` here (silent task-loss in prod uvicorn # caused PR #279's auto-Owner grant to never land for any AWS # scope until this fix). if ( req.status == SCOPE_REQUEST_STATUS_PROVISIONED and req.created_scope_id ): # Pass the request trace_id so the finalize audit rows correlate # with the manual_provision audit row that triggered them. await _post_provision_finalize( req.created_scope_id, trace_id=trace_id, ) [data] = await _hydrate_scope_requests(db, [req]) return success_response(data=data, request=request) # ── Lifecycle timelines (per-record audit-backed event stream) ─────── @router.get("/scope-requests/{req_id}") async def get_scope_request( req_id: str, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Single-fetch hydrated scope-request row by id. Mirrors the RBAC posture of ``get_scope_request_timeline`` — admin (``can_manage_cloud``) sees any row; non-admin sees only their own request. Returns the same shape ``list_scope_requests`` emits per row, so the frontend's detail drawer can render directly from the response without re-hydrating client-side. Powers the cockpit-style "open detail by ID" deep-link path on ``/admin/cloud/scope-requests`` — previously the page widened the list with ``page_size=200`` to find the orphan row, which 422'd at the BE ``le=100`` cap. A direct lookup is both correct and cheaper. """ req = await db.get(CmdbCloudScopeRequest, req_id) if req is None: raise NotFoundError("CmdbCloudScopeRequest", req_id) is_admin = can_manage_cloud(user.role_names) is_requester = str(req.requester_person_id) == str(user.id) if not (is_admin or is_requester): raise ForbiddenError() [data] = await _hydrate_scope_requests(db, [req]) return success_response(data=data, request=request) @router.get("/scope-requests/{req_id}/timeline") async def get_scope_request_timeline( req_id: str, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Per-record lifecycle timeline for one cloud scope-creation request. Sourced from the append-only ``audit_logs`` table (see :mod:`app.domains.cmdb.timeline_service` for the action whitelist). RBAC: * ``can_manage_cloud`` admins (global_admin / cloud_admin) see any request's timeline. * The original requester sees their own request's timeline. Other employees / the scope owner of the *resulting* scope do NOT get access — once the scope exists, owner visibility flows through the scope-detail page (grants on that scope are theirs to govern; the originating request's approval trail is between requester and IT). This keeps the read surface narrow on what is otherwise a record co-owning IT-internal approval context. """ req = await db.get(CmdbCloudScopeRequest, req_id) if req is None: raise NotFoundError("CmdbCloudScopeRequest", req_id) is_admin = can_manage_cloud(user.role_names) is_requester = str(req.requester_person_id) == str(user.id) if not (is_admin or is_requester): raise ForbiddenError() events = await fetch_scope_request_timeline(db, req_id) return success_response( data=[e.to_dict() for e in events], request=request, ) @router.get("/scopes/{scope_id}/access-ledger") async def get_scope_access_ledger( scope_id: uuid.UUID, request: Request, limit: int = Query(default=300, ge=1, le=500), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Scope-wide access ledger — what actually happened, newest first. Replaces the old 「申请记录」 tab, which re-rendered the members table's own grant rows sorted by ``granted_at``. That surface could not show a renewal at all (a renewal mutates an existing row and never touches ``granted_at``), labelled ``granted_at`` as 「申请时间」 when it is actually "when access became effective" and is re-stamped on every dispatch, and presented 218 of 230 cloud-discovered rows as if they were applications — while its own empty state claimed only submitted requests appear there. Read gate is the scope-READER superset, matching every other panel on the detail page so one tab does not 403 for a viewer who can see the rest. """ from app.domains.cmdb.timeline_service import list_scope_access_ledger await _require_scope_reader(db, user, str(scope_id)) grant_ids = [ str(g) for g in (await db.execute( select(CmdbCloudAccessGrant.id).where( CmdbCloudAccessGrant.scope_id == str(scope_id), ) )).scalars().all() ] rows = await list_scope_access_ledger( db, str(scope_id), grant_ids=grant_ids, limit=limit, ) return success_response(data=rows, request=request) @router.get("/grants/{grant_id}/timeline") async def get_grant_timeline( grant_id: uuid.UUID, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Per-record lifecycle timeline for one cloud access grant. Sourced from ``audit_logs`` via the whitelist in :mod:`app.domains.cmdb.timeline_service`. RBAC (matches the existing ``list_grants`` / ``get_scope`` read posture so the read surface stays coherent): * ``can_read_cloud`` — global_admin / auditor / cloud_admin / management — full read. * ``can_view_finance`` — finance / auditor / management — full read; finance ends up here when they pivot in from ``/admin/cloud/overview`` row click. * The grant holder (``grant.person_id == user.id``) — their own access record. * The scope owner — every grant on a scope they own (this is the surface that satisfies the owner's "show me who applied for access to my subscription" requirement). """ grant = await db.get(CmdbCloudAccessGrant, grant_id) if grant is None: raise NotFoundError("CmdbCloudAccessGrant", str(grant_id)) if not ( can_read_cloud(user.role_names) or can_view_finance(user.role_names) or (grant.person_id and str(grant.person_id) == str(user.id)) ): # Scope owner check — only run when none of the cheap gates hit. if not await is_cloud_scope_owner_of( db, str(user.id), str(grant.scope_id), ): raise ForbiddenError() events = await fetch_grant_timeline(db, str(grant_id)) return success_response( data=[e.to_dict() for e in events], request=request, ) # ── Role catalog promote/demote (admin) ──────────────────────────── class RolePromoteBody(BaseModel): role_ids: list[str] = Field(min_length=1, max_length=200) is_promoted: bool @router.post("/role-catalog/promote") async def promote_roles( body: RolePromoteBody, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Bulk flip the ``is_promoted`` flag on a set of role rows. Admin-only. Persisted across roles_sync runs (the sync handler deliberately skips ``is_promoted`` on UPDATE — see ``sync_roles_for_integration``). Row-scoped, so it leaves the (integration, role_key) group in a mixed state on purpose; no UI calls it (the catalog page uses the by-key variant). It still stamps ``promotion_overridden_by_admin`` so role sync stops seeding fresh rows for this key from the driver default. """ if not can_manage_cloud(user.role_names): raise ForbiddenError() rows = list((await db.execute( select(CmdbCloudRole).where( CmdbCloudRole.id.in_([uuid.UUID(rid) for rid in body.role_ids]), CmdbCloudRole.deleted_at.is_(None), ) )).scalars().all()) if not rows: return success_response(data={"updated": 0}, request=request) for r in rows: r.is_promoted = body.is_promoted r.promotion_overridden_by_admin = True audit = AuditService(db) trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) await audit.log( trace_id=trace_id, actor_type="user", actor_user_id=user.id, action=( "cloud.role_catalog.promote" if body.is_promoted else "cloud.role_catalog.demote" ), status="success", resource_type="cmdb_cloud_role", resource_id=",".join(str(r.id) for r in rows), detail={ "role_ids": [str(r.id) for r in rows], "is_promoted": body.is_promoted, }, ) await db.commit() return success_response( data={"updated": len(rows)}, request=request, ) class RoleGovernanceTierBody(BaseModel): role_ids: list[str] = Field(min_length=1, max_length=200) governance_tier: str @router.post("/role-catalog/governance-tier") async def set_role_governance_tier( body: RoleGovernanceTierBody, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Admin override of a role's governance tier (corrects a mis- classification). Sets ``governance_overridden_by_admin=True`` so the next role sync stops recomputing the tier (mirrors ``is_promoted``). Admin-only. See ``app/integrations/cloud_role_governance.py``. """ if not can_manage_cloud(user.role_names): raise ForbiddenError() from app.integrations.cloud_role_governance import ALL_TIERS tier = (body.governance_tier or "").strip() if tier not in ALL_TIERS: raise AppException( code="INVALID_GOVERNANCE_TIER", message=f"governance_tier must be one of {sorted(ALL_TIERS)}", status_code=422, ) rows = list((await db.execute( select(CmdbCloudRole).where( CmdbCloudRole.id.in_([uuid.UUID(rid) for rid in body.role_ids]), CmdbCloudRole.deleted_at.is_(None), ) )).scalars().all()) if not rows: return success_response(data={"updated": 0}, request=request) for r in rows: r.governance_tier = tier r.governance_reason = f"admin override by {user.id}" r.governance_overridden_by_admin = True audit = AuditService(db) trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) await audit.log( trace_id=trace_id, actor_type="user", actor_user_id=user.id, action="cloud.role_catalog.set_governance_tier", status="success", resource_type="cmdb_cloud_role", resource_id=",".join(str(r.id) for r in rows), detail={ "role_ids": [str(r.id) for r in rows], "governance_tier": tier, }, ) await db.commit() return success_response(data={"updated": len(rows)}, request=request) # ── Per-integration role catalog (admin) ─────────────────────────── @router.get("/role-catalog/by-integration") async def list_role_catalog_by_integration( request: Request, integration_id: str = Query(...), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Return the deduped role catalog for one cloud integration — one row per ``role_key`` regardless of how many scopes carry it. The current ``GET /scopes/{scope_id}/roles`` endpoint requires the admin to pick a scope first and shows ``~380`` rows per Azure subscription, with promote/demote acting only on that scope. For catalog management, that's the wrong unit — the admin's mental model is "this role should be primary in Azure", not "this role should be primary in *this specific* Azure subscription". This endpoint surfaces the integration-wide view, and the matching promote endpoint (``/role-catalog/promote-by-key``) flips every same-keyed row in one shot. Returns: { "integration_id": str, "provider": str, "default_roles_setting_key": str | None, "default_role_keys": [str, ...], # the curated list "roles": [ { "role_key": str, "display_name": str | None, "description": str | None, "is_custom": bool, "source": str | None, "is_promoted": bool, # consensus across scopes; # see fields below for # the per-scope shape "scope_count": int, # how many scopes carry it "promoted_scope_count": int, # how many of those are # currently promoted "promotion_mixed": bool, # promoted on SOME scopes — # 「否」 here is a lie; the # employee dropdown still # shows it on those scopes "promotion_overridden_by_admin": bool, "row_ids": [uuid, ...], # for the bulk flip path }, ... ] } """ if not can_manage_cloud(user.role_names): raise ForbiddenError() try: integration_uuid = uuid.UUID(integration_id) except ValueError as exc: raise NotFoundError("Integration", integration_id) from exc integration = await db.get(Integration, integration_uuid) if integration is None or integration.kind != "cloud": raise NotFoundError("Integration", integration_id) rows = list((await db.execute( select(CmdbCloudRole).where( CmdbCloudRole.integration_id == str(integration.id), CmdbCloudRole.deleted_at.is_(None), ) .order_by( CmdbCloudRole.is_custom.asc(), CmdbCloudRole.display_name.asc(), ) )).scalars().all()) # Map provider → platform_settings key holding the curated list, # mirrored from the migration. Used to render the "Reset to # default" button and the curated highlight in the UI. provider_key_map = { "azure": "cloud.azure.default_roles", "gcp": "cloud.gcp.default_roles", "aws": "cloud.aws_iam.default_policies", "aliyun": "cloud.aliyun.default_policies", "alicloud_intl": "cloud.aliyun.default_policies", "tencent": "cloud.tencent.default_policies", "ucloud": "cloud.ucloud.default_policies", } setting_key = provider_key_map.get(integration.provider) default_role_keys: list[str] = [] if setting_key: from app.integrations.service import get_platform_setting v = await get_platform_setting(setting_key, default=None, session=db) if isinstance(v, list): default_role_keys = [ str(x.get("role_key") or "") for x in v if isinstance(x, dict) and x.get("role_key") ] # Dedup by role_key across scopes. A role with mixed promote # state across scopes (rare but possible if an admin manually # demoted on one) is reported as ``is_promoted=False`` so the # "promote" action can flip every row consistently in the # one-button click. from app.integrations.cloud_role_governance import most_severe grouped: dict[str, dict[str, Any]] = {} for r in rows: bucket = grouped.setdefault(r.role_key, { "role_key": r.role_key, "display_name": r.display_name, # Effective (override-wins) description for display; the raw # synced description and the override are surfaced separately # so the edit UI can show the synced original + clear path. "description": r.description_override or r.description, "synced_description": r.description, "description_override": r.description_override, "is_custom": r.is_custom, "source": r.source, "row_ids": [], "promoted_scope_count": 0, "scope_count": 0, "sort_order": r.sort_order, "governance_tier": "normal", "governance_overridden_by_admin": False, "promotion_overridden_by_admin": False, }) bucket["row_ids"].append(str(r.id)) bucket["scope_count"] += 1 if r.is_promoted: bucket["promoted_scope_count"] += 1 # Report the most-severe tier across scopes (mixed state → the # higher risk wins, so the admin sees the cautious value). bucket["governance_tier"] = most_severe( bucket["governance_tier"], r.governance_tier or "normal", ) if r.governance_overridden_by_admin: bucket["governance_overridden_by_admin"] = True if r.promotion_overridden_by_admin: bucket["promotion_overridden_by_admin"] = True # sort_order / override are set consistently across scope rows via # update-by-key, but tolerate drift: lowest non-zero rank wins, # first present override wins. if r.sort_order and ( not bucket["sort_order"] or r.sort_order < bucket["sort_order"] ): bucket["sort_order"] = r.sort_order if not bucket["description_override"] and r.description_override: bucket["description_override"] = r.description_override bucket["description"] = r.description_override # Pick the most informative description across scopes. if not bucket["description"] and r.description: bucket["description"] = r.description if not bucket["synced_description"] and r.description: bucket["synced_description"] = r.description if not bucket["display_name"] and r.display_name: bucket["display_name"] = r.display_name out = [] for bucket in grouped.values(): bucket["is_promoted"] = ( bucket["promoted_scope_count"] == bucket["scope_count"] ) # Promoted on SOME scopes but not all. Reporting only the # consensus made this state indistinguishable from "demoted # everywhere" — and the switch, already sitting at 否, offered no # way to say "demote the stragglers too". That is how ten # ``DBaaS-Prod-*`` roles stayed in the employee dropdown on 8 of # 29 Aliyun sub-accounts for three months after being demoted. # The catalog page surfaces this as its own filter + a one-click # 「统一为否」; new drift is prevented at the source by # ``promotion_overridden_by_admin``. bucket["promotion_mixed"] = ( 0 < bucket["promoted_scope_count"] < bucket["scope_count"] ) out.append(bucket) out.sort(key=lambda b: ( # Manual rank first (0 = unranked, sorts after any explicit rank); # then today's promoted / custom / alphabetical tiebreakers. b["sort_order"] if b["sort_order"] else 10**9, not b["is_promoted"], # promoted first not b["is_custom"], # then customs (b["display_name"] or b["role_key"]).lower(), )) from app.domains.cmdb.sweep_settings import cloud_catalog_health return success_response( data={ "integration_id": str(integration.id), "provider": integration.provider, "default_roles_setting_key": setting_key, "default_role_keys": default_role_keys, "roles": out, # Is this list still being refreshed at all? A role catalog # fails silently in the worst direction — a role that exists # on the cloud is simply missing from the picker, and nobody # learns why. Prod: the Tencent roles cron was left disabled # after a single 2026-05-19 failure and the catalog sat # frozen for three months. "catalog_health": await cloud_catalog_health( db, str(integration.id), ), }, request=request, ) class RolePromoteByKeyBody(BaseModel): integration_id: str role_keys: list[str] = Field(min_length=1, max_length=400) is_promoted: bool @router.post("/role-catalog/promote-by-key") async def promote_roles_by_key( body: RolePromoteByKeyBody, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Flip ``is_promoted`` on every (integration, role_key) pair — a bulk version of ``/role-catalog/promote`` keyed by role_key instead of row id, so an admin can promote "Owner" once and have it apply across all 30 Azure subscriptions in one click. Also records the decision as ``promotion_overridden_by_admin`` so it survives the NEXT sub-account too: ``is_promoted`` lives per (scope, role_key) row, so before this flag a demote only reached the rows that existed when it was clicked and every sub-account discovered afterwards got a fresh row seeded from the driver default (custom ⇒ promoted). That is how ten demoted ``DBaaS-Prod-*`` roles were back in the employee dropdown on 8 of 29 Aliyun sub-accounts for three months (2026-06-11 → 2026-09-07) while this page showed 否. Soft-deleted rows are included: a role row revived by a later sync must come back carrying the decision, not the driver default. Admin-only. Audited identically to the row-id variant. """ if not can_manage_cloud(user.role_names): raise ForbiddenError() try: integration_uuid = uuid.UUID(body.integration_id) except ValueError as exc: raise NotFoundError("Integration", body.integration_id) from exc integration = await db.get(Integration, integration_uuid) if integration is None or integration.kind != "cloud": raise NotFoundError("Integration", body.integration_id) rows = list((await db.execute( select(CmdbCloudRole).where( CmdbCloudRole.integration_id == str(integration.id), CmdbCloudRole.role_key.in_(body.role_keys), ) )).scalars().all()) if not rows: return success_response(data={"updated": 0}, request=request) for r in rows: r.is_promoted = body.is_promoted r.promotion_overridden_by_admin = True audit = AuditService(db) trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) await audit.log( trace_id=trace_id, actor_type="user", actor_user_id=user.id, action=( "cloud.role_catalog.promote" if body.is_promoted else "cloud.role_catalog.demote" ), status="success", resource_type="cmdb_cloud_role", resource_id=str(integration.id), detail={ "integration_id": str(integration.id), "role_keys": body.role_keys, "is_promoted": body.is_promoted, "rows_affected": len(rows), }, ) await db.commit() return success_response( data={"updated": len(rows), "role_keys": body.role_keys}, request=request, ) class RoleCatalogUpdateByKeyBody(BaseModel): integration_id: str role_key: str # Both optional; only fields explicitly present in the request body # are applied (so a sort-order edit doesn't wipe the description and # vice-versa). ``description_override=""`` clears the override and # reverts to the provider-synced description. sort_order: int | None = None description_override: str | None = None @router.post("/role-catalog/update-by-key") async def update_role_catalog_by_key( body: RoleCatalogUpdateByKeyBody, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Set the admin-curated ``sort_order`` and/or ``description_override`` on every (integration, role_key) scope row in one shot — the catalog's mental model is "this role ranks here / reads like this in *this cloud*", not per-subscription. Mirrors ``/role-catalog/promote-by-key``. Only fields present in the body are applied. ``description_override`` set to an empty / whitespace string clears the override (reverts to the provider-synced ``description``). Admin-only, audited. """ if not can_manage_cloud(user.role_names): raise ForbiddenError() try: integration_uuid = uuid.UUID(body.integration_id) except ValueError as exc: raise NotFoundError("Integration", body.integration_id) from exc integration = await db.get(Integration, integration_uuid) if integration is None or integration.kind != "cloud": raise NotFoundError("Integration", body.integration_id) fields_set = body.model_fields_set set_sort = "sort_order" in fields_set set_desc = "description_override" in fields_set if not set_sort and not set_desc: return success_response(data={"updated": 0}, request=request) new_sort = max(0, int(body.sort_order or 0)) if set_sort else None new_override: str | None = None if set_desc: stripped = (body.description_override or "").strip() new_override = stripped or None # empty → clear, revert to synced rows = list((await db.execute( select(CmdbCloudRole).where( CmdbCloudRole.integration_id == str(integration.id), CmdbCloudRole.role_key == body.role_key, CmdbCloudRole.deleted_at.is_(None), ) )).scalars().all()) if not rows: return success_response(data={"updated": 0}, request=request) for r in rows: if set_sort: r.sort_order = new_sort if set_desc: r.description_override = new_override audit = AuditService(db) trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) detail: dict[str, Any] = { "integration_id": str(integration.id), "role_key": body.role_key, "rows_affected": len(rows), } if set_sort: detail["sort_order"] = new_sort if set_desc: detail["description_override"] = new_override await audit.log( trace_id=trace_id, actor_type="user", actor_user_id=user.id, action="cloud.role_catalog.update", status="success", resource_type="cmdb_cloud_role", resource_id=str(integration.id), detail=detail, ) await db.commit() return success_response( data={"updated": len(rows), "role_key": body.role_key}, request=request, ) @router.post("/role-catalog/reset-to-default") async def reset_role_catalog_to_default( request: Request, integration_id: str = Query(...), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Re-stamp ``is_promoted`` on every cmdb_cloud_role row for this integration based on: is_promoted = is_custom OR role_key ∈ platform_settings.cloud.. Same logic as migration 071's reconciliation pass. Useful when an admin's experiments diverged from the curated list and they want to start fresh — replaces "remember which roles I demoted individually" with one click. This is also the one place that CLEARS ``promotion_overridden_by_admin``: "reset to default" means "forget my promote/demote decisions", so role sync must go back to seeding new rows from the driver classification. **Soft-deleted rows are re-stamped too** — a single flagged soft-deleted row is enough for role sync to keep re-applying the decision this button just discarded, and it would come back carrying it when a later sync revives the row. """ if not can_manage_cloud(user.role_names): raise ForbiddenError() try: integration_uuid = uuid.UUID(integration_id) except ValueError as exc: raise NotFoundError("Integration", integration_id) from exc integration = await db.get(Integration, integration_uuid) if integration is None or integration.kind != "cloud": raise NotFoundError("Integration", integration_id) provider_key_map = { "azure": "cloud.azure.default_roles", "gcp": "cloud.gcp.default_roles", "aws": "cloud.aws_iam.default_policies", "aliyun": "cloud.aliyun.default_policies", "alicloud_intl": "cloud.aliyun.default_policies", "tencent": "cloud.tencent.default_policies", "ucloud": "cloud.ucloud.default_policies", } setting_key = provider_key_map.get(integration.provider) default_keys: set[str] = set() if setting_key: from app.integrations.service import get_platform_setting v = await get_platform_setting( setting_key, default=None, session=db, ) if isinstance(v, list): default_keys = { str(x.get("role_key") or "") for x in v if isinstance(x, dict) and x.get("role_key") } rows = list((await db.execute( select(CmdbCloudRole).where( CmdbCloudRole.integration_id == str(integration.id), ) )).scalars().all()) promoted = demoted = 0 for r in rows: target = bool(r.is_custom) or r.role_key in default_keys if r.is_promoted != target: r.is_promoted = target if target: promoted += 1 else: demoted += 1 # "Forget my edits" — otherwise the next role sync would re-stamp # every row back onto the decision this button just discarded. r.promotion_overridden_by_admin = False audit = AuditService(db) trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) await audit.log( trace_id=trace_id, actor_type="user", actor_user_id=user.id, action="cloud.role_catalog.reset", status="success", resource_type="cmdb_cloud_role", resource_id=str(integration.id), detail={ "integration_id": str(integration.id), "promoted": promoted, "demoted": demoted, "default_role_keys": sorted(default_keys), }, ) await db.commit() return success_response( data={ "promoted": promoted, "demoted": demoted, "default_role_keys": sorted(default_keys), }, request=request, ) # ── Orphan-assignment cleanup (cloud-side leaver residue) ───────── class OrphanCleanupTarget(BaseModel): """One row from the admin's confirmed cleanup list.""" scope_id: str role_key: str = Field(min_length=1, max_length=512) role_display_name: str | None = Field(default=None, max_length=255) principal_id: str = Field(min_length=1, max_length=255) external_ref: str | None = Field(default=None, max_length=512) recovered_email: str | None = Field(default=None, max_length=255) class OrphanCleanupRequest(BaseModel): integration_id: str targets: list[OrphanCleanupTarget] = Field(min_length=1, max_length=500) @router.get("/cloud-grants/orphan-scan") async def cloud_grant_orphan_scan( request: Request, integration_id: str, scope_id: str | None = None, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Scan one cloud integration for orphan principal assignments. ``scope_id`` (optional): scan ONLY that scope. The per-scope detail page's "扫描云端孤儿权限" button MUST pass this — without it the scan walks every active sub-account on the integration (an AWS Org with 50+ accounts × 20 permission sets explodes into tens of thousands of API calls and times out HTTP). Use the integration-wide form only from a deliberate "scan everything" batch tool. Returns the candidate list the admin reviews before confirming the cleanup. Read-only; no DB writes, no cloud-side mutations. Restricted to ``global_admin`` and ``cloud_admin`` via ``can_manage_cloud`` (2026-05-20: cloud_admin promoted from perpetual-grant marker to real cloud-domain write authority). """ if not can_manage_cloud(user.role_names): raise ForbiddenError() from app.domains.cmdb.orphan_cleanup import scan_orphan_assignments rows = await scan_orphan_assignments( integration_id, scope_id=scope_id, ) data = [ { "integration_id": r.integration_id, "integration_name": r.integration_name, "provider": r.provider, "scope_id": r.scope_id, "scope_external_id": r.scope_external_id, "scope_display_name": r.scope_display_name, "principal_id": r.principal_id, "role_key": r.role_key, "role_display_name": r.role_display_name, "external_ref": r.external_ref, "recovered_email": r.recovered_email, "recovered_display_name": r.recovered_display_name, } for r in rows ] return success_response(data=data, request=request) @router.post("/cloud-grants/orphan-cleanup", status_code=200) async def cloud_grant_orphan_cleanup( body: OrphanCleanupRequest, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), _gate: None = Depends(require_browser_session), ): """Execute cleanup for the admin-confirmed orphan list. Each target is processed independently — a single cloud-side failure does NOT abort the rest. The response reports per-target outcome so the admin can see which assignments still need attention. Every successful sweep writes a history row into ``cmdb_cloud_access_grants`` (visible in the scope detail's 已结束 tab) and a ``cloud.grant.orphan_principal_swept`` audit row. """ if not can_manage_cloud(user.role_names): raise ForbiddenError() from app.domains.cmdb.orphan_cleanup import cleanup_orphan_assignments trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) results = await cleanup_orphan_assignments( body.integration_id, [t.model_dump() for t in body.targets], actor_user_id=user.id, trace_id=trace_id, ) summary = { "requested": len(body.targets), "succeeded": sum(1 for r in results if r.ok), "already_absent": sum(1 for r in results if r.ok and r.already_absent), "failed": sum(1 for r in results if not r.ok), } return success_response( data={ "summary": summary, "results": [ { "principal_id": r.principal_id, "role_key": r.role_key, "scope_id": r.scope_id, "ok": r.ok, "already_absent": r.already_absent, "grant_id": r.grant_id, "error_message": r.error_message, } for r in results ], }, request=request, ) # ── Admin Cloud Overview (multi-tab dashboard) ────────────────────── @router.get("/overview/admin") async def get_admin_cloud_overview( request: Request, integration_id: str | None = None, month: str | None = Query( None, description=( "Optional billing month ``YYYY-MM`` for the per-subscription " "cost ranking. When omitted, the response carries cost rankings " "for the current UTC month (clamped to today). Past months show " "the settled total; current month is a partial MTD." ), pattern=r"^\d{4}-(0[1-9]|1[0-2])$", ), department: str | None = Query( None, description=( "Optional department (matches the whole subtree under the " "chosen node on the person's primary org membership — a parent " "rolls up every descendant) that narrows every owner/holder-" "attributable section to that department: subscriptions/IAM by " "owner, grants by holder, and the cost ranking by " "``cost_owner_department``. Composes AND with ``integration_id``." ), max_length=128, ), user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Admin-only multi-tab Cloud Overview payload. Fleet-wide aggregations across subscriptions (cloud sub-accounts, Azure subscriptions, GCP projects, RAM-member accounts, standalone IAM scopes), grants (active / pending / expiring / recovered), discovered IAM users, and governance alerts (unowned scopes, prod-overdue grants, pending grants). Optional ``integration_id`` narrows every section to a single cloud tenant. ``month`` (YYYY-MM) selects the billing month for the per-subscription cost ranking the Subscriptions tab shows; defaults to the current month. ``department`` narrows every owner/holder-attributable section to a single department. Gated by ``can_read_cloud`` (global_admin / cloud_admin / auditor / management) OR ``can_view_finance`` (global_admin / finance / auditor / management). Cloud admins were folded in 2026-05-20 when the role was promoted to a real authority; management was folded in the same day's role-matrix realignment (read-only leadership view). Auditor sees this surface for governance even though they also have ``/admin/audit`` — fleet aggregates are useful cross-referencing during reviews. Finance is read-only here; mutations on this surface are gated separately at each individual endpoint, not at the overview read. """ if not ( can_read_cloud(user.role_names) or can_view_finance(user.role_names) ): raise ForbiddenError() if integration_id: try: uuid.UUID(integration_id) except (ValueError, TypeError) as e: raise AppException( "VALIDATION_ERROR", "integration_id must be a UUID", status_code=422, ) from e # Confirm the integration exists and is a cloud-kind row; # otherwise an unknown id silently returns the empty fleet # response, which masks bad client state. integ = await db.get(Integration, uuid.UUID(integration_id)) # noqa: E501 (already validated above) if integ is None or integ.kind != "cloud": raise NotFoundError("CloudIntegration", integration_id) dept = department.strip() if department else None from app.domains.cmdb.admin_overview_service import get_admin_overview data = await get_admin_overview( db, integration_id=integration_id, month=month, department=dept or None, ) return success_response(data=data, request=request) @router.get("/overview/admin/department-cost") async def get_department_cloud_cost( request: Request, department: str = Query( ..., max_length=128, description=( "Department name (whole subtree under the chosen node rolls " "up) to report cloud spend for, both allocation and ownership " "口径." ), ), month: str | None = Query( None, pattern=r"^\d{4}-(0[1-9]|1[0-2])$", description="Billing month YYYY-MM; defaults to current UTC month.", ), integration_id: str | None = None, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Side-by-side dual-口径 cloud cost for one department + month. Answers "how much did spend on the cloud last month, by cloud account and by provider, and what's the total" — returning BOTH the allocation 口径 (spend attributed to the department's people via ``cost_owner_department``; shared accounts split) AND the ownership 口径 (whole spend of accounts the department currently owns), by provider, by account, and as totals, so the two reconcile in one place. Same RBAC as the admin overview (cloud-read OR finance). """ if not ( can_read_cloud(user.role_names) or can_view_finance(user.role_names) ): raise ForbiddenError() if integration_id: try: uuid.UUID(integration_id) except (ValueError, TypeError) as e: raise AppException( "VALIDATION_ERROR", "integration_id must be a UUID", status_code=422, ) from e dep = department.strip() if not dep: raise AppException( "VALIDATION_ERROR", "department must be non-empty", status_code=422, ) from app.domains.cmdb.admin_overview_service import ( department_cost_reconciliation, ) data = await department_cost_reconciliation( db, department=dep, month=month, integration_id=integration_id, ) return success_response(data=data, request=request) # ── Owner governance: confirmations timeline + respond actions ─── class OwnerConfirmationResponse(BaseModel): """Detail-page timeline entry for the "Owner 与治理记录" tab.""" # Stringly typed because the timeline is a union of two id spaces: # a confirmation-row PK for stored entries, an audit-log PK for # audit-derived ones. Both are UUIDs, but the service hands them # over already stringified and the client treats the field as an # opaque key — declaring ``uuid.UUID`` here only made pydantic # re-parse what it would serialize back to the identical string. id: str scope_id: str kind: str status: str owner_person_id: str | None = None owner_display_name: str | None = None # Leaver indicators for the owner only — responded_by / transferred_to # are NOT enriched here (no IdentityPerson object is in scope for them # beyond the display-name lookup, and they are out of scope for the # PersonTag owner-column swap). owner_employee_no: str | None = None owner_employment_status: str | None = None response_kind: str | None = None response_at: datetime | None = None responded_by_person_id: str | None = None responded_by_display_name: str | None = None transferred_to_person_id: str | None = None transferred_to_display_name: str | None = None reason: str | None = None spend_recent_30d_usd: Decimal | None = None spend_recent_180d_usd: Decimal | None = None spend_last_month_usd: Decimal | None = None member_count_snapshot: int | None = None notified_at: datetime | None = None escalated_at: datetime | None = None parent_confirmation_id: str | None = None created_at: datetime # True when the entry was synthesized from ``audit_logs`` rather # than read from ``cmdb_cloud_scope_owner_confirmations``. Derived # entries are owner-CHANGE facts (initial auto-claim, offboarding # force-transfer) that never produced a confirmation row; they are # read-only — there is nothing to respond to — and the frontend # must not render response buttons on them. derived: bool = False # Enriched successor label for derived rows (and for stored rows # that carry a transferee), so the timeline can say who took over # without the client resolving person ids. transferee_employee_no: str | None = None transferee_employment_status: str | None = None # An unanswered prompt is re-sent to its owner once a day until they # answer (IT escalation retired 2026-08-24). Surfacing the count is # what makes "this has been asked 14 times and still nobody # answered" visible without reading the audit log. reminder_count: int = 0 last_reminded_at: datetime | None = None class ZeroSpendRespondBody(BaseModel): action: str = Field(..., description="keep | cleanup | handover") reason: str | None = Field( default=None, max_length=4000, description="Required when action='keep'.", ) @router.get("/scopes/{scope_id}/owner-confirmations") async def list_owner_confirmations( scope_id: uuid.UUID, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """List every governance event on a scope, newest first. Visible to scope owner, cloud-admin, and finance readers (2026-05-16). The detail-page tab "Owner 与治理记录" calls this; the MCP tool ``cmdb_cloud_scope_list_owner_confirmations`` calls this same path. The owner-respond endpoints below remain owner/admin-only — finance is read-only on governance. """ await _require_scope_reader(db, user, str(scope_id)) from sqlalchemy import select as _select from app.domains.cmdb.owner_confirmation_service import ( OwnerConfirmationService, ) # Timeline, not just the stored rows: owner-change events that only # ever landed in ``audit_logs`` (initial auto-claim, offboarding # force-transfer) are merged in read-only so the owner history has # no holes. MCP's ``list_cloud_scope_owner_confirmations`` calls the # same service method. rows = await OwnerConfirmationService(db).list_timeline_for_scope( scope_id, limit=200, ) # Bulk-resolve display names for owner / responder / transferee # in one query rather than N round-trips. person_ids: set[str] = set() for r in rows: if r.owner_person_id: person_ids.add(str(r.owner_person_id)) if r.responded_by_person_id: person_ids.add(str(r.responded_by_person_id)) if r.transferred_to_person_id: person_ids.add(str(r.transferred_to_person_id)) name_by_id: dict[str, str] = {} # Leaver enrichment (employee_no / employment_status) for the owner # AND the transferee, so the frontend can render both through # with the 已离职 treatment. The transferee matters as # much as the owner on this timeline: a handover TO someone who has # since left is exactly the drift a reader is looking for. # ``responded_by`` keeps name-only. employee_no_by_id: dict[str, str | None] = {} employment_status_by_id: dict[str, str | None] = {} if person_ids: pid_list = [uuid.UUID(p) for p in person_ids] res = await db.execute( _select( IdentityPerson.id, IdentityPerson.display_name, IdentityPerson.work_email, IdentityPerson.employee_no, IdentityPerson.employment_status, ) .where(IdentityPerson.id.in_(pid_list)) ) for pid, dn, em, emp_no, emp_status in res.all(): name_by_id[str(pid)] = dn or em or "" employee_no_by_id[str(pid)] = emp_no employment_status_by_id[str(pid)] = emp_status def _name(pid: str | None) -> str | None: return name_by_id.get(str(pid)) if pid else None data = [ OwnerConfirmationResponse( id=r.id, scope_id=str(r.scope_id), kind=r.kind, status=r.status, owner_person_id=r.owner_person_id, owner_display_name=_name(r.owner_person_id), owner_employee_no=( employee_no_by_id.get(str(r.owner_person_id)) if r.owner_person_id else None ), owner_employment_status=( employment_status_by_id.get(str(r.owner_person_id)) if r.owner_person_id else None ), response_kind=r.response_kind, response_at=r.response_at, responded_by_person_id=r.responded_by_person_id, responded_by_display_name=_name(r.responded_by_person_id), transferred_to_person_id=r.transferred_to_person_id, transferred_to_display_name=_name(r.transferred_to_person_id), reason=r.reason, spend_recent_30d_usd=r.spend_recent_30d_usd, spend_recent_180d_usd=r.spend_recent_180d_usd, spend_last_month_usd=r.spend_last_month_usd, member_count_snapshot=r.member_count_snapshot, notified_at=r.notified_at, escalated_at=r.escalated_at, parent_confirmation_id=r.parent_confirmation_id, created_at=r.created_at, derived=r.derived, transferee_employee_no=( employee_no_by_id.get(str(r.transferred_to_person_id)) if r.transferred_to_person_id else None ), transferee_employment_status=( employment_status_by_id.get(str(r.transferred_to_person_id)) if r.transferred_to_person_id else None ), reminder_count=r.reminder_count, last_reminded_at=r.last_reminded_at, ).model_dump(mode="json") for r in rows ] return success_response(data=data, request=request) @router.post( "/scopes/{scope_id}/owner-confirmations/{confirmation_id}/semiannual-confirm", ) async def respond_semiannual_confirm( scope_id: uuid.UUID, confirmation_id: uuid.UUID, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Owner clicks ✅ on the semiannual confirmation card. Only the current scope owner OR a cloud admin may confirm — peers cannot confirm on someone else's behalf (the 180-day clock is personal accountability, not delegable). """ await _require_scope_owner_or_admin(db, user, str(scope_id)) # ``AuthUser`` IS an :class:`IdentityPerson` (see auth/models.py # alias) — ``user.id`` is the person id. No separate person link # to validate. Direct attribute access (not getattr) so a future # refactor surfaces as a clean error. from app.domains.cmdb.owner_confirmation_service import ( OwnerConfirmationService, ) trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) row = await OwnerConfirmationService(db).respond_semiannual_confirm( scope_id, confirmation_id, responder_person_id=str(user.id), actor_user_id=user.id, trace_id=trace_id, actor_is_admin=can_manage_cloud(user.role_names), ) await db.commit() return success_response( data={ "id": str(row.id), "status": row.status, "response_kind": row.response_kind, }, request=request, ) @router.post( "/scopes/{scope_id}/owner-confirmations/{confirmation_id}/zero-spend-respond", ) async def respond_zero_spend( scope_id: uuid.UUID, confirmation_id: uuid.UUID, body: ZeroSpendRespondBody, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Owner picks keep / cleanup / handover on the zero-spend card.""" await _require_scope_owner_or_admin(db, user, str(scope_id)) # ``AuthUser`` IS an :class:`IdentityPerson` (see auth/models.py # alias) — ``user.id`` is the person id. No separate person link # to validate. Direct attribute access (not getattr) so a future # refactor surfaces as a clean error. from app.domains.cmdb.owner_confirmation_service import ( OwnerConfirmationService, ) trace_id = getattr(request.state, "trace_id", str(uuid.uuid4())) row = await OwnerConfirmationService(db).respond_zero_spend( scope_id, confirmation_id, action=body.action, reason=body.reason, responder_person_id=str(user.id), actor_user_id=user.id, trace_id=trace_id, actor_is_admin=can_manage_cloud(user.role_names), ) await db.commit() return success_response( data={ "id": str(row.id), "status": row.status, "response_kind": row.response_kind, "reason": row.reason, }, request=request, ) __all__ = ["router"]