"""Read-only service boundary for Helix data-access management V1.""" from __future__ import annotations import json import logging import re import uuid from collections.abc import Sequence from dataclasses import dataclass from datetime import UTC, datetime, timedelta from types import SimpleNamespace from typing import Any, Final from urllib.parse import urlparse from sqlalchemy import func, select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from app.approval.feishu_client import FeishuApprovalClient from app.approval.models import ApprovalRequest, ApprovalSubmissionIntent from app.approval.submission_intents import ( compact_terminal_intent_payload, defer_submission_intent_retry, reconciliation_due_predicate, reconciliation_order, submit_with_durable_intent, ) from app.audit.models import AuditLog from app.audit.service import AuditService from app.auth.models import AuthRole, AuthUserRole from app.core.exceptions import AppException from app.core.rbac import GLOBAL_ADMIN_ROLE, has_role, normalize_role_names from app.core.safe_logging import sanitize_log_text from app.domains.data_access import constants as c from app.domains.data_access.approval_flow import ( DataAccessApprovalFlow, resolve_data_access_group_grant_flow, resolve_data_access_personal_grant_flow, resolve_data_access_service_account_grant_flow, ) from app.domains.data_access.binding_projection import provider_member_for_binding from app.domains.data_access.models import ( BINDING_CURRENT_STATUSES, BINDING_STATUS_ACTIVE, BINDING_STATUS_REVOKED, BINDING_STATUS_REVOKED_PENDING, EFFECT_DESIRED_ABSENT, EFFECT_DESIRED_PRESENT, EFFECT_MANAGED_REASON_GRANT_REQUEST, EFFECT_MANAGED_REASON_GROUP_RENAME_BRIDGE, EFFECT_MANAGED_REASON_REVOKE_REQUEST, EFFECT_STATUS_PENDING_RECONCILE, ENTITLEMENT_STATUS_ACTIVE, ENTITLEMENT_VERSION_V1, OWNER_RESOLUTION_RESOLVED, OWNER_SOURCE_BOOTSTRAP, OWNER_SOURCE_BOOTSTRAP_TRANSFERRED, OWNER_SOURCE_CATALOG, OWNER_SOURCE_CATALOG_TRANSFERRED, OWNER_SOURCE_DATASET_INHERITED, OWNER_SOURCE_MANUAL, PRINCIPAL_STATUS_ACTIVE, RECONCILE_STATUS_PENDING, REQUEST_ITEM_STATUS_APPROVED, REQUEST_ITEM_STATUS_PENDING, REQUEST_ITEM_STATUS_REJECTED, REQUEST_ITEM_STATUS_STALE, REQUEST_STATUS_APPROVED, REQUEST_STATUS_CANCELLED, REQUEST_STATUS_PENDING_FEISHU, REQUEST_STATUS_REJECTED, REQUEST_STATUS_STALE_REQUIRES_RESUBMIT, RESOURCE_STATUS_ACTIVE, RESOURCE_STATUS_DELETED, DataAccessBinding, DataAccessCompiledEffect, DataAccessEntitlement, DataAccessPrincipal, DataAccessRequest, DataAccessRequestItem, DataAccessResource, DataAccessServiceAccountProfile, ) from app.domains.data_access.schemas import ( DataAccessAuthorityBoundary, DataAccessCapabilityResponse, DataAccessProviderEffect, DataAccessV1Contract, ) from app.domains.email_group.models import EmailGroup, EmailGroupOwner from app.domains.email_group.service import EmailGroupService from app.identity.person_models import IdentityPerson from app.integrations.compat import get_feishu_config from app.integrations.driver_base import DataAccessDriver logger = logging.getLogger(__name__) DATA_ACCESS_V1_NON_GOALS: Final = ( "on_behalf_personal_access", "arbitrary_person_email_input", "bigquery_project_iam", "bigquery_job_user", "broad_bigquery_dataset_scan", "dataset_creation", "visible_manual_dataset_catalog_ui", "terraform_runtime_apply", "helix_feishu_template_authoring", "google_group_membership_reconciliation", ) DATA_ACCESS_COMPILER_VERSION: Final = "dax-compiler-v2" DATA_ACCESS_FEISHU_PAYLOAD_VERSION: Final = "dax-feishu-payload-v4" DATA_ACCESS_BUSINESS_REASON_MAX_LENGTH: Final = 2000 DATA_ACCESS_DETAIL_URL_MAX_LENGTH: Final = 2048 DATA_ACCESS_IDEMPOTENCY_KEY_MAX_LENGTH: Final = 128 DATA_ACCESS_TRACE_ID_MAX_LENGTH: Final = 128 DATA_ACCESS_SERVICE_ACCOUNT_EMAIL_MAX_LENGTH: Final = 255 DATA_ACCESS_FEISHU_DATASET_DESCRIPTION_MAX_LENGTH: Final = 500 DATA_ACCESS_AUDIT_DETAIL_SOFT_LIMIT_BYTES: Final = 12_000 DATA_ACCESS_STALENESS_REASON_RESOURCE_CONTEXT_CHANGED: Final = "resource_context_changed" DATA_ACCESS_STALENESS_REASON_RESOURCE_CATALOG_STALE: Final = "resource_catalog_version_stale" DATA_ACCESS_RESOURCE_REMOVAL_REASON_PROVIDER_DELETED: Final = "resource_provider_confirmed_deleted" DATA_ACCESS_RESOURCE_REMOVAL_REASON_ADMIN_REMOVED: Final = "resource_admin_confirmed_removed" TRANSFERRED_DATASET_OWNER_SOURCE_BY_SOURCE: Final = { OWNER_SOURCE_CATALOG: OWNER_SOURCE_CATALOG_TRANSFERRED, OWNER_SOURCE_CATALOG_TRANSFERRED: OWNER_SOURCE_CATALOG_TRANSFERRED, OWNER_SOURCE_BOOTSTRAP: OWNER_SOURCE_BOOTSTRAP_TRANSFERRED, OWNER_SOURCE_BOOTSTRAP_TRANSFERRED: OWNER_SOURCE_BOOTSTRAP_TRANSFERRED, } DATA_ACCESS_FEISHU_FORM_FIELD_ORDER: Final = ( "request_summary", "requested_access", "dataset_list", "business_reason", "detail_url", ) DATA_ACCESS_FEISHU_FORM_DEFAULT_TYPES: Final = { "request_summary": "textarea", "requested_access": "textarea", "dataset_list": "textarea", "business_reason": "textarea", "detail_url": "input", } @dataclass(frozen=True, slots=True) class ValidatedRequestableDatasetResource: """Snapshot of a resource that is safe to use for DAX request intent.""" resource_id: uuid.UUID provider: str resource_type: str project_id: str dataset_id: str owner_person_id: str | None owner_resolution_status: str owner_authority_enabled: bool source_version: str source_etag: str description: str | None = None table_id: str | None = None table_type: str | None = None owner_source: str | None = None @dataclass(frozen=True, slots=True) class DataAccessOffboardingOwnerHandoverPlan: """Locked current-owner rows that must follow a leaver's handover.""" resources: tuple[DataAccessResource, ...] @property def required(self) -> bool: return bool(self.resources) @property def dataset_count(self) -> int: return sum( row.resource_type == c.RESOURCE_TYPE_DATASET for row in self.resources ) @property def table_count(self) -> int: return sum( row.resource_type == c.RESOURCE_TYPE_TABLE for row in self.resources ) @dataclass(frozen=True, slots=True) class ValidatedGoogleGroupPrincipal: """Snapshot of the Google Group principal DAX can reference.""" principal_email_group_id: uuid.UUID requested_email_group_id: uuid.UUID | None google_group_id: str primary_email_snapshot: str provider_member: str resolved_via: str principal_id: uuid.UUID | None = None @dataclass(frozen=True, slots=True) class ValidatedServiceAccountPrincipal: """A service-account snapshot recorded from the self-service request.""" principal_id: uuid.UUID subject_snapshot: str email_snapshot: str source_version: str source_etag: str @dataclass(frozen=True, slots=True) class ValidatedPersonalPrincipal: """A personal Google account derived only from the authenticated employee.""" principal_id: uuid.UUID subject_snapshot: str email_snapshot: str source_version: str source_etag: str @dataclass(frozen=True, slots=True) class DataAccessRequesterAuthorization: """Requester authorization result for a validated DAX principal.""" requester_person_id: uuid.UUID principal_email_group_id: uuid.UUID authorized_as: str @dataclass(frozen=True, slots=True) class DataAccessDryRunCompiledEffect: """Provider effect preview compiled from validated semantic DAX state.""" resource_id: uuid.UUID principal_email_group_id: uuid.UUID | None principal_google_group_id_snapshot: str | None principal_email_snapshot: str entitlement_id: uuid.UUID provider: str resource_type: str target_project_id: str target_dataset_id: str target_table_id: str | None provider_member: str provider_role: str desired_state: str entitlement_version_snapshot: str resource_source_version_snapshot: str resource_source_etag_snapshot: str compiler_version: str managed_reason: str already_present_at_first_apply: bool principal_id: uuid.UUID | None = None principal_type_snapshot: str = c.PRINCIPAL_TYPE_GOOGLE_GROUP principal_subject_snapshot: str | None = None provider_access_kind: str = "groupByEmail" @dataclass(frozen=True, slots=True) class DataAccessFeishuDatasetContext: """Dataset context shown to Feishu approvers before approval.""" resource_id: uuid.UUID provider: str resource_type: str project_id: str dataset_id: str owner_person_id: str | None owner_resolution_status: str owner_authority_enabled: bool source_version: str source_etag: str description: str | None = None table_id: str | None = None table_type: str | None = None def as_payload(self) -> dict[str, Any]: payload = { "resource_id": str(self.resource_id), "provider": self.provider, "resource_type": self.resource_type, "project_id": self.project_id, "dataset_id": self.dataset_id, "owner_person_id": self.owner_person_id, "owner_resolution_status": self.owner_resolution_status, "owner_authority_enabled": self.owner_authority_enabled, "source_version": self.source_version, "source_etag": self.source_etag, "description": self.description, } if self.table_id is not None: payload["table_id"] = self.table_id payload["table_type"] = self.table_type return payload @dataclass(frozen=True, slots=True) class DataAccessFeishuCompiledPreview: """Provider preview embedded in the Feishu submission payload.""" resource_id: uuid.UUID target_project_id: str target_dataset_id: str provider_member: str provider_role: str desired_state: str entitlement_version_snapshot: str resource_source_version_snapshot: str resource_source_etag_snapshot: str compiler_version: str managed_reason: str already_present_at_first_apply: bool target_table_id: str | None = None def as_payload(self) -> dict[str, Any]: payload = { "resource_id": str(self.resource_id), "target_project_id": self.target_project_id, "target_dataset_id": self.target_dataset_id, "provider_member": self.provider_member, "provider_role": self.provider_role, "desired_state": self.desired_state, "entitlement_version_snapshot": self.entitlement_version_snapshot, "resource_source_version_snapshot": (self.resource_source_version_snapshot), "resource_source_etag_snapshot": self.resource_source_etag_snapshot, "compiler_version": self.compiler_version, "managed_reason": self.managed_reason, "already_present_at_first_apply": (self.already_present_at_first_apply), } if self.target_table_id is not None: payload["target_table_id"] = self.target_table_id return payload @dataclass(frozen=True, slots=True) class DataAccessFeishuSubmissionPayload: """Canonical data-access payload for a future Feishu instance create.""" flow_key: str external_code: str payload_version: str business_reason: str detail_url: str principal_email_group_id: uuid.UUID principal_google_group_id_snapshot: str principal_email_snapshot: str entitlement_id: uuid.UUID entitlement_version: str dataset_count: int datasets: tuple[DataAccessFeishuDatasetContext, ...] compiled_preview: tuple[DataAccessFeishuCompiledPreview, ...] def semantic_intent(self) -> dict[str, Any]: return { "domain": c.DOMAIN_DATA_ACCESS, "provider": c.PROVIDER_BIGQUERY, "resource_type": _payload_resource_type(self.datasets), "principal_type": c.PRINCIPAL_TYPE_GOOGLE_GROUP, "entitlement": c.ENTITLEMENT_READ, "approval_flow": self.flow_key, "principal_email_group_id": str(self.principal_email_group_id), "principal_google_group_id_snapshot": (self.principal_google_group_id_snapshot), "principal_email_snapshot": self.principal_email_snapshot, "entitlement_id": str(self.entitlement_id), "entitlement_version": self.entitlement_version, } def to_audit_fields(self) -> dict[str, Any]: return { "payload_version": self.payload_version, "semantic_intent": self.semantic_intent(), "dataset_count": self.dataset_count, "business_reason": self.business_reason, "detail_url": self.detail_url, "dataset_owner_context": [dataset.as_payload() for dataset in self.datasets], "compiled_preview": [preview.as_payload() for preview in self.compiled_preview], } def to_submission_fields(self) -> dict[str, Any]: return { "request_summary": _feishu_request_summary(self), "requested_access": _feishu_requested_access(self), "dataset_list": _feishu_dataset_list(self), "business_reason": self.business_reason, "detail_url": self.detail_url, } @dataclass(frozen=True, slots=True) class DataAccessServiceAccountFeishuSubmissionPayload: """Feishu/audit projection for a typed GCP service-account request.""" flow_key: str external_code: str payload_version: str business_reason: str detail_url: str principal: ValidatedServiceAccountPrincipal entitlement_id: uuid.UUID entitlement_version: str datasets: tuple[DataAccessFeishuDatasetContext, ...] compiled_preview: tuple[DataAccessFeishuCompiledPreview, ...] @property def dataset_count(self) -> int: return len(self.datasets) def semantic_intent(self) -> dict[str, Any]: return { "domain": c.DOMAIN_DATA_ACCESS, "provider": c.PROVIDER_BIGQUERY, "resource_type": _payload_resource_type(self.datasets), "principal_type": c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT, "approval_flow": self.flow_key, "principal_id": str(self.principal.principal_id), "principal_subject_snapshot": self.principal.subject_snapshot, "principal_email_snapshot": self.principal.email_snapshot, "source_version": self.principal.source_version, "source_etag": self.principal.source_etag, "entitlement": c.ENTITLEMENT_READ, "entitlement_id": str(self.entitlement_id), "entitlement_version": self.entitlement_version, } def to_audit_fields(self) -> dict[str, Any]: return { "payload_version": self.payload_version, "semantic_intent": self.semantic_intent(), "dataset_count": self.dataset_count, "business_reason": self.business_reason, "detail_url": self.detail_url, "dataset_owner_context": [item.as_payload() for item in self.datasets], "compiled_preview": [item.as_payload() for item in self.compiled_preview], } def to_submission_fields(self) -> dict[str, Any]: locator_lines = [ f"{index}. {_resource_locator(dataset)}" for index, dataset in enumerate(self.datasets, start=1) ] return { "request_summary": ( f"GCP service account {self.principal.email_snapshot} is requesting " f"read access to {self.dataset_count} BigQuery resources." ), "requested_access": ( "Grant BigQuery data viewer to GCP service account " f"{self.principal.email_snapshot}." ), "dataset_list": "\n".join(locator_lines), "business_reason": self.business_reason, "detail_url": self.detail_url, } @dataclass(frozen=True, slots=True) class DataAccessPersonalFeishuSubmissionPayload: """Feishu/audit projection for a direct employee dataset-read request.""" flow_key: str external_code: str payload_version: str business_reason: str detail_url: str principal: ValidatedPersonalPrincipal entitlement_id: uuid.UUID entitlement_version: str datasets: tuple[DataAccessFeishuDatasetContext, ...] compiled_preview: tuple[DataAccessFeishuCompiledPreview, ...] @property def dataset_count(self) -> int: return len(self.datasets) def semantic_intent(self) -> dict[str, Any]: return { "domain": c.DOMAIN_DATA_ACCESS, "provider": c.PROVIDER_BIGQUERY, "resource_type": _payload_resource_type(self.datasets), "principal_type": c.PRINCIPAL_TYPE_GOOGLE_USER, "approval_flow": self.flow_key, "principal_id": str(self.principal.principal_id), "principal_subject_snapshot": self.principal.subject_snapshot, "principal_email_snapshot": self.principal.email_snapshot, "source_version": self.principal.source_version, "source_etag": self.principal.source_etag, "entitlement": c.ENTITLEMENT_READ, "entitlement_id": str(self.entitlement_id), "entitlement_version": self.entitlement_version, } def to_audit_fields(self) -> dict[str, Any]: return { "payload_version": self.payload_version, "semantic_intent": self.semantic_intent(), "dataset_count": self.dataset_count, "business_reason": self.business_reason, "detail_url": self.detail_url, "dataset_owner_context": [item.as_payload() for item in self.datasets], "compiled_preview": [item.as_payload() for item in self.compiled_preview], } def to_submission_fields(self) -> dict[str, Any]: locator_lines = [ f"{index}. {_resource_locator(dataset)}" for index, dataset in enumerate(self.datasets, start=1) ] return { "request_summary": ( f"Personal work account {self.principal.email_snapshot} is requesting " f"read access to {self.dataset_count} BigQuery resources." ), "requested_access": ( "Grant BigQuery data viewer to personal work account " f"{self.principal.email_snapshot}." ), "dataset_list": "\n".join(locator_lines), "business_reason": self.business_reason, "detail_url": self.detail_url, } @dataclass(frozen=True, slots=True) class DataAccessRequestSubmissionResult: """Result of creating or replaying a DAX Feishu-backed submission.""" request_id: uuid.UUID approval_request_id: uuid.UUID | None feishu_instance_code: str | None reused_existing: bool status: str idempotency_key: str | None trace_id: str | None dataset_count: int @dataclass(frozen=True, slots=True) class DataAccessRequestCallbackResult: """Result of applying a Feishu callback to DAX durable state.""" request_id: uuid.UUID | None approval_request_id: uuid.UUID approval_status: str request_status: str | None binding_count: int effect_count: int refusal_reason: str | None = None @dataclass(frozen=True, slots=True) class DataAccessRevokeResult: """Outcome of creating semantic revoke or rename-bridge intent.""" status: str reason: str | None authorized_as: str revoked_pending_count: int = 0 renamed_binding_count: int = 0 effect_count: int = 0 already_pending_count: int = 0 already_revoked_count: int = 0 class DataAccessResourceValidationError(AppException): """Raised when a resource cannot enter the DAX request path.""" def __init__( self, reason: str, message: str, *, resource_id: uuid.UUID | None = None, ) -> None: self.reason = reason self.resource_id = resource_id super().__init__( code=f"DATA_ACCESS_{reason.upper()}", message=message, status_code=422, field="resource_id", ) class DataAccessPrincipalValidationError(AppException): """Raised when a principal cannot enter the DAX request path.""" def __init__( self, reason: str, message: str, *, principal: str | uuid.UUID | None = None, ) -> None: self.reason = reason self.principal = principal super().__init__( code=f"DATA_ACCESS_{reason.upper()}", message=message, status_code=422, field="principal", ) class DataAccessRequesterAuthorizationError(AppException): """Raised when a requester cannot submit DAX intent for a principal.""" def __init__( self, message: str = ( "Requester must be a current Google Group owner, group member, or global admin." ), *, requester_person_id: uuid.UUID | None = None, principal_email_group_id: uuid.UUID | None = None, ) -> None: self.requester_person_id = requester_person_id self.principal_email_group_id = principal_email_group_id super().__init__( code="DATA_ACCESS_REQUESTER_NOT_AUTHORIZED", message=message, status_code=403, field="principal", ) class DataAccessRevokeAuthorizationError(AppException): """Raised when an actor cannot revoke a DAX binding.""" def __init__( self, *, actor_person_id: uuid.UUID, binding_id: uuid.UUID, ) -> None: self.actor_person_id = actor_person_id self.binding_id = binding_id super().__init__( code="DATA_ACCESS_REVOKE_NOT_AUTHORIZED", message=( "Actor must be a current Google Group owner, resource owner, " "or global admin to revoke this data-access binding." ), status_code=403, field="binding_id", ) class DataAccessCompilerError(AppException): """Raised when semantic DAX state cannot compile to a V1 effect.""" def __init__( self, reason: str, message: str, *, field: str = "compiler", ) -> None: self.reason = reason super().__init__( code=f"DATA_ACCESS_COMPILER_{reason.upper()}", message=message, status_code=422, field=field, ) class DataAccessFeishuPayloadError(AppException): """Raised when DAX state cannot be rendered for Feishu submission.""" def __init__( self, reason: str, message: str, *, field: str = "feishu_payload", ) -> None: self.reason = reason super().__init__( code=f"DATA_ACCESS_FEISHU_PAYLOAD_{reason.upper()}", message=message, status_code=422, field=field, ) class DataAccessSubmissionError(AppException): """Raised when DAX request submission fails before durable state is safe.""" def __init__( self, reason: str, message: str, *, field: str = "submission", status_code: int = 422, ) -> None: self.reason = reason super().__init__( code=f"DATA_ACCESS_SUBMISSION_{reason.upper()}", message=message, status_code=status_code, field=field, ) class DataAccessService: """Expose the DAX V1 contract without touching provider state.""" def get_capabilities(self) -> DataAccessCapabilityResponse: return DataAccessCapabilityResponse( status="router_service_schema_skeleton", provider_mutation_enabled=False, contract=DataAccessV1Contract( domain=c.DOMAIN_DATA_ACCESS, providers=c.ALL_PROVIDERS, resource_types=c.ALL_RESOURCE_TYPES, principal_types=c.ALL_PRINCIPAL_TYPES, entitlements=c.ALL_ENTITLEMENTS, approval_flows=c.ALL_APPROVAL_FLOWS, provider_effect=DataAccessProviderEffect( member_prefix=c.BIGQUERY_GROUP_MEMBER_PREFIX, role=c.BIGQUERY_DATA_VIEWER_ROLE, ), ), authority_boundary=DataAccessAuthorityBoundary( semantic_principal_field=c.SEMANTIC_PRINCIPAL_FIELD, provider_member_email_field=c.PROVIDER_MEMBER_EMAIL_FIELD, runtime_owner_authority_field=c.RUNTIME_OWNER_AUTHORITY_FIELD, approval_transport_owner="approval", provider_mutation_owner="data_access_reconciler", request_handlers_mutate_provider=False, feishu_callbacks_mutate_provider=False, ), non_goals=DATA_ACCESS_V1_NON_GOALS, ) async def validate_requestable_bigquery_dataset_resource( self, db: AsyncSession, resource_id: uuid.UUID, *, require_resource_owner_authority: bool = False, ) -> ValidatedRequestableDatasetResource: resource = await db.scalar( select(DataAccessResource).where(DataAccessResource.id == resource_id), ) if resource is None: raise DataAccessResourceValidationError( "non_federated_resource", "Resource is not available in the data-access catalog.", resource_id=resource_id, ) if resource.provider != c.PROVIDER_BIGQUERY: raise DataAccessResourceValidationError( "unsupported_provider", "Only BigQuery resources are requestable through data-access V1.", resource_id=resource_id, ) if resource.resource_type not in { c.RESOURCE_TYPE_DATASET, c.RESOURCE_TYPE_TABLE, }: raise DataAccessResourceValidationError( "unsupported_resource_type", "Only BigQuery datasets and tables are requestable through data access.", resource_id=resource_id, ) if resource.resource_type == c.RESOURCE_TYPE_TABLE and ( resource.parent_resource_id is None or not resource.table_id ): raise DataAccessResourceValidationError( "unsupported_resource_type", "Table resources must be materialized below a requestable dataset.", resource_id=resource_id, ) if resource.status != RESOURCE_STATUS_ACTIVE: raise DataAccessResourceValidationError( "inactive_resource", "Inactive data-access resources cannot be requested.", resource_id=resource_id, ) if not resource.requestable: raise DataAccessResourceValidationError( "resource_not_requestable", "This data-access resource is not marked requestable.", resource_id=resource_id, ) if require_resource_owner_authority and not resource.resource_owner_authority_enabled: raise DataAccessResourceValidationError( "resource_owner_authority_unavailable", "Resource owner approval is unavailable for this resource.", resource_id=resource_id, ) return ValidatedRequestableDatasetResource( resource_id=resource.id, provider=resource.provider, resource_type=resource.resource_type, project_id=resource.project_id, dataset_id=resource.dataset_id, owner_person_id=resource.owner_person_id, owner_resolution_status=resource.owner_resolution_status, owner_authority_enabled=resource.resource_owner_authority_enabled, source_version=resource.source_version, source_etag=resource.source_etag, description=_clean_feishu_dataset_description(resource.notes), table_id=resource.table_id, table_type=resource.table_type, owner_source=resource.owner_source, ) async def resolve_requestable_bigquery_table_resource( self, db: AsyncSession, *, dataset_resource_id: uuid.UUID, table_id: str, table_type: str, ) -> ValidatedRequestableDatasetResource: """Resolve or materialise one table resource under a requestable dataset. Table metadata remains provider-owned. Helix persists only the stable table locator and its DAX owner so requests, approval snapshots, and provider effects share one resource id. New rows inherit the dataset owner until that table is explicitly reassigned. """ dataset = await self.validate_requestable_bigquery_dataset_resource( db, dataset_resource_id, require_resource_owner_authority=True, ) if dataset.resource_type != c.RESOURCE_TYPE_DATASET: raise DataAccessResourceValidationError( "table_parent_not_dataset", "A table resource must be resolved from a dataset resource.", resource_id=dataset_resource_id, ) clean_table_id = str(table_id or "").strip() if not clean_table_id or len(clean_table_id) > 1024 or "/" in clean_table_id: raise DataAccessResourceValidationError( "invalid_table_id", "BigQuery table id must stay within the selected dataset.", resource_id=dataset_resource_id, ) clean_table_type = str(table_type or "TABLE").strip().upper()[:32] or "TABLE" dataset_row = await db.get(DataAccessResource, dataset_resource_id) if dataset_row is None: # defensive: validation above already rejects this raise DataAccessResourceValidationError( "table_parent_not_dataset", "A table resource must be resolved from a dataset resource.", resource_id=dataset_resource_id, ) row = await db.scalar( select(DataAccessResource).where( DataAccessResource.parent_resource_id == dataset_resource_id, DataAccessResource.resource_type == c.RESOURCE_TYPE_TABLE, DataAccessResource.table_id == clean_table_id, DataAccessResource.status == RESOURCE_STATUS_ACTIVE, ), ) if row is None: row = DataAccessResource( integration_id=str(dataset_row.integration_id), provider=c.PROVIDER_BIGQUERY, resource_type=c.RESOURCE_TYPE_TABLE, project_id=dataset.project_id, dataset_id=dataset.dataset_id, parent_resource_id=dataset_resource_id, table_id=clean_table_id, table_type=clean_table_type, display_name=clean_table_id, requestable=True, status=RESOURCE_STATUS_ACTIVE, owner_person_id=dataset.owner_person_id, owner_resolution_status=dataset.owner_resolution_status, owner_source=OWNER_SOURCE_DATASET_INHERITED, source_version=dataset.source_version, source_etag=dataset.source_etag, ) db.add(row) await db.flush() else: row_changed = False if row.table_type != clean_table_type: row.table_type = clean_table_type row_changed = True if not row.owner_person_id: row.owner_person_id = dataset.owner_person_id row.owner_resolution_status = dataset.owner_resolution_status row.owner_source = OWNER_SOURCE_DATASET_INHERITED row_changed = True if row_changed: await db.flush() return await self.validate_requestable_bigquery_dataset_resource( db, row.id, require_resource_owner_authority=True, ) async def transfer_bigquery_dataset_owner( self, db: AsyncSession, *, dataset_resource_id: uuid.UUID, actor: IdentityPerson, new_owner_person_id: uuid.UUID, trace_id: str | None = None, ) -> ValidatedRequestableDatasetResource: """Transfer one dataset and its inherited tables to an active employee.""" dataset_row = await db.scalar( select(DataAccessResource) .where(DataAccessResource.id == dataset_resource_id) .with_for_update(), ) if ( dataset_row is None or dataset_row.resource_type != c.RESOURCE_TYPE_DATASET or dataset_row.status != RESOURCE_STATUS_ACTIVE or not dataset_row.requestable ): raise DataAccessResourceValidationError( "dataset_not_requestable", "The dataset is not available for owner transfer.", resource_id=dataset_resource_id, ) if str(dataset_row.owner_person_id or "") != str(actor.id): raise AppException( code="DATA_ACCESS_DATASET_OWNER_CHANGE_FORBIDDEN", message="Only the current dataset owner can transfer this dataset.", status_code=403, ) if str(new_owner_person_id) == str(actor.id): raise AppException( code="DATA_ACCESS_DATASET_OWNER_SAME_PERSON", message="The new dataset owner must be another active employee.", status_code=422, ) new_owner = await db.get(IdentityPerson, new_owner_person_id) if new_owner is None or new_owner.employment_status != "active": raise AppException( code="DATA_ACCESS_DATASET_OWNER_INVALID", message="The new dataset owner must be an active employee.", status_code=422, ) previous_owner_person_id = str(dataset_row.owner_person_id or "") or None previous_owner_source = dataset_row.owner_source dataset_row.owner_person_id = str(new_owner.id) dataset_row.owner_resolution_status = OWNER_RESOLUTION_RESOLVED dataset_row.owner_source = TRANSFERRED_DATASET_OWNER_SOURCE_BY_SOURCE.get( dataset_row.owner_source, OWNER_SOURCE_MANUAL, ) inherited_tables = ( await db.scalars( select(DataAccessResource) .where( DataAccessResource.parent_resource_id == dataset_resource_id, DataAccessResource.resource_type == c.RESOURCE_TYPE_TABLE, DataAccessResource.owner_source == OWNER_SOURCE_DATASET_INHERITED, ) .with_for_update(), ) ).all() for table_row in inherited_tables: table_row.owner_person_id = str(new_owner.id) table_row.owner_resolution_status = "resolved" await AuditService(db).log( trace_id=trace_id or f"data_access_dataset_owner:{dataset_row.id}", actor_type="user", actor_user_id=actor.id, action="data_access.dataset_owner.transfer", resource_type="data_access_resource", resource_id=str(dataset_row.id), status="success", detail={ "previous_owner_person_id": previous_owner_person_id, "new_owner_person_id": str(new_owner.id), "previous_owner_source": previous_owner_source, "new_owner_source": dataset_row.owner_source, "inherited_table_count": len(inherited_tables), }, ) await db.flush() return await self.validate_requestable_bigquery_dataset_resource( db, dataset_row.id, require_resource_owner_authority=True, ) async def plan_offboarding_owner_handover( self, db: AsyncSession, *, leaver_person_id: uuid.UUID, ) -> DataAccessOffboardingOwnerHandoverPlan: """Lock non-deleted dataset/table rows currently owned by a leaver.""" resources = ( await db.scalars( select(DataAccessResource) .where( DataAccessResource.resource_type.in_( ( c.RESOURCE_TYPE_DATASET, c.RESOURCE_TYPE_TABLE, ), ), DataAccessResource.status != RESOURCE_STATUS_DELETED, DataAccessResource.owner_person_id == str(leaver_person_id), ) .order_by( DataAccessResource.resource_type, DataAccessResource.id, ) .with_for_update(), ) ).all() return DataAccessOffboardingOwnerHandoverPlan(tuple(resources)) async def apply_offboarding_owner_handover( self, db: AsyncSession, *, plan: DataAccessOffboardingOwnerHandoverPlan, leaver_person_id: uuid.UUID, new_owner: IdentityPerson, offboarding_process_id: uuid.UUID, successor_resolution_detail: dict[str, Any], trace_id: str, actor_user_id: uuid.UUID | None, ) -> DataAccessOffboardingOwnerHandoverPlan: """Replace only current resource-owner facts; preserve DAX history/IAM.""" if not plan.required: return plan if ( new_owner.id == leaver_person_id or new_owner.deleted_at is not None or new_owner.employment_status != "active" or not (new_owner.work_email or "").strip() ): raise AppException( code="DATA_ACCESS_OWNER_HANDOVER_INVALID", message=( "Data-access ownership requires another active Helix " "employee as the offboarding handover person." ), status_code=409, ) previous_source_counts: dict[str, int] = {} for resource in plan.resources: previous_source_counts[resource.owner_source] = ( previous_source_counts.get(resource.owner_source, 0) + 1 ) resource.owner_person_id = str(new_owner.id) resource.owner_resolution_status = OWNER_RESOLUTION_RESOLVED if resource.resource_type == c.RESOURCE_TYPE_DATASET: resource.owner_source = TRANSFERRED_DATASET_OWNER_SOURCE_BY_SOURCE.get( resource.owner_source, OWNER_SOURCE_MANUAL, ) await AuditService(db).log( trace_id=trace_id, actor_type="user" if actor_user_id is not None else "system", actor_user_id=actor_user_id, action="offboarding.data_access_owner_handover", resource_type="offboarding_process", resource_id=str(offboarding_process_id), status="success", detail={ "previous_owner_person_id": str(leaver_person_id), "new_owner_person_id": str(new_owner.id), "dataset_count": plan.dataset_count, "table_count": plan.table_count, "previous_owner_source_counts": previous_source_counts, "provider_mutation_attempted": False, **successor_resolution_detail, }, ) await db.flush() return plan async def reassign_bigquery_table_owner( self, db: AsyncSession, *, dataset_resource_id: uuid.UUID, table_id: str, actor: IdentityPerson, new_owner_person_id: uuid.UUID, trace_id: str | None = None, ) -> ValidatedRequestableDatasetResource: """Transfer one table from its current owner to an active employee.""" dataset_row = await db.get(DataAccessResource, dataset_resource_id) if ( dataset_row is None or dataset_row.resource_type != c.RESOURCE_TYPE_DATASET or dataset_row.status != RESOURCE_STATUS_ACTIVE or not dataset_row.requestable ): raise DataAccessResourceValidationError( "dataset_not_requestable", "The dataset is not available for table owner changes.", resource_id=dataset_resource_id, ) clean_table_id = str(table_id or "").strip() table_row = await db.scalar( select(DataAccessResource) .where( DataAccessResource.parent_resource_id == dataset_resource_id, DataAccessResource.resource_type == c.RESOURCE_TYPE_TABLE, DataAccessResource.table_id == clean_table_id, DataAccessResource.status == RESOURCE_STATUS_ACTIVE, ) .with_for_update(), ) if table_row is None: raise AppException( code="DATA_ACCESS_TABLE_NOT_FOUND", message="The selected table is not registered for data access.", status_code=404, ) if str(table_row.owner_person_id or "") != str(actor.id): raise AppException( code="DATA_ACCESS_TABLE_OWNER_CHANGE_FORBIDDEN", message="Only the current table owner can transfer this table.", status_code=403, ) if str(new_owner_person_id) == str(actor.id): raise AppException( code="DATA_ACCESS_TABLE_OWNER_SAME_PERSON", message="The new table owner must be another active employee.", status_code=422, ) new_owner = await db.get(IdentityPerson, new_owner_person_id) if new_owner is None or new_owner.employment_status != "active": raise AppException( code="DATA_ACCESS_TABLE_OWNER_INVALID", message="The new table owner must be an active employee.", status_code=422, ) previous_owner_person_id = str(table_row.owner_person_id or "") or None table_row.owner_person_id = str(new_owner.id) table_row.owner_resolution_status = "resolved" table_row.owner_source = OWNER_SOURCE_MANUAL await AuditService(db).log( trace_id=trace_id or f"data_access_table_owner:{table_row.id}", actor_type="user", actor_user_id=actor.id, action="data_access.table_owner.reassign", resource_type="data_access_resource", resource_id=str(table_row.id), status="success", detail={ "dataset_resource_id": str(dataset_resource_id), "table_id": clean_table_id, "previous_owner_person_id": previous_owner_person_id, "new_owner_person_id": str(new_owner.id), }, ) await db.flush() return await self.validate_requestable_bigquery_dataset_resource( db, table_row.id, require_resource_owner_authority=True, ) async def resolve_table_selection_resource_ids( self, db: AsyncSession, *, selections: Sequence[Any], driver: DataAccessDriver, ) -> tuple[uuid.UUID, ...]: """Validate table selections against BigQuery and return DAX resource ids.""" if not selections: raise DataAccessSubmissionError( "empty_table_selection", "At least one BigQuery table is required.", field="table_selections", ) seen: set[tuple[uuid.UUID, str]] = set() available_by_dataset: dict[uuid.UUID, dict[str, str]] = {} result: list[uuid.UUID] = [] for selection in selections: dataset_resource_id = uuid.UUID(str(selection.dataset_resource_id)) clean_table_id = str(selection.table_id or "").strip() key = (dataset_resource_id, clean_table_id) if key in seen: raise DataAccessSubmissionError( "duplicate_table_selection", "The same BigQuery table cannot be selected twice.", field="table_selections", ) seen.add(key) if dataset_resource_id not in available_by_dataset: dataset = await self.validate_requestable_bigquery_dataset_resource( db, dataset_resource_id, require_resource_owner_authority=True, ) if dataset.resource_type != c.RESOURCE_TYPE_DATASET: raise DataAccessSubmissionError( "table_parent_not_dataset", "Table selections must reference a requestable dataset.", field="table_selections", ) provider_tables = await driver.list_dataset_tables( dataset.project_id, dataset.dataset_id, ) available_by_dataset[dataset_resource_id] = { table.table_id: table.table_type for table in provider_tables } table_type = available_by_dataset[dataset_resource_id].get(clean_table_id) if table_type is None: raise DataAccessSubmissionError( "table_not_requestable", "The selected table is not present in the requestable dataset.", field="table_selections", ) resource = await self.resolve_requestable_bigquery_table_resource( db, dataset_resource_id=dataset_resource_id, table_id=clean_table_id, table_type=table_type, ) result.append(resource.resource_id) return tuple(result) async def validate_google_group_principal( self, db: AsyncSession, principal: uuid.UUID | str, *, principal_type: str = c.PRINCIPAL_TYPE_GOOGLE_GROUP, allow_deleted_rename_bridge: bool = True, ) -> ValidatedGoogleGroupPrincipal: if principal_type != c.PRINCIPAL_TYPE_GOOGLE_GROUP: raise DataAccessPrincipalValidationError( "unsupported_principal_type", "Data-access V1 only accepts Google Group principals.", principal=principal, ) if isinstance(principal, uuid.UUID): group = await db.get(EmailGroup, principal) if group is None: raise DataAccessPrincipalValidationError( "unknown_group", "Google Group principal is unknown to Helix.", principal=principal, ) return await self._validate_email_group_row( db, group, requested_email_group_id=principal, resolved_via="active_id", allow_deleted_rename_bridge=allow_deleted_rename_bridge, ) principal_text = principal.strip().lower() if not principal_text: raise DataAccessPrincipalValidationError( "unknown_group", "Google Group principal is required.", principal=principal, ) try: principal_id = uuid.UUID(principal_text) except ValueError: principal_id = None if principal_id is not None: return await self.validate_google_group_principal( db, principal_id, principal_type=principal_type, allow_deleted_rename_bridge=allow_deleted_rename_bridge, ) active_groups = ( await db.scalars( select(EmailGroup).where(EmailGroup.status == "active"), ) ).all() primary_matches = [ group for group in active_groups if group.email.lower() == principal_text ] alias_matches = [ group for group in active_groups if principal_text in _normalised_aliases(group) ] unique_matches = _unique_groups([*primary_matches, *alias_matches]) if len(unique_matches) > 1: raise DataAccessPrincipalValidationError( "alias_collision", "Google Group principal resolves to multiple active groups.", principal=principal, ) if unique_matches: resolved_via = "active_primary_email" if primary_matches else "active_alias" return await self._validate_email_group_row( db, unique_matches[0], requested_email_group_id=unique_matches[0].id, resolved_via=resolved_via, allow_deleted_rename_bridge=allow_deleted_rename_bridge, ) deleted_matches = ( await db.scalars( select(EmailGroup).where( EmailGroup.status == "deleted", func.lower(EmailGroup.email) == principal_text, ), ) ).all() if len(deleted_matches) > 1: raise DataAccessPrincipalValidationError( "deleted_group_ambiguous", "Deleted Google Group principal is ambiguous.", principal=principal, ) if deleted_matches: return await self._validate_email_group_row( db, deleted_matches[0], requested_email_group_id=deleted_matches[0].id, resolved_via="deleted_primary_email", allow_deleted_rename_bridge=allow_deleted_rename_bridge, ) raise DataAccessPrincipalValidationError( "unknown_group", "Google Group principal is unknown to Helix.", principal=principal, ) async def _validate_email_group_row( self, db: AsyncSession, group: EmailGroup, *, requested_email_group_id: uuid.UUID | None, resolved_via: str, allow_deleted_rename_bridge: bool, ) -> ValidatedGoogleGroupPrincipal: if group.status == "deleted": if not allow_deleted_rename_bridge: raise DataAccessPrincipalValidationError( "deleted_group", "Deleted Google Group principals are not requestable.", principal=group.id, ) group = await self._active_rename_bridge(db, group) resolved_via = "rename_bridge" elif group.status != "active": raise DataAccessPrincipalValidationError( "deleted_group", "Only active Google Group principals are requestable.", principal=group.id, ) if not group.google_group_id: raise DataAccessPrincipalValidationError( "missing_google_group_id", "Google Group principal has no Google group id anchor.", principal=group.id, ) active_same_google_id = ( await db.scalars( select(EmailGroup).where( EmailGroup.status == "active", EmailGroup.google_group_id == group.google_group_id, ), ) ).all() if len(active_same_google_id) != 1: raise DataAccessPrincipalValidationError( "duplicate_active_google_group_id", "Google group id resolves to multiple active groups.", principal=group.id, ) primary_email = group.email.lower() return ValidatedGoogleGroupPrincipal( principal_email_group_id=group.id, requested_email_group_id=requested_email_group_id, google_group_id=group.google_group_id, primary_email_snapshot=primary_email, provider_member=f"{c.BIGQUERY_GROUP_MEMBER_PREFIX}{primary_email}", resolved_via=resolved_via, principal_id=await self._group_data_access_principal_id(db, group), ) async def _group_data_access_principal_id( self, db: AsyncSession, group: EmailGroup, ) -> uuid.UUID | None: subject_id = group.google_group_id or f"email-group:{group.id}" principal_id = await db.scalar( select(DataAccessPrincipal.id).where( DataAccessPrincipal.principal_type == c.PRINCIPAL_TYPE_GOOGLE_GROUP, DataAccessPrincipal.external_subject_id == subject_id, ), ) return uuid.UUID(str(principal_id)) if principal_id is not None else None async def authorize_requester_for_google_group_principal( self, db: AsyncSession, requester: IdentityPerson, principal: ValidatedGoogleGroupPrincipal, ) -> DataAccessRequesterAuthorization: requester_role_names = await self._requester_role_names(db, requester) if has_role(requester_role_names, GLOBAL_ADMIN_ROLE): return DataAccessRequesterAuthorization( requester_person_id=requester.id, principal_email_group_id=principal.principal_email_group_id, authorized_as="global_admin", ) owner_row = await db.scalar( select(EmailGroupOwner.id).where( EmailGroupOwner.email_group_id == principal.principal_email_group_id, EmailGroupOwner.person_id == requester.id, ), ) if owner_row is not None: return DataAccessRequesterAuthorization( requester_person_id=requester.id, principal_email_group_id=principal.principal_email_group_id, authorized_as="group_owner", ) member_group_ids = await self._requester_member_group_ids(db, requester) if principal.principal_email_group_id in member_group_ids: return DataAccessRequesterAuthorization( requester_person_id=requester.id, principal_email_group_id=principal.principal_email_group_id, authorized_as="group_member", ) raise DataAccessRequesterAuthorizationError( requester_person_id=requester.id, principal_email_group_id=principal.principal_email_group_id, ) def compile_bigquery_dataset_read_dry_run( self, resource: ValidatedRequestableDatasetResource, principal: ValidatedGoogleGroupPrincipal, entitlement: DataAccessEntitlement, ) -> DataAccessDryRunCompiledEffect: if resource.provider != c.PROVIDER_BIGQUERY or resource.resource_type not in { c.RESOURCE_TYPE_DATASET, c.RESOURCE_TYPE_TABLE, } or ( resource.resource_type == c.RESOURCE_TYPE_TABLE and not resource.table_id ): raise DataAccessCompilerError( "unsupported_resource", "Dry-run compiler only supports exact BigQuery dataset or table resources.", field="resource", ) if ( not entitlement.is_canonical_read_v1 or entitlement.status != ENTITLEMENT_STATUS_ACTIVE or not entitlement.requestable ): raise DataAccessCompilerError( "unsupported_entitlement", "Dry-run compiler only supports requestable read:v1.", field="entitlement", ) primary_email = principal.primary_email_snapshot.strip().lower() google_group_id = principal.google_group_id.strip() if not primary_email or not google_group_id: raise DataAccessCompilerError( "invalid_principal_snapshot", "Google Group id and primary email snapshots are required.", field="principal", ) return DataAccessDryRunCompiledEffect( resource_id=resource.resource_id, principal_email_group_id=principal.principal_email_group_id, principal_google_group_id_snapshot=google_group_id, principal_email_snapshot=primary_email, entitlement_id=entitlement.id, provider=c.PROVIDER_BIGQUERY, resource_type=resource.resource_type, target_project_id=resource.project_id, target_dataset_id=resource.dataset_id, target_table_id=resource.table_id, provider_member=f"{c.BIGQUERY_GROUP_MEMBER_PREFIX}{primary_email}", provider_role=c.BIGQUERY_DATA_VIEWER_ROLE, desired_state=EFFECT_DESIRED_PRESENT, entitlement_version_snapshot=entitlement.version, resource_source_version_snapshot=resource.source_version, resource_source_etag_snapshot=resource.source_etag, compiler_version=DATA_ACCESS_COMPILER_VERSION, managed_reason=EFFECT_MANAGED_REASON_GRANT_REQUEST, already_present_at_first_apply=False, principal_id=principal.principal_id, principal_type_snapshot=c.PRINCIPAL_TYPE_GOOGLE_GROUP, principal_subject_snapshot=google_group_id, provider_access_kind="groupByEmail", ) async def validate_gcp_service_account_principal( self, db: AsyncSession, principal_id: uuid.UUID, ) -> ValidatedServiceAccountPrincipal: principal = await db.get(DataAccessPrincipal, principal_id) if ( principal is None or principal.principal_type != c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT or principal.status != PRINCIPAL_STATUS_ACTIVE ): raise DataAccessPrincipalValidationError( "principal_not_requestable", "Service-account principal is not requestable.", principal=principal_id, ) if ( not principal.source_version or not principal.source_etag or not _is_valid_service_account_email(principal.email_snapshot) ): raise DataAccessPrincipalValidationError( "principal_invalid", "Service-account request record is incomplete.", principal=principal_id, ) return ValidatedServiceAccountPrincipal( principal_id=principal.id, subject_snapshot=principal.external_subject_id, email_snapshot=principal.email_snapshot, source_version=principal.source_version, source_etag=principal.source_etag, ) async def record_service_account_principal( self, db: AsyncSession, service_account_email: str, ) -> ValidatedServiceAccountPrincipal: """Record a typed GCP service-account principal for one DAX request. This is deliberately an input projection, not a pre-registration or ownership directory. The request still needs the normal data-owner approval before a reconciler can write a BigQuery dataset ACL. """ email = _normalise_service_account_email(service_account_email) principal = await db.scalar( select(DataAccessPrincipal) .where( DataAccessPrincipal.principal_type == c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT, func.lower(DataAccessPrincipal.email_snapshot) == email, ) .with_for_update(), ) if principal is None: try: async with db.begin_nested(): principal = DataAccessPrincipal( principal_type=c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT, external_subject_id=email, email_snapshot=email, status=PRINCIPAL_STATUS_ACTIVE, source_version=( c.GCP_SERVICE_ACCOUNT_SELF_SERVICE_SOURCE_VERSION ), source_etag=_self_service_principal_etag(email), source_payload_digest=_self_service_principal_etag(email), ) db.add(principal) await db.flush() except IntegrityError: principal = await db.scalar( select(DataAccessPrincipal) .where( DataAccessPrincipal.principal_type == c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT, func.lower(DataAccessPrincipal.email_snapshot) == email, ) .with_for_update(), ) if principal is None: raise DataAccessPrincipalValidationError( "principal_not_requestable", "Service-account principal could not be recorded.", principal=email, ) if ( principal.source_version != c.GCP_SERVICE_ACCOUNT_SELF_SERVICE_SOURCE_VERSION ): # A historical catalog projection must not become an authoring # prerequisite. Reuse the durable row (and its request history), # but make this request path's selector/provenance authoritative. principal.email_snapshot = email principal.status = PRINCIPAL_STATUS_ACTIVE principal.catalog_id = None principal.source_version = c.GCP_SERVICE_ACCOUNT_SELF_SERVICE_SOURCE_VERSION principal.source_etag = _self_service_principal_etag(email) principal.source_payload_digest = _self_service_principal_etag(email) profile = await db.get(DataAccessServiceAccountProfile, principal.id) if profile is not None: profile.lifecycle_provenance = "observed_service_account_catalog" await db.flush() return await self.validate_gcp_service_account_principal(db, principal.id) async def validate_google_user_principal( self, db: AsyncSession, *, principal_id: uuid.UUID, requester: IdentityPerson, ) -> ValidatedPersonalPrincipal: """Validate that a stored personal principal still belongs to its user.""" if requester.employment_status != "active": raise DataAccessPrincipalValidationError( "requester_inactive", "Personal data access requires an active employee.", principal=requester.id, ) email = _normalise_personal_work_email(requester.work_email) principal = await db.get(DataAccessPrincipal, principal_id) expected_subject = str(requester.id) expected_etag = _personal_principal_etag(expected_subject, email) if ( principal is None or principal.principal_type != c.PRINCIPAL_TYPE_GOOGLE_USER or principal.status != PRINCIPAL_STATUS_ACTIVE or principal.external_subject_id != expected_subject or principal.email_snapshot != email or principal.source_version != c.GOOGLE_USER_SELF_SERVICE_SOURCE_VERSION or principal.source_etag != expected_etag ): raise DataAccessPrincipalValidationError( "principal_not_requestable", "Personal work-account principal is not current.", principal=principal_id, ) return ValidatedPersonalPrincipal( principal_id=principal.id, subject_snapshot=principal.external_subject_id, email_snapshot=principal.email_snapshot, source_version=principal.source_version, source_etag=principal.source_etag, ) async def record_personal_principal( self, db: AsyncSession, *, requester: IdentityPerson, ) -> ValidatedPersonalPrincipal: """Project the authenticated employee as their own DAX principal. The client never supplies an email or person id. The immutable subject is the Helix identity UUID; the work-email snapshot is the compiled BigQuery ``user:`` member and is refreshed only from that identity. """ if requester.employment_status != "active": raise DataAccessSubmissionError( "personal_requester_inactive", "Only active employees can submit personal data-access requests.", field="requester", status_code=422, ) subject = str(requester.id) email = _normalise_personal_work_email(requester.work_email) etag = _personal_principal_etag(subject, email) principal = await db.scalar( select(DataAccessPrincipal) .where( DataAccessPrincipal.principal_type == c.PRINCIPAL_TYPE_GOOGLE_USER, DataAccessPrincipal.external_subject_id == subject, ) .with_for_update(), ) if principal is None: try: async with db.begin_nested(): principal = DataAccessPrincipal( principal_type=c.PRINCIPAL_TYPE_GOOGLE_USER, external_subject_id=subject, email_snapshot=email, status=PRINCIPAL_STATUS_ACTIVE, source_version=c.GOOGLE_USER_SELF_SERVICE_SOURCE_VERSION, source_etag=etag, source_payload_digest=etag, ) db.add(principal) await db.flush() except IntegrityError: principal = await db.scalar( select(DataAccessPrincipal) .where( DataAccessPrincipal.principal_type == c.PRINCIPAL_TYPE_GOOGLE_USER, DataAccessPrincipal.external_subject_id == subject, ) .with_for_update(), ) if principal is None: raise DataAccessPrincipalValidationError( "principal_not_requestable", "Personal work-account principal could not be recorded.", principal=requester.id, ) principal.email_snapshot = email principal.status = PRINCIPAL_STATUS_ACTIVE principal.catalog_id = None principal.source_version = c.GOOGLE_USER_SELF_SERVICE_SOURCE_VERSION principal.source_etag = etag principal.source_payload_digest = etag await db.flush() return await self.validate_google_user_principal( db, principal_id=principal.id, requester=requester, ) def compile_bigquery_dataset_service_account_read_dry_run( self, resource: ValidatedRequestableDatasetResource, principal: ValidatedServiceAccountPrincipal, entitlement: DataAccessEntitlement, ) -> DataAccessDryRunCompiledEffect: if not entitlement.is_canonical_read_v1: raise DataAccessCompilerError( "unsupported_entitlement", "Dry-run compiler only supports read:v1.", field="entitlement", ) return DataAccessDryRunCompiledEffect( resource_id=resource.resource_id, principal_email_group_id=None, principal_google_group_id_snapshot=None, principal_email_snapshot=principal.email_snapshot, entitlement_id=entitlement.id, provider=c.PROVIDER_BIGQUERY, resource_type=resource.resource_type, target_project_id=resource.project_id, target_dataset_id=resource.dataset_id, target_table_id=resource.table_id, provider_member=( f"{c.BIGQUERY_SERVICE_ACCOUNT_MEMBER_PREFIX}{principal.email_snapshot}" ), provider_role=c.BIGQUERY_DATA_VIEWER_ROLE, desired_state=EFFECT_DESIRED_PRESENT, entitlement_version_snapshot=entitlement.version, resource_source_version_snapshot=resource.source_version, resource_source_etag_snapshot=resource.source_etag, compiler_version=DATA_ACCESS_COMPILER_VERSION, managed_reason=EFFECT_MANAGED_REASON_GRANT_REQUEST, already_present_at_first_apply=False, principal_id=principal.principal_id, principal_type_snapshot=c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT, principal_subject_snapshot=principal.subject_snapshot, provider_access_kind="userByEmail", ) def compile_bigquery_dataset_personal_read_dry_run( self, resource: ValidatedRequestableDatasetResource, principal: ValidatedPersonalPrincipal, entitlement: DataAccessEntitlement, ) -> DataAccessDryRunCompiledEffect: if not entitlement.is_canonical_read_v1: raise DataAccessCompilerError( "unsupported_entitlement", "Dry-run compiler only supports read:v1.", field="entitlement", ) return DataAccessDryRunCompiledEffect( resource_id=resource.resource_id, principal_email_group_id=None, principal_google_group_id_snapshot=None, principal_email_snapshot=principal.email_snapshot, entitlement_id=entitlement.id, provider=c.PROVIDER_BIGQUERY, resource_type=resource.resource_type, target_project_id=resource.project_id, target_dataset_id=resource.dataset_id, target_table_id=resource.table_id, provider_member=f"{c.BIGQUERY_USER_MEMBER_PREFIX}{principal.email_snapshot}", provider_role=c.BIGQUERY_DATA_VIEWER_ROLE, desired_state=EFFECT_DESIRED_PRESENT, entitlement_version_snapshot=entitlement.version, resource_source_version_snapshot=resource.source_version, resource_source_etag_snapshot=resource.source_etag, compiler_version=DATA_ACCESS_COMPILER_VERSION, managed_reason=EFFECT_MANAGED_REASON_GRANT_REQUEST, already_present_at_first_apply=False, principal_id=principal.principal_id, principal_type_snapshot=c.PRINCIPAL_TYPE_GOOGLE_USER, principal_subject_snapshot=principal.subject_snapshot, provider_access_kind="userByEmail", ) def build_feishu_submission_payload( self, *, flow: DataAccessApprovalFlow, resources: Sequence[ValidatedRequestableDatasetResource], principal: ValidatedGoogleGroupPrincipal, entitlement: DataAccessEntitlement, business_reason: str, detail_url: str, ) -> DataAccessFeishuSubmissionPayload: if flow.flow_key != c.APPROVAL_FLOW_DATA_ACCESS_GROUP_GRANT: raise DataAccessFeishuPayloadError( "unsupported_flow", "Data-access Feishu payload requires data_access_group_grant.", field="flow_key", ) external_code = flow.external_code.strip() if not external_code: raise DataAccessFeishuPayloadError( "missing_external_code", "Data-access Feishu approval flow has no external code.", field="external_code", ) clean_reason = _clean_required_multiline( business_reason, field="business_reason", max_length=DATA_ACCESS_BUSINESS_REASON_MAX_LENGTH, ) clean_detail_url = _clean_required_single_line( detail_url, field="detail_url", max_length=DATA_ACCESS_DETAIL_URL_MAX_LENGTH, ) if not _is_valid_detail_url(clean_detail_url): raise DataAccessFeishuPayloadError( "invalid_detail_url", "Data-access Feishu detail URL must be relative or HTTP(S).", field="detail_url", ) if not resources: raise DataAccessFeishuPayloadError( "empty_dataset_selection", "At least one BigQuery dataset is required.", field="resources", ) seen_resource_ids: set[uuid.UUID] = set() dataset_context: list[DataAccessFeishuDatasetContext] = [] compiled_preview: list[DataAccessFeishuCompiledPreview] = [] for resource in resources: if resource.resource_id in seen_resource_ids: raise DataAccessFeishuPayloadError( "duplicate_dataset", "Data-access Feishu payload cannot repeat a dataset.", field="resources", ) seen_resource_ids.add(resource.resource_id) effect = self.compile_bigquery_dataset_read_dry_run( resource, principal, entitlement, ) dataset_context.append( DataAccessFeishuDatasetContext( resource_id=resource.resource_id, provider=resource.provider, resource_type=resource.resource_type, project_id=resource.project_id, dataset_id=resource.dataset_id, owner_person_id=resource.owner_person_id, owner_resolution_status=resource.owner_resolution_status, owner_authority_enabled=resource.owner_authority_enabled, source_version=resource.source_version, source_etag=resource.source_etag, description=resource.description, table_id=resource.table_id, table_type=resource.table_type, ), ) compiled_preview.append( DataAccessFeishuCompiledPreview( resource_id=effect.resource_id, target_project_id=effect.target_project_id, target_dataset_id=effect.target_dataset_id, target_table_id=effect.target_table_id, provider_member=effect.provider_member, provider_role=effect.provider_role, desired_state=effect.desired_state, entitlement_version_snapshot=(effect.entitlement_version_snapshot), resource_source_version_snapshot=(effect.resource_source_version_snapshot), resource_source_etag_snapshot=(effect.resource_source_etag_snapshot), compiler_version=effect.compiler_version, managed_reason=effect.managed_reason, already_present_at_first_apply=(effect.already_present_at_first_apply), ), ) return DataAccessFeishuSubmissionPayload( flow_key=flow.flow_key, external_code=external_code, payload_version=DATA_ACCESS_FEISHU_PAYLOAD_VERSION, business_reason=clean_reason, detail_url=clean_detail_url, principal_email_group_id=principal.principal_email_group_id, principal_google_group_id_snapshot=principal.google_group_id.strip(), principal_email_snapshot=principal.primary_email_snapshot.strip().lower(), entitlement_id=entitlement.id, entitlement_version=entitlement.version, dataset_count=len(dataset_context), datasets=tuple(dataset_context), compiled_preview=tuple(compiled_preview), ) def build_feishu_service_account_submission_payload( self, *, flow: DataAccessApprovalFlow, resources: Sequence[ValidatedRequestableDatasetResource], principal: ValidatedServiceAccountPrincipal, entitlement: DataAccessEntitlement, business_reason: str, detail_url: str, ) -> DataAccessServiceAccountFeishuSubmissionPayload: if flow.flow_key != c.APPROVAL_FLOW_DATA_ACCESS_SERVICE_ACCOUNT_GRANT: raise DataAccessFeishuPayloadError( "unsupported_flow", "Service-account request requires its dedicated approval flow.", field="flow_key", ) if not flow.external_code.strip(): raise DataAccessFeishuPayloadError( "missing_external_code", "Service-account approval flow has no external code.", field="external_code", ) clean_reason = _clean_required_multiline( business_reason, field="business_reason", max_length=DATA_ACCESS_BUSINESS_REASON_MAX_LENGTH, ) clean_detail_url = _clean_required_single_line( detail_url, field="detail_url", max_length=DATA_ACCESS_DETAIL_URL_MAX_LENGTH, ) if not _is_valid_detail_url(clean_detail_url): raise DataAccessFeishuPayloadError( "invalid_detail_url", "Data-access Feishu detail URL must be relative or HTTP(S).", field="detail_url", ) if not resources: raise DataAccessFeishuPayloadError( "empty_dataset_selection", "At least one BigQuery dataset is required.", field="resources", ) datasets: list[DataAccessFeishuDatasetContext] = [] previews: list[DataAccessFeishuCompiledPreview] = [] seen_resources: set[uuid.UUID] = set() for resource in resources: if resource.resource_id in seen_resources: raise DataAccessFeishuPayloadError( "duplicate_dataset", "Data-access request cannot repeat a dataset.", field="resources", ) seen_resources.add(resource.resource_id) effect = self.compile_bigquery_dataset_service_account_read_dry_run( resource, principal, entitlement, ) datasets.append(DataAccessFeishuDatasetContext( resource_id=resource.resource_id, provider=resource.provider, resource_type=resource.resource_type, project_id=resource.project_id, dataset_id=resource.dataset_id, owner_person_id=resource.owner_person_id, owner_resolution_status=resource.owner_resolution_status, owner_authority_enabled=resource.owner_authority_enabled, source_version=resource.source_version, source_etag=resource.source_etag, description=resource.description, table_id=resource.table_id, table_type=resource.table_type, )) previews.append(DataAccessFeishuCompiledPreview( resource_id=effect.resource_id, target_project_id=effect.target_project_id, target_dataset_id=effect.target_dataset_id, target_table_id=effect.target_table_id, provider_member=effect.provider_member, provider_role=effect.provider_role, desired_state=effect.desired_state, entitlement_version_snapshot=effect.entitlement_version_snapshot, resource_source_version_snapshot=effect.resource_source_version_snapshot, resource_source_etag_snapshot=effect.resource_source_etag_snapshot, compiler_version=effect.compiler_version, managed_reason=effect.managed_reason, already_present_at_first_apply=effect.already_present_at_first_apply, )) return DataAccessServiceAccountFeishuSubmissionPayload( flow_key=flow.flow_key, external_code=flow.external_code.strip(), payload_version=DATA_ACCESS_FEISHU_PAYLOAD_VERSION, business_reason=clean_reason, detail_url=clean_detail_url, principal=principal, entitlement_id=entitlement.id, entitlement_version=entitlement.version, datasets=tuple(datasets), compiled_preview=tuple(previews), ) def build_feishu_personal_submission_payload( self, *, flow: DataAccessApprovalFlow, resources: Sequence[ValidatedRequestableDatasetResource], principal: ValidatedPersonalPrincipal, entitlement: DataAccessEntitlement, business_reason: str, detail_url: str, ) -> DataAccessPersonalFeishuSubmissionPayload: if flow.flow_key != c.APPROVAL_FLOW_DATA_ACCESS_PERSONAL_GRANT: raise DataAccessFeishuPayloadError( "unsupported_flow", "Personal request requires its dedicated approval flow.", field="flow_key", ) if not flow.external_code.strip(): raise DataAccessFeishuPayloadError( "missing_external_code", "Personal approval flow has no external code.", field="external_code", ) clean_reason = _clean_required_multiline( business_reason, field="business_reason", max_length=DATA_ACCESS_BUSINESS_REASON_MAX_LENGTH, ) clean_detail_url = _clean_required_single_line( detail_url, field="detail_url", max_length=DATA_ACCESS_DETAIL_URL_MAX_LENGTH, ) if not _is_valid_detail_url(clean_detail_url): raise DataAccessFeishuPayloadError( "invalid_detail_url", "Data-access Feishu detail URL must be relative or HTTP(S).", field="detail_url", ) if not resources: raise DataAccessFeishuPayloadError( "empty_dataset_selection", "At least one BigQuery dataset is required.", field="resources", ) datasets: list[DataAccessFeishuDatasetContext] = [] previews: list[DataAccessFeishuCompiledPreview] = [] seen_resources: set[uuid.UUID] = set() for resource in resources: if resource.resource_id in seen_resources: raise DataAccessFeishuPayloadError( "duplicate_dataset", "Data-access request cannot repeat a dataset.", field="resources", ) seen_resources.add(resource.resource_id) effect = self.compile_bigquery_dataset_personal_read_dry_run( resource, principal, entitlement, ) datasets.append(DataAccessFeishuDatasetContext( resource_id=resource.resource_id, provider=resource.provider, resource_type=resource.resource_type, project_id=resource.project_id, dataset_id=resource.dataset_id, owner_person_id=resource.owner_person_id, owner_resolution_status=resource.owner_resolution_status, owner_authority_enabled=resource.owner_authority_enabled, source_version=resource.source_version, source_etag=resource.source_etag, description=resource.description, table_id=resource.table_id, table_type=resource.table_type, )) previews.append(DataAccessFeishuCompiledPreview( resource_id=effect.resource_id, target_project_id=effect.target_project_id, target_dataset_id=effect.target_dataset_id, target_table_id=effect.target_table_id, provider_member=effect.provider_member, provider_role=effect.provider_role, desired_state=effect.desired_state, entitlement_version_snapshot=effect.entitlement_version_snapshot, resource_source_version_snapshot=effect.resource_source_version_snapshot, resource_source_etag_snapshot=effect.resource_source_etag_snapshot, compiler_version=effect.compiler_version, managed_reason=effect.managed_reason, already_present_at_first_apply=effect.already_present_at_first_apply, )) return DataAccessPersonalFeishuSubmissionPayload( flow_key=flow.flow_key, external_code=flow.external_code.strip(), payload_version=DATA_ACCESS_FEISHU_PAYLOAD_VERSION, business_reason=clean_reason, detail_url=clean_detail_url, principal=principal, entitlement_id=entitlement.id, entitlement_version=entitlement.version, datasets=tuple(datasets), compiled_preview=tuple(previews), ) def build_feishu_submission_form( self, *, flow: DataAccessApprovalFlow, payload: ( DataAccessFeishuSubmissionPayload | DataAccessServiceAccountFeishuSubmissionPayload | DataAccessPersonalFeishuSubmissionPayload ), ) -> str: """Render canonical DAX payload fields into configured Feishu widgets.""" if not flow.payload_schema: raise DataAccessSubmissionError( "missing_payload_schema", "Data-access Feishu flow must map canonical fields to widgets.", field="payload_schema", status_code=503, ) fields = payload.to_submission_fields() form: list[dict[str, Any]] = [] for field_name in DATA_ACCESS_FEISHU_FORM_FIELD_ORDER: widget_id, widget_type = _resolve_feishu_widget_spec( field_name, flow.payload_schema.get(field_name), ) form.append( { "id": widget_id, "type": widget_type, "value": _feishu_form_value(fields[field_name]), } ) return json.dumps(form, ensure_ascii=False, sort_keys=True) async def _resolve_feishu_identity_by_email( self, client: Any, email: str, *, field: str, subject_label: str, prefer_open_id: bool = False, required_id_type: str | None = None, ) -> tuple[str, str]: clean_email = str(email or "").strip().lower() if not clean_email: raise DataAccessSubmissionError( "missing_feishu_identity_email", f"{subject_label} must have a work email before Feishu submission.", field=field, status_code=503, ) if required_id_type not in {None, "user_id", "open_id"}: raise ValueError(f"unsupported Feishu id type: {required_id_type!r}") last_error: AppException | None = None async def _resolve_open_id() -> tuple[str, str] | None: if not hasattr(client, "resolve_open_id_by_email"): return None open_id = await client.resolve_open_id_by_email(clean_email) if open_id: return "open_id", str(open_id) return None if required_id_type == "open_id": try: resolved = await _resolve_open_id() except AppException: raise if resolved is not None: return resolved raise DataAccessSubmissionError( "feishu_identity_not_found", f"{subject_label} could not be resolved to a Feishu open_id.", field=field, status_code=503, ) if required_id_type == "user_id": try: user_id = await client.resolve_user_id_by_email(clean_email) except AppException: raise if user_id: return "user_id", str(user_id) raise DataAccessSubmissionError( "feishu_identity_not_found", f"{subject_label} could not be resolved to a Feishu user_id.", field=field, status_code=503, ) if prefer_open_id and hasattr(client, "resolve_open_id_by_email"): try: resolved = await _resolve_open_id() if resolved is not None: return resolved except AppException as exc: last_error = exc try: user_id = await client.resolve_user_id_by_email(clean_email) if user_id: return "user_id", str(user_id) except AppException as exc: last_error = exc if not hasattr(client, "resolve_open_id_by_email"): raise if not prefer_open_id: try: resolved = await _resolve_open_id() if resolved is not None: return resolved except AppException as exc: last_error = exc if last_error is not None: raise last_error raise DataAccessSubmissionError( "feishu_identity_not_found", f"{subject_label} could not be resolved to a Feishu user id.", field=field, status_code=503, ) async def _build_feishu_group_owner_cc_payload( self, db: AsyncSession, *, client: Any, principal: ValidatedGoogleGroupPrincipal, requester: IdentityPerson, actor_id_type: str, ) -> dict[str, Any]: if actor_id_type not in {"user_id", "open_id"}: raise ValueError(f"unsupported Feishu id type: {actor_id_type!r}") owners = ( await db.scalars( select(IdentityPerson) .join(EmailGroupOwner, EmailGroupOwner.person_id == IdentityPerson.id) .where( EmailGroupOwner.email_group_id == principal.principal_email_group_id, IdentityPerson.id != requester.id, IdentityPerson.employment_status == "active", ) .order_by(IdentityPerson.work_email, IdentityPerson.id), ) ).all() if not owners: return {} cc_user_ids: list[str] = [] for owner in owners: _, id_value = await self._resolve_feishu_identity_by_email( client, owner.work_email, field="principal.group_owner", subject_label="Google Group owner", required_id_type=actor_id_type, ) if id_value not in cc_user_ids: cc_user_ids.append(id_value) if not cc_user_ids: return {} return { "id_type": actor_id_type, "cc_user_ids": cc_user_ids, "comment": ( "FYI: a Google Group member submitted this BigQuery dataset " "access request for a group you own." ), } async def _build_feishu_data_owner_approver_payload( self, db: AsyncSession, *, client: Any, flow: DataAccessApprovalFlow, resources: Sequence[ValidatedRequestableDatasetResource], additional_owner_ids: Sequence[uuid.UUID] = (), ) -> dict[str, Any]: node_id = _resolve_feishu_data_owner_review_node_id(flow.payload_schema) if node_id is None: return {} owner_ids: list[uuid.UUID] = [] seen_owner_ids: set[uuid.UUID] = set() for resource in resources: owner_id = _uuid_snapshot( resource.owner_person_id, field="resource_owner_person_id", ) if owner_id is None: raise DataAccessSubmissionError( "missing_resource_owner", "Dynamic data-owner approval requires every dataset to have a resolved owner.", field="resource.owner_person_id", status_code=503, ) if owner_id not in seen_owner_ids: seen_owner_ids.add(owner_id) owner_ids.append(owner_id) for owner_id in additional_owner_ids: if owner_id not in seen_owner_ids: seen_owner_ids.add(owner_id) owner_ids.append(owner_id) owners = ( await db.scalars( select(IdentityPerson).where(IdentityPerson.id.in_(owner_ids)), ) ).all() owners_by_id = {owner.id: owner for owner in owners} node_user_ids: list[str] = [] node_open_ids: list[str] = [] for owner_id in owner_ids: owner = owners_by_id.get(owner_id) if owner is None: raise DataAccessSubmissionError( "resource_owner_identity_missing", "Dynamic data-owner approval requires every dataset owner " "to exist in Helix identity.", field="resource.owner_person_id", status_code=503, ) # Departed-owner guard (2026-07-05): the data-owner is the # AUTHORITATIVE approver for the grant — routing to a departed # owner leaves the approval hanging on a frozen/removed Feishu # account (the callback only re-checks the requester, never the # approver). Fail loud with an actionable error so the dataset is # re-owned before access is requested, rather than silently # misrouting. Mirrors the group-owner CC path which already # filters owners to employment_status == "active". if owner.employment_status != "active": raise DataAccessSubmissionError( "resource_owner_departed", "Dataset owner " f"{owner.display_name or owner.work_email} is no longer an " "active employee (has left or is leaving); reassign the " "dataset owner before requesting access to this dataset.", field="resource.owner_person_id", status_code=409, ) id_type, id_value = await self._resolve_feishu_identity_by_email( client, owner.work_email, field="resource.owner_person_id", subject_label="Resource owner", prefer_open_id=True, ) if id_type == "open_id": node_open_ids.append(id_value) else: node_user_ids.append(id_value) payload: dict[str, Any] = {} if node_user_ids: payload["node_approver_user_id_list"] = [ {"key": node_id, "value": node_user_ids}, ] if node_open_ids: payload["node_approver_open_id_list"] = [ {"key": node_id, "value": node_open_ids}, ] return payload async def submit_service_account_grant_request( self, db: AsyncSession, *, requester: IdentityPerson, service_account_email: str, resource_ids: Sequence[uuid.UUID], business_reason: str, detail_url: str, idempotency_key: str, trace_id: str | None = None, feishu_client: Any | None = None, ) -> DataAccessRequestSubmissionResult: """Submit a typed-SA request; provider state remains untouched.""" clean_service_account_email = _normalise_service_account_email( service_account_email, ) clean_key = _clean_optional_single_line( idempotency_key, field="idempotency_key", max_length=DATA_ACCESS_IDEMPOTENCY_KEY_MAX_LENGTH, ) if not clean_key: raise DataAccessSubmissionError( "idempotency_key_required", "Idempotency-Key is required for service-account grants.", field="idempotency_key", status_code=422, ) clean_trace_id = _clean_optional_single_line( trace_id, field="trace_id", max_length=DATA_ACCESS_TRACE_ID_MAX_LENGTH, ) existing = await self._existing_request_for_idempotency_key( db, clean_key, requester_person_id=requester.id, ) if existing is not None: if ( existing.principal_type_snapshot != c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT or existing.principal_email_snapshot != clean_service_account_email or not await self._request_matches_submission_fingerprint( db, request=existing, resource_ids=resource_ids, business_reason=business_reason, ) ): raise DataAccessSubmissionError( "idempotency_key_reused", "Idempotency-Key was used for a different request.", field="idempotency_key", status_code=409, ) return await self._submission_result_from_request( db, existing, reused_existing=True, ) if not resource_ids: raise DataAccessSubmissionError( "empty_dataset_selection", "At least one BigQuery dataset is required.", field="resource_ids", ) flow = await resolve_data_access_service_account_grant_flow(session=db) resources = [ await self.validate_requestable_bigquery_dataset_resource( db, resource_id, require_resource_owner_authority=True, ) for resource_id in resource_ids ] principal = await self.record_service_account_principal( db, clean_service_account_email, ) entitlement = await self._load_read_v1_entitlement(db) payload = self.build_feishu_service_account_submission_payload( flow=flow, resources=resources, principal=principal, entitlement=entitlement, business_reason=business_reason, detail_url=detail_url, ) form_payload = self.build_feishu_submission_form(flow=flow, payload=payload) client = feishu_client or await self._data_access_feishu_client(db) feishu_user_id_type, feishu_user_id = await self._resolve_feishu_identity_by_email( client, requester.work_email, field="requester", subject_label="Requester", ) approver_payload = await self._build_feishu_data_owner_approver_payload( db, client=client, flow=flow, resources=resources, ) request_row = DataAccessRequest( requester_person_id=requester.id, principal_id=principal.principal_id, principal_type_snapshot=c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT, principal_subject_snapshot=principal.subject_snapshot, principal_email_snapshot=principal.email_snapshot, entitlement_id=entitlement.id, entitlement_snapshot=c.ENTITLEMENT_READ, entitlement_version_snapshot=ENTITLEMENT_VERSION_V1, approval_flow_key=flow.flow_key, idempotency_key=clean_key, business_reason=payload.business_reason, status=REQUEST_STATUS_PENDING_FEISHU, trace_id=clean_trace_id, items=[ DataAccessRequestItem( resource_id=resource.resource_id, line_no=line_no, resource_project_id_snapshot=resource.project_id, resource_dataset_id_snapshot=resource.dataset_id, resource_source_version_snapshot=resource.source_version, resource_source_etag_snapshot=resource.source_etag, resource_owner_person_id_snapshot=_uuid_snapshot( resource.owner_person_id, field="resource_owner_person_id", ), status=REQUEST_ITEM_STATUS_PENDING, ) for line_no, resource in enumerate(resources, start=1) ], ) try: async with db.begin_nested(): db.add(request_row) await db.flush() except IntegrityError: existing = await self._existing_request_for_idempotency_key( db, clean_key, requester_person_id=requester.id, lock=True, ) if existing is not None: if ( existing.principal_type_snapshot != c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT or existing.principal_email_snapshot != clean_service_account_email or not await self._request_matches_submission_fingerprint( db, request=existing, resource_ids=resource_ids, business_reason=business_reason, ) ): raise DataAccessSubmissionError( "idempotency_key_reused", "Idempotency-Key was used for a different request.", field="idempotency_key", status_code=409, ) return await self._submission_result_from_request( db, existing, dataset_count=len(resources), reused_existing=True, ) raise instance_code: str | None = None try: create_kwargs: dict[str, Any] = { "approval_code": payload.external_code, "form": form_payload, **approver_payload, } create_kwargs["open_id" if feishu_user_id_type == "open_id" else "user_id"] = str(feishu_user_id) instance = await client.create_instance( instance_uuid=str(request_row.id), **create_kwargs, ) instance_code = str(instance.get("instance_code") or "").strip() if not instance_code: raise DataAccessSubmissionError( "missing_feishu_instance_code", "Feishu approval response did not include a code.", field="feishu", status_code=503, ) approval = ApprovalRequest( external_id=instance_code, approval_type=flow.flow_key, applicant_id=requester.id, status="pending", source_payload=_approval_source_payload( request_id=request_row.id, payload=payload, form_payload=form_payload, instance=instance, approver_payload=approver_payload, ), ) db.add(approval) await db.flush() request_row.feishu_instance_code = instance_code request_row.approval_request_id = approval.id except Exception: if instance_code: try: cancel_kwargs = { "instance_code": instance_code, "approval_code": payload.external_code, "open_id" if feishu_user_id_type == "open_id" else "user_id": str(feishu_user_id), } await client.cancel_instance(**cancel_kwargs) except Exception: # noqa: BLE001 - compensation is best effort logger.warning("data-access service-account approval cancellation failed", exc_info=True) raise audit_trace_id = clean_trace_id or f"data_access_request:{request_row.id}" await AuditService(db).log( trace_id=audit_trace_id, actor_type="user", actor_user_id=requester.id, action="data_access.service_account_request.submit", resource_type="data_access_request", resource_id=str(request_row.id), status="success", detail={ "principal_id": str(principal.principal_id), "principal_subject": principal.subject_snapshot, "approval_flow_key": flow.flow_key, "principal_recorded_from": "self_service_input", "dataset_count": len(resources), "pre_approval_preview_only": True, }, ) await db.flush() return await self._submission_result_from_request( db, request_row, dataset_count=len(resources), ) async def submit_personal_grant_request( self, db: AsyncSession, *, requester: IdentityPerson, resource_ids: Sequence[uuid.UUID], business_reason: str, detail_url: str, idempotency_key: str, trace_id: str | None = None, feishu_client: Any | None = None, ) -> DataAccessRequestSubmissionResult: """Submit a direct-user request derived from the authenticated employee.""" if requester.employment_status != "active": raise DataAccessSubmissionError( "personal_requester_inactive", "Only active employees can submit personal data-access requests.", field="requester", status_code=422, ) personal_email = _normalise_personal_work_email(requester.work_email) clean_key = _clean_optional_single_line( idempotency_key, field="idempotency_key", max_length=DATA_ACCESS_IDEMPOTENCY_KEY_MAX_LENGTH, ) if not clean_key: raise DataAccessSubmissionError( "idempotency_key_required", "Idempotency-Key is required for personal grants.", field="idempotency_key", status_code=422, ) clean_trace_id = _clean_optional_single_line( trace_id, field="trace_id", max_length=DATA_ACCESS_TRACE_ID_MAX_LENGTH, ) existing = await self._existing_request_for_idempotency_key( db, clean_key, requester_person_id=requester.id, ) if existing is not None: if ( existing.principal_type_snapshot != c.PRINCIPAL_TYPE_GOOGLE_USER or existing.principal_email_snapshot != personal_email or existing.principal_subject_snapshot != str(requester.id) or not await self._request_matches_submission_fingerprint( db, request=existing, resource_ids=resource_ids, business_reason=business_reason, ) ): raise DataAccessSubmissionError( "idempotency_key_reused", "Idempotency-Key was used for a different request.", field="idempotency_key", status_code=409, ) return await self._submission_result_from_request( db, existing, reused_existing=True, ) if not resource_ids: raise DataAccessSubmissionError( "empty_dataset_selection", "At least one BigQuery dataset is required.", field="resource_ids", ) flow = await resolve_data_access_personal_grant_flow(session=db) resources = [ await self.validate_requestable_bigquery_dataset_resource( db, resource_id, require_resource_owner_authority=True, ) for resource_id in resource_ids ] principal = await self.record_personal_principal(db, requester=requester) entitlement = await self._load_read_v1_entitlement(db) payload = self.build_feishu_personal_submission_payload( flow=flow, resources=resources, principal=principal, entitlement=entitlement, business_reason=business_reason, detail_url=detail_url, ) form_payload = self.build_feishu_submission_form(flow=flow, payload=payload) client = feishu_client or await self._data_access_feishu_client(db) feishu_user_id_type, feishu_user_id = await self._resolve_feishu_identity_by_email( client, requester.work_email, field="requester", subject_label="Requester", ) approver_payload = await self._build_feishu_data_owner_approver_payload( db, client=client, flow=flow, resources=resources, ) request_row = DataAccessRequest( requester_person_id=requester.id, principal_id=principal.principal_id, principal_type_snapshot=c.PRINCIPAL_TYPE_GOOGLE_USER, principal_subject_snapshot=principal.subject_snapshot, principal_email_snapshot=principal.email_snapshot, entitlement_id=entitlement.id, entitlement_snapshot=c.ENTITLEMENT_READ, entitlement_version_snapshot=ENTITLEMENT_VERSION_V1, approval_flow_key=flow.flow_key, idempotency_key=clean_key, business_reason=payload.business_reason, status=REQUEST_STATUS_PENDING_FEISHU, trace_id=clean_trace_id, items=[ DataAccessRequestItem( resource_id=resource.resource_id, line_no=line_no, resource_project_id_snapshot=resource.project_id, resource_dataset_id_snapshot=resource.dataset_id, resource_source_version_snapshot=resource.source_version, resource_source_etag_snapshot=resource.source_etag, resource_owner_person_id_snapshot=_uuid_snapshot( resource.owner_person_id, field="resource_owner_person_id", ), status=REQUEST_ITEM_STATUS_PENDING, ) for line_no, resource in enumerate(resources, start=1) ], ) try: async with db.begin_nested(): db.add(request_row) await db.flush() except IntegrityError: existing = await self._existing_request_for_idempotency_key( db, clean_key, requester_person_id=requester.id, lock=True, ) if existing is not None: if ( existing.principal_type_snapshot != c.PRINCIPAL_TYPE_GOOGLE_USER or existing.principal_email_snapshot != personal_email or existing.principal_subject_snapshot != str(requester.id) or not await self._request_matches_submission_fingerprint( db, request=existing, resource_ids=resource_ids, business_reason=business_reason, ) ): raise DataAccessSubmissionError( "idempotency_key_reused", "Idempotency-Key was used for a different request.", field="idempotency_key", status_code=409, ) return await self._submission_result_from_request( db, existing, dataset_count=len(resources), reused_existing=True, ) raise instance_code: str | None = None try: create_kwargs: dict[str, Any] = { "approval_code": payload.external_code, "form": form_payload, **approver_payload, } create_kwargs["open_id" if feishu_user_id_type == "open_id" else "user_id"] = str(feishu_user_id) instance = await client.create_instance( instance_uuid=str(request_row.id), **create_kwargs, ) instance_code = str(instance.get("instance_code") or "").strip() if not instance_code: raise DataAccessSubmissionError( "missing_feishu_instance_code", "Feishu approval response did not include a code.", field="feishu", status_code=503, ) approval = ApprovalRequest( external_id=instance_code, approval_type=flow.flow_key, applicant_id=requester.id, status="pending", source_payload=_approval_source_payload( request_id=request_row.id, payload=payload, form_payload=form_payload, instance=instance, approver_payload=approver_payload, ), ) db.add(approval) await db.flush() request_row.feishu_instance_code = instance_code request_row.approval_request_id = approval.id except Exception: if instance_code: try: cancel_kwargs = { "instance_code": instance_code, "approval_code": payload.external_code, "open_id" if feishu_user_id_type == "open_id" else "user_id": str(feishu_user_id), } await client.cancel_instance(**cancel_kwargs) except Exception: # noqa: BLE001 - compensation is best effort logger.warning("data-access personal approval cancellation failed", exc_info=True) raise audit_trace_id = clean_trace_id or f"data_access_request:{request_row.id}" await AuditService(db).log( trace_id=audit_trace_id, actor_type="user", actor_user_id=requester.id, action="data_access.personal_request.submit", resource_type="data_access_request", resource_id=str(request_row.id), status="success", detail={ "principal_id": str(principal.principal_id), "principal_subject": principal.subject_snapshot, "approval_flow_key": flow.flow_key, "principal_recorded_from": "authenticated_identity", "dataset_count": len(resources), "pre_approval_preview_only": True, }, ) await db.flush() return await self._submission_result_from_request( db, request_row, dataset_count=len(resources), ) async def submit_group_grant_request( self, db: AsyncSession, *, requester: IdentityPerson, principal: uuid.UUID | str, resource_ids: Sequence[uuid.UUID], business_reason: str, detail_url: str, principal_type: str = c.PRINCIPAL_TYPE_GOOGLE_GROUP, entitlement: str = c.ENTITLEMENT_READ, entitlement_version: str = ENTITLEMENT_VERSION_V1, idempotency_key: str | None = None, trace_id: str | None = None, feishu_client: Any | None = None, ) -> DataAccessRequestSubmissionResult: """Create the stateful DAX request and Feishu approval instance. The method stops at request + approval state. It intentionally does not create bindings/effects and never calls a BigQuery provider. """ clean_idempotency_key = _clean_optional_single_line( idempotency_key, field="idempotency_key", max_length=DATA_ACCESS_IDEMPOTENCY_KEY_MAX_LENGTH, ) clean_trace_id = _clean_optional_single_line( trace_id, field="trace_id", max_length=DATA_ACCESS_TRACE_ID_MAX_LENGTH, ) existing = await self._existing_request_for_idempotency_key( db, clean_idempotency_key, requester_person_id=requester.id, ) if existing is not None: if existing.requester_person_id != requester.id: raise DataAccessSubmissionError( "idempotency_key_conflict", "Idempotency key belongs to a different requester.", field="idempotency_key", status_code=409, ) if existing.feishu_instance_code is None: await self._resume_group_grant_submission( db, existing, feishu_client=feishu_client, ) return await self._submission_result_from_request( db, existing, reused_existing=True, ) try: c.assert_canonical("entitlement", entitlement) except ValueError as exc: raise DataAccessSubmissionError( "unsupported_entitlement_input", str(exc), field="entitlement", ) from exc if entitlement_version != ENTITLEMENT_VERSION_V1: raise DataAccessSubmissionError( "unsupported_entitlement_version", "Data-access V1 only supports read:v1 submissions.", field="entitlement_version", ) flow = await resolve_data_access_group_grant_flow(session=db) principal_snapshot = await self.validate_google_group_principal( db, principal, principal_type=principal_type, allow_deleted_rename_bridge=False, ) requester_authorization = await self.authorize_requester_for_google_group_principal( db, requester, principal_snapshot, ) if not resource_ids: raise DataAccessSubmissionError( "empty_dataset_selection", "At least one BigQuery dataset is required.", field="resource_ids", ) resources = [ await self.validate_requestable_bigquery_dataset_resource( db, resource_id, require_resource_owner_authority=True, ) for resource_id in resource_ids ] entitlement_row = await self._load_read_v1_entitlement(db) payload = self.build_feishu_submission_payload( flow=flow, resources=resources, principal=principal_snapshot, entitlement=entitlement_row, business_reason=business_reason, detail_url=detail_url, ) form_payload = self.build_feishu_submission_form( flow=flow, payload=payload, ) client = feishu_client or await self._data_access_feishu_client(db) feishu_user_id_type, feishu_user_id = await self._resolve_feishu_identity_by_email( client, requester.work_email, field="requester", subject_label="Requester", ) feishu_approver_payload = await self._build_feishu_data_owner_approver_payload( db, client=client, flow=flow, resources=resources, ) feishu_group_owner_cc_payload = ( await self._build_feishu_group_owner_cc_payload( db, client=client, principal=principal_snapshot, requester=requester, actor_id_type=feishu_user_id_type, ) if requester_authorization.authorized_as == "group_member" else {} ) request_row = DataAccessRequest( requester_person_id=requester.id, principal_email_group_id=principal_snapshot.principal_email_group_id, principal_google_group_id_snapshot=principal_snapshot.google_group_id, principal_email_snapshot=principal_snapshot.primary_email_snapshot, entitlement_id=entitlement_row.id, entitlement_snapshot=c.ENTITLEMENT_READ, entitlement_version_snapshot=ENTITLEMENT_VERSION_V1, approval_flow_key=flow.flow_key, idempotency_key=clean_idempotency_key, business_reason=payload.business_reason, status=REQUEST_STATUS_PENDING_FEISHU, trace_id=clean_trace_id, items=[ DataAccessRequestItem( resource_id=resource.resource_id, line_no=line_no, resource_project_id_snapshot=resource.project_id, resource_dataset_id_snapshot=resource.dataset_id, resource_source_version_snapshot=resource.source_version, resource_source_etag_snapshot=resource.source_etag, resource_owner_person_id_snapshot=_uuid_snapshot( resource.owner_person_id, field="resource_owner_person_id", ), status=REQUEST_ITEM_STATUS_PENDING, ) for line_no, resource in enumerate(resources, start=1) ], ) try: # ``add`` + ``flush`` MUST happen inside the savepoint. If the # row is added before ``begin_nested``, its pre-flush INSERTs it # *before* the SAVEPOINT is established, so a unique collision # poisons the whole outer transaction (the recovery SELECT below # would then raise PendingRollbackError, not replay). Adding # inside confines the collision to the savepoint. async with db.begin_nested(): db.add(request_row) await db.flush() except IntegrityError: # Idempotency-key unique collision from a concurrent same-key # submit that inserted between our earlier existence check and # this flush. The constraint kept it fail-closed (no dup); turn # the loser into the same replay the pre-check would have # returned, not an unhandled 500. existing = await self._existing_request_for_idempotency_key( db, clean_idempotency_key, requester_person_id=requester.id, lock=True, ) if existing is not None and existing.requester_person_id == requester.id: # dataset_count from the resources we just tried to submit: # the winner is the same logical request (same idempotency # key), and a consistent-read count of its items would be # stale under this txn's snapshot. return await self._submission_result_from_request( db, existing, dataset_count=len(resources), reused_existing=True, ) raise approval_request_id = uuid.uuid4() cc_result: dict[str, Any] | None = None try: create_kwargs: dict[str, Any] = { "approval_code": payload.external_code, "form": form_payload, **feishu_approver_payload, } if feishu_user_id_type == "open_id": create_kwargs["open_id"] = str(feishu_user_id) else: create_kwargs["user_id"] = str(feishu_user_id) source_payload = _approval_source_payload( request_id=request_row.id, payload=payload, form_payload=form_payload, instance={}, approver_payload=feishu_approver_payload, cc_payload=feishu_group_owner_cc_payload, ) provider_payload = { "approval_code": payload.external_code, "identity_type": feishu_user_id_type, "identity_value": str(feishu_user_id), "form": form_payload, **feishu_approver_payload, } intent = await submit_with_durable_intent( db, client=client, scope=flow.flow_key, applicant_id=requester.id, intent_key=str(request_row.id), semantic_payload={ "request_id": str(request_row.id), "idempotency_key": clean_idempotency_key, "principal_email_group_id": str(principal_snapshot.principal_email_group_id), "resource_ids": [str(item.resource_id) for item in resources], "business_reason": payload.business_reason, }, provider_payload=provider_payload, finalization_payload={ "request_id": str(request_row.id), "approval_request_id": str(approval_request_id), "approval_source_payload": source_payload, "dataset_count": len(resources), "requester_authorized_as": requester_authorization.authorized_as, "group_owner_cc_count": len( feishu_group_owner_cc_payload.get("cc_user_ids", []) ), }, trace_id=clean_trace_id or f"data_access_request:{request_row.id}", ) instance_code = str(intent.external_id or "") if feishu_group_owner_cc_payload: # CC-ing the group owner is a courtesy FYI, NOT part of the # approval chain — a transient CC failure (or a client that # can't CC) must never cancel an otherwise-valid member # submission and force a full resubmit. Best-effort: record # the outcome in the source payload and carry on. cc_kwargs: dict[str, Any] = { "approval_code": payload.external_code, "instance_code": instance_code, "cc_user_ids": feishu_group_owner_cc_payload["cc_user_ids"], "comment": feishu_group_owner_cc_payload["comment"], } if feishu_user_id_type == "open_id": cc_kwargs["open_id"] = str(feishu_user_id) else: cc_kwargs["user_id"] = str(feishu_user_id) try: if not hasattr(client, "cc_instance"): raise RuntimeError("client does not support cc_instance") cc_result = await client.cc_instance(**cc_kwargs) except Exception as cc_exc: # noqa: BLE001 logger.warning( "data-access group-owner CC failed for request=%s " "(instance=%s); submission proceeds", request_row.id, instance_code, exc_info=True, ) cc_result = { "status": "failed", "error": sanitize_log_text(cc_exc, max_len=512), } await self._finalize_group_grant_submission( db, intent.id, recovered=intent.attempt_count > 1, cc_result=cc_result, ) except AppException as exc: if exc.code not in { "APPROVAL_SUBMISSION_RESULT_UNKNOWN", "APPROVAL_SUBMISSION_IN_PROGRESS", }: await self._mark_group_grant_submission_rejected( db, request_row.id, ) raise return await self._submission_result_from_request( db, request_row, dataset_count=len(resources), ) async def _resume_group_grant_submission( self, db: AsyncSession, request: DataAccessRequest, *, feishu_client: Any | None = None, ) -> None: intent = await db.scalar( select(ApprovalSubmissionIntent).where( ApprovalSubmissionIntent.scope == request.approval_flow_key, ApprovalSubmissionIntent.applicant_id == request.requester_person_id, ApprovalSubmissionIntent.idempotency_key == str(request.id), ) ) if intent is None: raise DataAccessSubmissionError( "missing_submission_intent", "This request has no durable Feishu submission intent.", field="feishu", status_code=409, ) stored = dict(intent.payload or {}) client = feishu_client or await self._data_access_feishu_client(db) try: intent = await submit_with_durable_intent( db, client=client, scope=intent.scope, applicant_id=intent.applicant_id, intent_key=intent.idempotency_key, semantic_payload=dict(stored.get("semantic") or {}), provider_payload=dict(stored.get("provider") or {}), finalization_payload=dict(stored.get("finalization") or {}), trace_id=str(stored.get("trace_id") or f"data-access:{request.id}"), ) await self._finalize_group_grant_submission( db, intent.id, recovered=True, ) except AppException as exc: if exc.code not in { "APPROVAL_SUBMISSION_RESULT_UNKNOWN", "APPROVAL_SUBMISSION_IN_PROGRESS", }: await self._mark_group_grant_submission_rejected(db, request.id) raise async def _mark_group_grant_submission_rejected( self, db: AsyncSession, request_id: uuid.UUID, ) -> None: """Close the request and every item after a definite refusal.""" await db.rollback() failed_request = await db.scalar( select(DataAccessRequest) .where(DataAccessRequest.id == request_id) .options(selectinload(DataAccessRequest.items)) ) if failed_request is not None: failed_request.status = REQUEST_STATUS_REJECTED for item in failed_request.items: item.status = REQUEST_ITEM_STATUS_REJECTED await db.commit() async def _finalize_group_grant_submission( self, db: AsyncSession, intent_id: uuid.UUID, *, recovered: bool, cc_result: dict[str, Any] | None = None, ) -> None: intent = await db.scalar( select(ApprovalSubmissionIntent) .where(ApprovalSubmissionIntent.id == intent_id) .with_for_update() .execution_options(populate_existing=True) ) if intent is None: raise RuntimeError("data-access submission intent disappeared") if intent.state == "completed": return final = dict((intent.payload or {}).get("finalization") or {}) request_id = uuid.UUID(str(final["request_id"])) request = await db.get(DataAccessRequest, request_id) if request is None: raise RuntimeError("data-access intent request disappeared") if intent.state == "completed": return if intent.state != "provider_created" or not intent.external_id: raise RuntimeError(f"cannot finalize data-access intent in {intent.state!r}") response = dict((intent.payload or {}).get("provider_response") or {}) source_payload = dict(final["approval_source_payload"]) feishu = dict(source_payload.get("feishu") or {}) feishu["instance_code"] = intent.external_id feishu["instance_status"] = response.get("instance_status", "PENDING") if cc_result: feishu["cc_result"] = cc_result source_payload["feishu"] = feishu approval_id = uuid.UUID(str(final["approval_request_id"])) approval = await db.get(ApprovalRequest, approval_id) if approval is None: approval = ApprovalRequest( id=approval_id, external_id=intent.external_id, approval_type=intent.scope, applicant_id=intent.applicant_id, status="pending", source_payload=source_payload, ) db.add(approval) await db.flush() request.feishu_instance_code = intent.external_id request.approval_request_id = approval.id trace_id = str( (intent.payload or {}).get("trace_id") or f"data_access_request:{request.id}" ) compiled = source_payload.get("compiled_preview") or [] await self._audit_compile_output( db, trace_id=(f"approval-intent-reconcile:{intent.id}" if recovered else trace_id), actor_type="system" if recovered else "user", actor_user_id=None if recovered else intent.applicant_id, resource_type="data_access_request", resource_id=str(request.id), phase="submission_preview", effects=[SimpleNamespace(**effect) for effect in compiled], request_id=request.id, detail_extra={ "approval_flow_key": intent.scope, "approval_request_id": str(approval.id), "feishu_instance_code": intent.external_id, "requester_authorized_as": final["requester_authorized_as"], "group_owner_cc_count": final["group_owner_cc_count"], "pre_approval_preview_only": True, "recovered": recovered, }, ) await AuditService(db).log( trace_id=(f"approval-intent-reconcile:{intent.id}" if recovered else trace_id), actor_type="system" if recovered else "user", actor_user_id=None if recovered else intent.applicant_id, action=( "approval.submission_intent.reconciled" if recovered else "data_access.request.submit" ), resource_type="data_access_request", resource_id=str(request.id), status="success", detail={ "approval_flow_key": intent.scope, "feishu_instance_code": intent.external_id, "dataset_count": final["dataset_count"], "idempotency_key": request.idempotency_key, "requester_authorized_as": final["requester_authorized_as"], "group_owner_cc_count": final["group_owner_cc_count"], "provider_request_uuid": intent.provider_request_uuid, "recovered": recovered, }, ) intent.approval_request_id = approval.id intent.state = "completed" intent.completed_at = datetime.now(UTC) intent.last_error = None compact_terminal_intent_payload( intent, resource_type="data_access_request", resource_id=request.id, ) await db.commit() async def reconcile_submission_intents( self, db: AsyncSession, *, min_age_minutes: int = 2, batch_limit: int = 100, feishu_client: Any | None = None, ) -> dict[str, int]: cutoff = datetime.now(UTC) - timedelta(minutes=min_age_minutes) ids = list( ( await db.scalars( select(ApprovalSubmissionIntent.id) .where( ApprovalSubmissionIntent.scope == c.APPROVAL_FLOW_DATA_ACCESS_GROUP_GRANT, ApprovalSubmissionIntent.state.in_( ( "prepared", "submitting", "unknown", "provider_created", ) ), (ApprovalSubmissionIntent.state == "provider_created") | (ApprovalSubmissionIntent.updated_at < cutoff), reconciliation_due_predicate(datetime.now(UTC)), ) .order_by(*reconciliation_order()) .limit(batch_limit) ) ).all() ) summary = {"checked": len(ids), "reconciled": 0, "still_unknown": 0, "failed": 0} for intent_id in ids: try: intent = await db.get(ApprovalSubmissionIntent, intent_id) if intent is None: continue request = await db.get(DataAccessRequest, uuid.UUID(intent.idempotency_key)) if request is None: raise RuntimeError("data-access intent has no request") await self._resume_group_grant_submission( db, request, feishu_client=feishu_client, ) summary["reconciled"] += 1 except AppException as exc: if exc.code == "APPROVAL_SUBMISSION_RESULT_UNKNOWN": await db.rollback() else: await defer_submission_intent_retry(db, intent_id, exc) summary[ "still_unknown" if exc.code == "APPROVAL_SUBMISSION_RESULT_UNKNOWN" else "failed" ] += 1 except Exception as exc: # noqa: BLE001 await defer_submission_intent_retry(db, intent_id, exc) summary["failed"] += 1 return summary async def _existing_request_for_idempotency_key( self, db: AsyncSession, idempotency_key: str | None, *, requester_person_id: uuid.UUID, lock: bool = False, ) -> DataAccessRequest | None: if not idempotency_key: return None stmt = select(DataAccessRequest).where( DataAccessRequest.idempotency_key == idempotency_key, DataAccessRequest.requester_person_id == requester_person_id, ) if lock: # Locking read: the caller is recovering from a unique-collision # in its own transaction, whose REPEATABLE READ snapshot predates # the concurrent winner's commit. A consistent read would not see # the winner (same class as the ensure_run fix); with_for_update # reads the latest committed row. stmt = stmt.with_for_update().execution_options( populate_existing=True, ) return await db.scalar(stmt) async def _request_matches_submission_fingerprint( self, db: AsyncSession, *, request: DataAccessRequest, resource_ids: Sequence[uuid.UUID], business_reason: str, ) -> bool: existing_resource_ids = set((await db.scalars( select(DataAccessRequestItem.resource_id).where( DataAccessRequestItem.request_id == request.id, ), )).all()) requested_resource_ids = set(resource_ids) try: requested_reason = _clean_required_multiline( business_reason, field="business_reason", max_length=DATA_ACCESS_BUSINESS_REASON_MAX_LENGTH, ) except DataAccessFeishuPayloadError: return False return ( existing_resource_ids == requested_resource_ids and request.business_reason == requested_reason ) async def _submission_result_from_request( self, db: AsyncSession, request: DataAccessRequest, *, dataset_count: int | None = None, reused_existing: bool = False, ) -> DataAccessRequestSubmissionResult: if dataset_count is None: dataset_count = int( await db.scalar( select(func.count()) .select_from(DataAccessRequestItem) .where(DataAccessRequestItem.request_id == request.id), ) or 0, ) return DataAccessRequestSubmissionResult( request_id=request.id, approval_request_id=request.approval_request_id, feishu_instance_code=request.feishu_instance_code, reused_existing=reused_existing, status=request.status, idempotency_key=request.idempotency_key, trace_id=request.trace_id, dataset_count=dataset_count, ) async def handle_feishu_callback( self, db: AsyncSession, *, approval_request: ApprovalRequest, instance_detail: dict[str, Any], approval_status: str, trace_id: str | None = None, ) -> DataAccessRequestCallbackResult: """Apply a Feishu callback to DAX request/binding/effect state.""" source_payload = self._callback_source_payload( approval_request, instance_detail, ) callback_trace_id = trace_id or (f"data_access_callback:{approval_request.external_id}") request = await self._callback_request_for_approval( db, approval_request=approval_request, source_payload=source_payload, ) await self._audit_callback_received( db, approval_request=approval_request, request=request, instance_detail=instance_detail, source_payload=source_payload, approval_status=approval_status, trace_id=callback_trace_id, ) if request is None: return await self._refuse_callback( db, approval_request=approval_request, request=None, items=[], source_payload=source_payload, approval_status=approval_status, reason="request_not_found", trace_id=callback_trace_id, detail={"approval_request_id": str(approval_request.id)}, ) items = await self._callback_request_items(db, request) if approval_status == "pending": source_payload["data_access_callback_outcome"] = { "status": "pending", "request_id": str(request.id), } approval_request.status = "pending" approval_request.source_payload = source_payload approval_request.updated_at = datetime.now(UTC) await db.flush() return DataAccessRequestCallbackResult( request_id=request.id, approval_request_id=approval_request.id, approval_status=approval_status, request_status=request.status, binding_count=0, effect_count=0, ) if request.status != REQUEST_STATUS_PENDING_FEISHU: source_payload["data_access_callback_outcome"] = { "status": "idempotent", "request_id": str(request.id), "request_status": request.status, "approval_status": approval_status, } approval_request.status = approval_status approval_request.source_payload = source_payload approval_request.updated_at = datetime.now(UTC) binding_count, effect_count = await self._request_effect_counts( db, request, ) await db.flush() return DataAccessRequestCallbackResult( request_id=request.id, approval_request_id=approval_request.id, approval_status=approval_status, request_status=request.status, binding_count=binding_count, effect_count=effect_count, ) mismatch_reason = self._callback_binding_mismatch( approval_request=approval_request, request=request, source_payload=source_payload, instance_detail=instance_detail, ) if mismatch_reason is not None: return await self._refuse_callback( db, approval_request=approval_request, request=request, items=items, source_payload=source_payload, approval_status=approval_status, reason=mismatch_reason, trace_id=callback_trace_id, detail={"feishu_instance_code": request.feishu_instance_code}, ) if approval_status != "approved": return await self._settle_non_approved_callback( db, approval_request=approval_request, request=request, items=items, source_payload=source_payload, approval_status=approval_status, trace_id=callback_trace_id, ) compiled = await self._compile_approved_callback_effects( db, request=request, items=items, source_payload=source_payload, ) if compiled["refusal_reason"] is not None: return await self._refuse_callback( db, approval_request=approval_request, request=request, items=items, source_payload=source_payload, approval_status=approval_status, reason=str(compiled["refusal_reason"]), trace_id=callback_trace_id, detail=dict(compiled["refusal_detail"]), ) now = datetime.now(UTC) binding_count = 0 effect_count = 0 created_bindings: list[DataAccessBinding] = [] created_effects: list[DataAccessCompiledEffect] = [] # Lifecycle clock (migration 606): personal grants get ``expires_at``, # service-account grants their first owner-confirmation due date. from app.domains.data_access.lifecycle import ( # noqa: PLC0415 load_lifecycle_settings, stamp_new_binding, ) lifecycle_settings = await load_lifecycle_settings(db) for item, effect in compiled["effects"]: binding = DataAccessBinding( resource_id=effect.resource_id, principal_email_group_id=effect.principal_email_group_id, principal_id=effect.principal_id, principal_type_snapshot=effect.principal_type_snapshot, principal_subject_snapshot=effect.principal_subject_snapshot, principal_google_group_id_snapshot=( effect.principal_google_group_id_snapshot ), principal_email_snapshot=effect.principal_email_snapshot, entitlement_id=effect.entitlement_id, entitlement_snapshot=c.ENTITLEMENT_READ, entitlement_version_snapshot=effect.entitlement_version_snapshot, source_request_id=request.id, source_request_item_id=item.id, status=BINDING_STATUS_ACTIVE, ) stamp_new_binding(binding, settings=lifecycle_settings, now=now) db.add(binding) await db.flush() compiled_row = DataAccessCompiledEffect( binding_id=binding.id, resource_id=effect.resource_id, principal_email_group_id=effect.principal_email_group_id, principal_id=effect.principal_id, principal_type_snapshot=effect.principal_type_snapshot, principal_subject_snapshot=effect.principal_subject_snapshot, entitlement_id=effect.entitlement_id, provider=effect.provider, resource_type=effect.resource_type, target_project_id=effect.target_project_id, target_dataset_id=effect.target_dataset_id, target_table_id=effect.target_table_id, provider_member=effect.provider_member, provider_access_kind=effect.provider_access_kind, provider_role=effect.provider_role, desired_state=effect.desired_state, entitlement_version_snapshot=(effect.entitlement_version_snapshot), resource_source_version_snapshot=(effect.resource_source_version_snapshot), resource_source_etag_snapshot=(effect.resource_source_etag_snapshot), compiler_version=effect.compiler_version, managed_reason=effect.managed_reason, already_present_at_first_apply=(effect.already_present_at_first_apply), status=EFFECT_STATUS_PENDING_RECONCILE, ) db.add(compiled_row) item.status = REQUEST_ITEM_STATUS_APPROVED item.staleness_reason = None created_bindings.append(binding) created_effects.append(compiled_row) binding_count += 1 effect_count += 1 request.status = REQUEST_STATUS_APPROVED request.decided_at = now request.updated_at = now approval_request.status = "approved" source_payload["data_access_callback_outcome"] = { "status": "approved", "request_id": str(request.id), "binding_count": binding_count, "effect_count": effect_count, "compiler_version": DATA_ACCESS_COMPILER_VERSION, } approval_request.source_payload = source_payload approval_request.updated_at = now if created_effects: request.reconcile_status = RECONCILE_STATUS_PENDING request.last_reconcile_error = None await db.flush() for binding in created_bindings: await self._audit_binding_created( db, trace_id=callback_trace_id, request=request, approval_request=approval_request, binding=binding, ) await self._audit_compile_output( db, trace_id=callback_trace_id, actor_type="system", actor_user_id=request.requester_person_id, resource_type="data_access_request", resource_id=str(request.id), phase="approval_callback", effects=created_effects, request_id=request.id, detail_extra={ "approval_request_id": str(approval_request.id), "feishu_instance_code": request.feishu_instance_code, "binding_count": binding_count, }, ) await AuditService(db).log( trace_id=callback_trace_id, actor_type="system", actor_user_id=request.requester_person_id, action="data_access.request.approved", resource_type="data_access_request", resource_id=str(request.id), status="success", detail={ "approval_request_id": str(approval_request.id), "feishu_instance_code": request.feishu_instance_code, "binding_count": binding_count, "effect_count": effect_count, }, ) await db.flush() return DataAccessRequestCallbackResult( request_id=request.id, approval_request_id=approval_request.id, approval_status=approval_status, request_status=request.status, binding_count=binding_count, effect_count=effect_count, ) async def revoke_data_access_binding( self, db: AsyncSession, *, actor: IdentityPerson | None, binding_id: uuid.UUID, reason: str, trace_id: str | None = None, system_context: bool = False, system_actor_user_id: uuid.UUID | None = None, ) -> DataAccessRevokeResult: """Create authorized semantic revoke intent without Feishu/provider IO. ``system_context=True`` (migration 606) is the policy-driven caller — offboarding T+0, expiry, owner-confirmation lapse. It skips the actor authorisation (the caller IS the policy) and audits as ``system``; everything downstream (semantic flip → absent effect → reconciler) is identical to a human revoke, so BigQuery still has exactly one writer. """ binding = await self._binding_for_update(db, binding_id) if binding is None: raise AppException( code="DATA_ACCESS_BINDING_NOT_FOUND", message="Data-access binding was not found.", status_code=404, field="binding_id", ) resource = await self._binding_resource(db, binding) if system_context: authorized_as = "system" actor_type = "system" actor_user_id: uuid.UUID | None = system_actor_user_id else: if actor is None: raise AppException( code="DATA_ACCESS_REVOKE_ACTOR_REQUIRED", message="An acting person is required to revoke a data-access binding.", status_code=400, field="actor", ) authorized_as = await self._authorize_revoke_actor( db, actor=actor, binding=binding, resource=resource, ) actor_type = "user" actor_user_id = actor.id if binding.status == BINDING_STATUS_REVOKED_PENDING: return DataAccessRevokeResult( status=BINDING_STATUS_REVOKED_PENDING, reason=binding.revoke_reason, authorized_as=authorized_as, already_pending_count=1, ) if binding.status == BINDING_STATUS_REVOKED: return DataAccessRevokeResult( status=BINDING_STATUS_REVOKED, reason=binding.revoke_reason, authorized_as=authorized_as, already_revoked_count=1, ) if binding.status != BINDING_STATUS_ACTIVE: return DataAccessRevokeResult( status=binding.status, reason=binding.revoke_reason, authorized_as=authorized_as, ) entitlement = await self._binding_entitlement(db, binding) reference_effect = await self._grant_effect_snapshot_for_binding( db, binding, ) clean_reason = _clean_revoke_reason(reason) self._mark_binding_revoked_pending(binding, reason=clean_reason) effect = self._compiled_effect_from_binding( binding=binding, resource=resource, entitlement=entitlement, desired_state=EFFECT_DESIRED_ABSENT, managed_reason=EFFECT_MANAGED_REASON_REVOKE_REQUEST, reference_effect=reference_effect, ) db.add(effect) await db.flush() await self._audit_compile_output( db, trace_id=trace_id or f"data_access_revoke:{binding.id}", actor_type=actor_type, actor_user_id=actor_user_id, resource_type="data_access_binding", resource_id=str(binding.id), phase="revoke_request", effects=[effect], request_id=binding.source_request_id, detail_extra={ "binding_id": str(binding.id), "authorized_as": authorized_as, "reason": clean_reason, }, ) await AuditService(db).log( trace_id=trace_id or f"data_access_revoke:{binding.id}", actor_type=actor_type, actor_user_id=actor_user_id, action="data_access.binding.revoke_requested", status="success", resource_type="data_access_binding", resource_id=str(binding.id), detail={ "binding_id": str(binding.id), "resource_id": str(binding.resource_id), "principal_email_group_id": str( binding.principal_email_group_id, ), "entitlement_id": str(binding.entitlement_id), "reason": clean_reason, "authorized_as": authorized_as, "effect_count": 1, }, ) await db.flush() return DataAccessRevokeResult( status=BINDING_STATUS_REVOKED_PENDING, reason=clean_reason, authorized_as=authorized_as, revoked_pending_count=1, effect_count=1, ) async def revoke_deleted_service_account_bindings( self, db: AsyncSession, *, principal_id: uuid.UUID, trace_id: str | None = None, ) -> DataAccessRevokeResult: """Compile absent effects after trusted catalog deletion confirmation. This is a system-owned desired-state transition. It is deliberately separate from catalog omission/suspension and does not call BigQuery. """ principal = await db.get(DataAccessPrincipal, principal_id) if ( principal is None or principal.principal_type != c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT or principal.status != "deleted_confirmed" ): raise AppException( code="DATA_ACCESS_SERVICE_ACCOUNT_NOT_DELETED_CONFIRMED", message="Service-account principal is not deletion-confirmed.", status_code=409, field="principal_id", ) bindings = ( await db.scalars( select(DataAccessBinding) .where( DataAccessBinding.principal_id == principal.id, DataAccessBinding.principal_type_snapshot == c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT, DataAccessBinding.status == BINDING_STATUS_ACTIVE, ) .with_for_update(), ) ).all() if not bindings: return DataAccessRevokeResult( status="noop", reason="service_account_deleted_confirmed", authorized_as="system", ) effects: list[DataAccessCompiledEffect] = [] for binding in bindings: resource = await self._binding_resource(db, binding) entitlement = await self._binding_entitlement(db, binding) reference_effect = await self._grant_effect_snapshot_for_binding( db, binding, ) self._mark_binding_revoked_pending( binding, reason="service_account_deleted_confirmed", ) effect = self._compiled_effect_from_binding( binding=binding, resource=resource, entitlement=entitlement, desired_state=EFFECT_DESIRED_ABSENT, managed_reason=EFFECT_MANAGED_REASON_REVOKE_REQUEST, reference_effect=reference_effect, ) db.add(effect) effects.append(effect) revoke_trace_id = trace_id or f"data_access_service_account_deleted:{principal.id}" await db.flush() await self._audit_compile_output( db, trace_id=revoke_trace_id, actor_type="system", actor_user_id=None, resource_type="data_access_principal", resource_id=str(principal.id), phase="service_account_deleted_confirmed", effects=effects, request_id=None, detail_extra={ "principal_id": str(principal.id), "principal_subject_snapshot": principal.external_subject_id, "revoked_pending_count": len(bindings), "provider_mutation_attempted": False, }, ) await AuditService(db).log( trace_id=revoke_trace_id, actor_type="system", actor_user_id=None, action="data_access.service_account.deleted_confirmed", status="success", resource_type="data_access_principal", resource_id=str(principal.id), detail={ "principal_id": str(principal.id), "principal_subject_snapshot": principal.external_subject_id, "revoked_pending_count": len(bindings), "effect_count": len(effects), "provider_mutation_attempted": False, }, ) await db.flush() return DataAccessRevokeResult( status=BINDING_STATUS_REVOKED_PENDING, reason="service_account_deleted_confirmed", authorized_as="system", revoked_pending_count=len(bindings), effect_count=len(effects), ) async def revoke_stale_personal_bindings( self, db: AsyncSession, *, trace_id: str | None = None, batch_size: int = 500, ) -> DataAccessRevokeResult: """Turn invalid direct-user bindings into reconciler-owned revokes. Personal DAX effects are valid only while the referenced identity is an active employee with the same work-email snapshot. This transition is semantic state only; the scheduled apply-absent reconciler remains the only BigQuery writer. """ effects: list[DataAccessCompiledEffect] = [] revoked_reasons: dict[str, int] = {} page_size = max(1, min(batch_size, 500)) after_binding_id: uuid.UUID | None = None while True: conditions = [ DataAccessBinding.principal_type_snapshot == c.PRINCIPAL_TYPE_GOOGLE_USER, DataAccessBinding.status == BINDING_STATUS_ACTIVE, ] if after_binding_id is not None: conditions.append(DataAccessBinding.id > after_binding_id) bindings = ( await db.scalars( select(DataAccessBinding) .where(*conditions) .order_by(DataAccessBinding.id) .limit(page_size) .with_for_update(), ) ).all() if not bindings: break for binding in bindings: reason = await self._personal_binding_staleness_reason(db, binding) if reason is None: continue resource = await self._binding_resource(db, binding) entitlement = await self._binding_entitlement(db, binding) reference_effect = await self._grant_effect_snapshot_for_binding( db, binding, ) self._mark_binding_revoked_pending(binding, reason=reason) effect = self._compiled_effect_from_binding( binding=binding, resource=resource, entitlement=entitlement, desired_state=EFFECT_DESIRED_ABSENT, managed_reason=EFFECT_MANAGED_REASON_REVOKE_REQUEST, reference_effect=reference_effect, ) db.add(effect) effects.append(effect) revoked_reasons[reason] = revoked_reasons.get(reason, 0) + 1 # Keep each database lock acquisition bounded while allowing a # valid early page to never starve an invalid later binding. after_binding_id = bindings[-1].id if not effects: return DataAccessRevokeResult( status="noop", reason=None, authorized_as="system", ) revoke_trace_id = trace_id or "data_access_personal_identity_sweep" await db.flush() await self._audit_compile_output( db, trace_id=revoke_trace_id, actor_type="system", actor_user_id=None, resource_type="data_access_personal_principal", resource_id="personal_identity_sweep", phase="personal_identity_revoke", effects=effects, request_id=None, detail_extra={ "revoked_pending_count": len(effects), "reasons": revoked_reasons, "provider_mutation_attempted": False, }, ) await AuditService(db).log( trace_id=revoke_trace_id, actor_type="system", actor_user_id=None, action="data_access.personal_principal.stale_revoked", status="success", resource_type="data_access_personal_principal", resource_id="personal_identity_sweep", detail={ "revoked_pending_count": len(effects), "reasons": revoked_reasons, "provider_mutation_attempted": False, }, ) await db.flush() return DataAccessRevokeResult( status=BINDING_STATUS_REVOKED_PENDING, reason="personal_identity_stale", authorized_as="system", revoked_pending_count=len(effects), effect_count=len(effects), ) async def _personal_binding_staleness_reason( self, db: AsyncSession, binding: DataAccessBinding, ) -> str | None: if binding.principal_subject_snapshot is None: return "personal_identity_missing" try: person_id = uuid.UUID(binding.principal_subject_snapshot) except ValueError: return "personal_identity_invalid" person = await db.get(IdentityPerson, person_id) if person is None: return "personal_identity_missing" if person.employment_status != "active": return "personal_identity_inactive" try: current_email = _normalise_personal_work_email(person.work_email) except DataAccessSubmissionError: return "personal_work_email_invalid" if current_email != binding.principal_email_snapshot: return "personal_work_email_changed" return None async def confirm_data_access_resource_removed( self, db: AsyncSession, *, resource_id: uuid.UUID, reason: str, trace_id: str | None = None, actor_type: str = "system", actor_user_id: uuid.UUID | None = None, ) -> DataAccessRevokeResult: """Confirm a dataset removal and compile semantic absent effects. This is a Helix semantic transition only. It never calls BigQuery; the DAX apply-absent reconciler remains the provider mutation owner. """ resource = await self._resource_for_update(db, resource_id) if resource is None: raise AppException( code="DATA_ACCESS_RESOURCE_NOT_FOUND", message="Data-access resource was not found.", status_code=404, field="resource_id", ) clean_reason = _clean_revoke_reason(reason) previous_status = resource.status resource.status = RESOURCE_STATUS_DELETED resource.requestable = False bindings = ( await db.scalars( select(DataAccessBinding) .where( DataAccessBinding.resource_id == resource.id, DataAccessBinding.status == BINDING_STATUS_ACTIVE, ) .with_for_update(), ) ).all() already_pending_count = int( await db.scalar( select(func.count()) .select_from(DataAccessBinding) .where( DataAccessBinding.resource_id == resource.id, DataAccessBinding.status == BINDING_STATUS_REVOKED_PENDING, ), ) or 0, ) already_revoked_count = int( await db.scalar( select(func.count()) .select_from(DataAccessBinding) .where( DataAccessBinding.resource_id == resource.id, DataAccessBinding.status == BINDING_STATUS_REVOKED, ), ) or 0, ) effects: list[DataAccessCompiledEffect] = [] for binding in bindings: entitlement = await self._binding_entitlement(db, binding) reference_effect = await self._grant_effect_snapshot_for_binding( db, binding, ) self._mark_binding_revoked_pending(binding, reason=clean_reason) effect = self._compiled_effect_from_binding( binding=binding, resource=resource, entitlement=entitlement, desired_state=EFFECT_DESIRED_ABSENT, managed_reason=EFFECT_MANAGED_REASON_REVOKE_REQUEST, reference_effect=reference_effect, ) db.add(effect) effects.append(effect) removal_trace_id = trace_id or f"data_access_resource_removed:{resource.id}" await db.flush() if effects: await self._audit_compile_output( db, trace_id=removal_trace_id, actor_type=actor_type, actor_user_id=actor_user_id, resource_type="data_access_resource", resource_id=str(resource.id), phase="resource_removal", effects=effects, request_id=None, detail_extra={ "resource_id": str(resource.id), "reason": clean_reason, "previous_status": previous_status, "new_status": resource.status, "revoked_pending_count": len(bindings), }, ) await AuditService(db).log( trace_id=removal_trace_id, actor_type=actor_type, actor_user_id=actor_user_id, action="data_access.resource.removal_confirmed", status="success", resource_type="data_access_resource", resource_id=str(resource.id), detail={ "resource_id": str(resource.id), "reason": clean_reason, "previous_status": previous_status, "new_status": resource.status, "requestable": resource.requestable, "revoked_pending_count": len(bindings), "already_pending_count": already_pending_count, "already_revoked_count": already_revoked_count, "effect_count": len(effects), "provider_mutation_attempted": False, }, ) await db.flush() return DataAccessRevokeResult( status=resource.status, reason=clean_reason, authorized_as=actor_type, revoked_pending_count=len(bindings), effect_count=len(effects), already_pending_count=already_pending_count, already_revoked_count=already_revoked_count, ) async def revoke_invalid_google_group_bindings( self, db: AsyncSession, *, email_group_id: uuid.UUID, trace_id: str | None = None, ) -> DataAccessRevokeResult: """System cleanup for deleted/invalid Google Group read-model signals.""" group = await db.get(EmailGroup, email_group_id) if group is None: raise AppException( code="DATA_ACCESS_GROUP_NOT_FOUND", message="Google Group read-model row was not found.", status_code=404, field="email_group_id", ) bindings = ( await db.scalars( select(DataAccessBinding) .where( DataAccessBinding.principal_email_group_id == group.id, DataAccessBinding.status == BINDING_STATUS_ACTIVE, ) .with_for_update(), ) ).all() if not bindings: return DataAccessRevokeResult( status="noop", reason=None, authorized_as="system", ) bridge, bridge_blocked_reason = await self._system_group_rename_bridge( db, group, ) if bridge_blocked_reason is not None: return await self._audit_system_revoke_blocked( db, group=group, reason=bridge_blocked_reason, trace_id=trace_id, ) if bridge is not None: return await self._bridge_renamed_group_bindings( db, old_group=group, new_group=bridge, bindings=list(bindings), trace_id=trace_id, ) active_invalid_reason = await self._active_group_invalid_revoke_reason( db, group, ) if active_invalid_reason is None and group.status == "active": return DataAccessRevokeResult( status="noop", reason=None, authorized_as="system", ) reason = active_invalid_reason or ( "group_deleted_without_bridge" if group.status == "deleted" else "group_invalid" ) effects: list[DataAccessCompiledEffect] = [] for binding in bindings: resource = await self._binding_resource(db, binding) entitlement = await self._binding_entitlement(db, binding) reference_effect = await self._grant_effect_snapshot_for_binding( db, binding, ) self._mark_binding_revoked_pending(binding, reason=reason) effect = self._compiled_effect_from_binding( binding=binding, resource=resource, entitlement=entitlement, desired_state=EFFECT_DESIRED_ABSENT, managed_reason=EFFECT_MANAGED_REASON_REVOKE_REQUEST, reference_effect=reference_effect, ) db.add(effect) effects.append(effect) await db.flush() await self._audit_compile_output( db, trace_id=trace_id or f"data_access_group_revoke:{group.id}", actor_type="system", actor_user_id=None, resource_type="email_group", resource_id=str(group.id), phase="system_group_revoke", effects=effects, request_id=None, detail_extra={ "email_group_id": str(group.id), "google_group_id": group.google_group_id, "reason": reason, "revoked_pending_count": len(bindings), }, ) await AuditService(db).log( trace_id=trace_id or f"data_access_group_revoke:{group.id}", actor_type="system", actor_user_id=None, action="data_access.binding.system_revoked_group", status="success", resource_type="email_group", resource_id=str(group.id), detail={ "email_group_id": str(group.id), "google_group_id": group.google_group_id, "reason": reason, "revoked_pending_count": len(bindings), "effect_count": len(effects), }, ) await db.flush() return DataAccessRevokeResult( status=BINDING_STATUS_REVOKED_PENDING, reason=reason, authorized_as="system", revoked_pending_count=len(bindings), effect_count=len(effects), ) def _callback_source_payload( self, approval_request: ApprovalRequest, instance_detail: dict[str, Any], ) -> dict[str, Any]: source_payload = dict(approval_request.source_payload or {}) source_payload["latest_instance_detail"] = instance_detail source_payload.pop("data_access_callback_deferred_until", None) return source_payload async def _callback_request_for_approval( self, db: AsyncSession, *, approval_request: ApprovalRequest, source_payload: dict[str, Any], ) -> DataAccessRequest | None: request = await db.scalar( select(DataAccessRequest) .where(DataAccessRequest.approval_request_id == approval_request.id) .with_for_update(), ) if request is not None: return request request_id_raw = source_payload.get("data_access_request_id") if not request_id_raw: return None try: request_id = uuid.UUID(str(request_id_raw)) except ValueError: return None return await db.scalar( select(DataAccessRequest).where(DataAccessRequest.id == request_id).with_for_update(), ) async def _callback_request_items( self, db: AsyncSession, request: DataAccessRequest, ) -> list[DataAccessRequestItem]: return ( await db.scalars( select(DataAccessRequestItem) .where(DataAccessRequestItem.request_id == request.id) .order_by(DataAccessRequestItem.line_no) .with_for_update(), ) ).all() def _callback_binding_mismatch( self, *, approval_request: ApprovalRequest, request: DataAccessRequest, source_payload: dict[str, Any], instance_detail: dict[str, Any], ) -> str | None: if approval_request.approval_type != request.approval_flow_key: return "approval_type_mismatch" if request.approval_request_id != approval_request.id: return "approval_request_mismatch" if approval_request.external_id != request.feishu_instance_code: return "approval_external_id_mismatch" instance_code = str(instance_detail.get("instance_code") or "").strip() if instance_code and instance_code != request.feishu_instance_code: return "callback_instance_mismatch" request_id_raw = source_payload.get("data_access_request_id") if request_id_raw: try: source_request_id = uuid.UUID(str(request_id_raw)) except ValueError: return "source_request_id_invalid" if source_request_id != request.id: return "source_request_id_mismatch" return None async def _settle_non_approved_callback( self, db: AsyncSession, *, approval_request: ApprovalRequest, request: DataAccessRequest, items: Sequence[DataAccessRequestItem], source_payload: dict[str, Any], approval_status: str, trace_id: str, ) -> DataAccessRequestCallbackResult: now = datetime.now(UTC) request.status = ( REQUEST_STATUS_REJECTED if approval_status == "rejected" else REQUEST_STATUS_CANCELLED ) request.decided_at = now request.updated_at = now for item in items: item.status = REQUEST_ITEM_STATUS_REJECTED item.staleness_reason = None approval_request.status = approval_status source_payload["data_access_callback_outcome"] = { "status": request.status, "request_id": str(request.id), "approval_status": approval_status, } approval_request.source_payload = source_payload approval_request.updated_at = now await AuditService(db).log( trace_id=trace_id, actor_type="system", actor_user_id=request.requester_person_id, action=f"data_access.request.{request.status}", resource_type="data_access_request", resource_id=str(request.id), status="success", detail={ "approval_request_id": str(approval_request.id), "feishu_instance_code": request.feishu_instance_code, "approval_status": approval_status, }, ) await db.flush() return DataAccessRequestCallbackResult( request_id=request.id, approval_request_id=approval_request.id, approval_status=approval_status, request_status=request.status, binding_count=0, effect_count=0, ) async def _refuse_callback( self, db: AsyncSession, *, approval_request: ApprovalRequest, request: DataAccessRequest | None, items: Sequence[DataAccessRequestItem], source_payload: dict[str, Any], approval_status: str, reason: str, trace_id: str, detail: dict[str, Any], ) -> DataAccessRequestCallbackResult: now = datetime.now(UTC) request_id = request.id if request is not None else None if request is not None: request.status = REQUEST_STATUS_STALE_REQUIRES_RESUBMIT request.decided_at = now request.updated_at = now for item in items: item.status = REQUEST_ITEM_STATUS_STALE item.staleness_reason = reason approval_request.status = approval_status source_payload["data_access_callback_refusal"] = { "reason": reason, "request_id": str(request_id) if request_id is not None else None, "approval_status": approval_status, **detail, } approval_request.source_payload = source_payload approval_request.updated_at = now await AuditService(db).log( trace_id=trace_id, actor_type="system", actor_user_id=(request.requester_person_id if request is not None else None), action="data_access.request.callback_refused", resource_type=("data_access_request" if request is not None else "approval_request"), resource_id=(str(request.id) if request is not None else str(approval_request.id)), status="failure", detail={ "approval_request_id": str(approval_request.id), "approval_status": approval_status, "reason": reason, **detail, }, ) await db.flush() return DataAccessRequestCallbackResult( request_id=request_id, approval_request_id=approval_request.id, approval_status=approval_status, request_status=request.status if request is not None else None, binding_count=0, effect_count=0, refusal_reason=reason, ) async def _audit_callback_received( self, db: AsyncSession, *, approval_request: ApprovalRequest, request: DataAccessRequest | None, instance_detail: dict[str, Any], source_payload: dict[str, Any], approval_status: str, trace_id: str, ) -> None: request_id_raw = source_payload.get("data_access_request_id") resource_type = "approval_request" resource_id = str(approval_request.id) if request is not None: resource_type = "data_access_request" resource_id = str(request.id) elif request_id_raw: resource_type = "data_access_request" resource_id = str(request_id_raw) coerced_trace_id = AuditService._coerce_trace_id(trace_id) existing = await db.scalar( select(AuditLog.id).where( AuditLog.action == "data_access.request.callback_received", AuditLog.trace_id == coerced_trace_id, AuditLog.resource_type == resource_type, AuditLog.resource_id == resource_id, ), ) if existing is not None: return await AuditService(db).log( trace_id=trace_id, actor_type="integration", actor_user_id=approval_request.applicant_id, action="data_access.request.callback_received", resource_type=resource_type, resource_id=resource_id, status="success", detail={ "approval_request_id": str(approval_request.id), "approval_type": approval_request.approval_type, "feishu_instance_code": approval_request.external_id, "instance_code": instance_detail.get("instance_code"), "approval_status": approval_status, "request_id_raw": (str(request_id_raw) if request_id_raw is not None else None), }, ) async def _audit_binding_created( self, db: AsyncSession, *, trace_id: str, request: DataAccessRequest, approval_request: ApprovalRequest, binding: DataAccessBinding, ) -> None: await AuditService(db).log( trace_id=trace_id, actor_type="system", actor_user_id=request.requester_person_id, action="data_access.binding.created", resource_type="data_access_binding", resource_id=str(binding.id), status="success", detail={ "request_id": str(request.id), "approval_request_id": str(approval_request.id), "feishu_instance_code": request.feishu_instance_code, "source_request_item_id": str(binding.source_request_item_id), "resource_id": str(binding.resource_id), "principal_email_group_id": str( binding.principal_email_group_id, ), "principal_google_group_id_snapshot": (binding.principal_google_group_id_snapshot), "principal_email_snapshot": binding.principal_email_snapshot, "entitlement_id": str(binding.entitlement_id), "entitlement_snapshot": binding.entitlement_snapshot, "entitlement_version_snapshot": (binding.entitlement_version_snapshot), "binding_status": binding.status, }, ) async def _audit_compile_output( self, db: AsyncSession, *, trace_id: str, actor_type: str, actor_user_id: uuid.UUID | None, resource_type: str, resource_id: str, phase: str, effects: Sequence[Any], request_id: uuid.UUID | None, detail_extra: dict[str, Any] | None = None, ) -> None: effect_payloads = [self._effect_audit_payload(effect) for effect in effects] base_detail: dict[str, Any] = { "phase": phase, "request_id": str(request_id) if request_id is not None else None, "effect_count": len(effect_payloads), } if detail_extra: base_detail.update(detail_extra) chunks = self._audit_effect_payload_chunks(base_detail, effect_payloads) for chunk_index, chunk in enumerate(chunks, start=1): detail: dict[str, Any] = { **base_detail, "effect_chunk_index": chunk_index, "effect_chunk_count": len(chunks), "effect_chunk_size": len(chunk), "effects": chunk, } await AuditService(db).log( trace_id=trace_id, actor_type=actor_type, actor_user_id=actor_user_id, action="data_access.compile.output", resource_type=resource_type, resource_id=resource_id, status="success", detail=detail, ) def _effect_audit_payload(self, effect: Any) -> dict[str, Any]: payload: dict[str, Any] = {} audit_fields = ( "id", "binding_id", "resource_id", "principal_email_group_id", "entitlement_id", "provider", "resource_type", "target_project_id", "target_dataset_id", "provider_member", "provider_role", "desired_state", "entitlement_version_snapshot", "resource_source_version_snapshot", "resource_source_etag_snapshot", "compiler_version", "managed_reason", "already_present_at_first_apply", "status", ) for field in audit_fields: if not hasattr(effect, field): continue value = getattr(effect, field) if value is None: continue key = "effect_id" if field == "id" else field payload[key] = str(value) if isinstance(value, uuid.UUID) else value if "provider" not in payload and "target_project_id" in payload: payload["provider"] = c.PROVIDER_BIGQUERY if "resource_type" not in payload and "target_dataset_id" in payload: payload["resource_type"] = c.RESOURCE_TYPE_DATASET return payload def _audit_effect_payload_chunks( self, base_detail: dict[str, Any], effect_payloads: Sequence[dict[str, Any]], ) -> list[list[dict[str, Any]]]: if not effect_payloads: return [[]] chunks: list[list[dict[str, Any]]] = [] current: list[dict[str, Any]] = [] for payload in effect_payloads: candidate = [*current, payload] if current and ( self._audit_compile_detail_size(base_detail, candidate) > DATA_ACCESS_AUDIT_DETAIL_SOFT_LIMIT_BYTES ): chunks.append(current) current = [payload] else: current = candidate if current: chunks.append(current) return chunks def _audit_compile_detail_size( self, base_detail: dict[str, Any], effects: Sequence[dict[str, Any]], ) -> int: # Match AuditService's size check while keeping room for actual chunk # indexes/counts that are filled after chunking. return len( json.dumps( { **base_detail, "effect_chunk_index": 999, "effect_chunk_count": 999, "effect_chunk_size": len(effects), "effects": list(effects), }, default=str, ), ) async def _compile_approved_callback_effects( self, db: AsyncSession, *, request: DataAccessRequest, items: Sequence[DataAccessRequestItem], source_payload: dict[str, Any], ) -> dict[str, Any]: entitlement = await db.get(DataAccessEntitlement, request.entitlement_id) if ( entitlement is None or not entitlement.is_canonical_read_v1 or entitlement.status != ENTITLEMENT_STATUS_ACTIVE or not entitlement.requestable or request.entitlement_snapshot != c.ENTITLEMENT_READ or request.entitlement_version_snapshot != ENTITLEMENT_VERSION_V1 ): return self._callback_refusal_compile_result( "entitlement_stale", entitlement_id=str(request.entitlement_id), ) requester = await db.get(IdentityPerson, request.requester_person_id) if requester is None or requester.employment_status != "active": return self._callback_refusal_compile_result( "requester_stale", requester_person_id=str(request.requester_person_id), ) if request.principal_type_snapshot == c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT: return await self._compile_approved_service_account_callback_effects( db, request=request, requester=requester, items=items, source_payload=source_payload, entitlement=entitlement, ) if request.principal_type_snapshot == c.PRINCIPAL_TYPE_GOOGLE_USER: return await self._compile_approved_personal_callback_effects( db, request=request, requester=requester, items=items, source_payload=source_payload, entitlement=entitlement, ) if request.principal_email_group_id is None: return self._callback_refusal_compile_result( "principal_snapshot_missing", ) try: principal = await self.validate_google_group_principal( db, request.principal_email_group_id, principal_type=c.PRINCIPAL_TYPE_GOOGLE_GROUP, allow_deleted_rename_bridge=True, ) except DataAccessPrincipalValidationError as exc: return self._callback_refusal_compile_result( f"principal_{exc.reason}", principal_email_group_id=str(request.principal_email_group_id), ) if principal.google_group_id != request.principal_google_group_id_snapshot: return self._callback_refusal_compile_result( "principal_google_group_changed", request_google_group_id=(request.principal_google_group_id_snapshot), current_google_group_id=principal.google_group_id, ) try: await self.authorize_requester_for_google_group_principal( db, requester, principal, ) except DataAccessRequesterAuthorizationError: return self._callback_refusal_compile_result( "requester_authorization_stale", requester_person_id=str(request.requester_person_id), principal_email_group_id=str(principal.principal_email_group_id), ) preview_index, preview_error = self._callback_compiled_preview_index( source_payload, ) if preview_error is not None: return self._callback_refusal_compile_result( preview_error, request_id=str(request.id), ) compiled_effects: list[tuple[DataAccessRequestItem, DataAccessDryRunCompiledEffect]] = [] for item in items: preview = preview_index.get(item.resource_id) if preview is None: return self._callback_refusal_compile_result( "compiler_version_missing", resource_id=str(item.resource_id), line_no=item.line_no, ) compiler_version = str(preview.get("compiler_version") or "").strip() if compiler_version != DATA_ACCESS_COMPILER_VERSION: return self._callback_refusal_compile_result( "compiler_version_stale", resource_id=str(item.resource_id), line_no=item.line_no, callback_compiler_version=compiler_version, current_compiler_version=DATA_ACCESS_COMPILER_VERSION, ) preview_entitlement_version = str( preview.get("entitlement_version_snapshot") or "", ).strip() if preview_entitlement_version != request.entitlement_version_snapshot: return self._callback_refusal_compile_result( "entitlement_version_stale", resource_id=str(item.resource_id), line_no=item.line_no, callback_entitlement_version=preview_entitlement_version, request_entitlement_version=request.entitlement_version_snapshot, ) try: resource = await self.validate_requestable_bigquery_dataset_resource( db, item.resource_id, require_resource_owner_authority=True, ) except DataAccessResourceValidationError as exc: return self._callback_refusal_compile_result( f"resource_{exc.reason}", resource_id=str(item.resource_id), line_no=item.line_no, ) stale_result = self._stale_resource_detail(item, resource) if stale_result is not None: stale_reason, stale_detail = stale_result return self._callback_refusal_compile_result( stale_reason, **stale_detail, ) existing_binding = await db.scalar( select(DataAccessBinding) .where( DataAccessBinding.resource_id == resource.resource_id, DataAccessBinding.principal_email_group_id == principal.principal_email_group_id, DataAccessBinding.entitlement_id == entitlement.id, DataAccessBinding.status.in_(tuple(BINDING_CURRENT_STATUSES)), ) .with_for_update(), ) if existing_binding is not None and existing_binding.source_request_item_id != item.id: return self._callback_refusal_compile_result( "active_binding_conflict", resource_id=str(item.resource_id), binding_id=str(existing_binding.id), ) effect = self.compile_bigquery_dataset_read_dry_run( resource, principal, entitlement, ) compiled_effects.append((item, effect)) return { "refusal_reason": None, "refusal_detail": {}, "effects": compiled_effects, } async def _compile_approved_service_account_callback_effects( self, db: AsyncSession, *, request: DataAccessRequest, requester: IdentityPerson, items: Sequence[DataAccessRequestItem], source_payload: dict[str, Any], entitlement: DataAccessEntitlement, ) -> dict[str, Any]: if request.principal_id is None: return self._callback_refusal_compile_result("principal_snapshot_missing") try: principal = await self.validate_gcp_service_account_principal( db, request.principal_id, ) except DataAccessPrincipalValidationError as exc: return self._callback_refusal_compile_result( "principal_stale", detail=str(exc), ) if ( request.principal_subject_snapshot != principal.subject_snapshot or request.principal_email_snapshot != principal.email_snapshot ): return self._callback_refusal_compile_result( "principal_context_changed", principal_id=str(principal.principal_id), ) semantic = source_payload.get("semantic_intent") if not isinstance(semantic, dict) or any(( semantic.get("principal_id") != str(principal.principal_id), semantic.get("principal_subject_snapshot") != principal.subject_snapshot, semantic.get("source_version") != principal.source_version, semantic.get("source_etag") != principal.source_etag, )): return self._callback_refusal_compile_result( "principal_record_stale", principal_id=str(principal.principal_id), ) previews, preview_error = self._callback_compiled_preview_index(source_payload) if preview_error is not None: return self._callback_refusal_compile_result(preview_error) compiled: list[tuple[DataAccessRequestItem, DataAccessDryRunCompiledEffect]] = [] for item in items: preview = previews.get(item.resource_id) if ( preview is None or preview.get("compiler_version") != DATA_ACCESS_COMPILER_VERSION or preview.get("entitlement_version_snapshot") != request.entitlement_version_snapshot ): return self._callback_refusal_compile_result( "compiler_version_stale", resource_id=str(item.resource_id), ) try: resource = await self.validate_requestable_bigquery_dataset_resource( db, item.resource_id, require_resource_owner_authority=True, ) except DataAccessResourceValidationError as exc: return self._callback_refusal_compile_result( f"resource_{exc.reason}", resource_id=str(item.resource_id), ) stale = self._stale_resource_detail(item, resource) if stale is not None: reason, detail = stale return self._callback_refusal_compile_result(reason, **detail) existing = await db.scalar( select(DataAccessBinding).where( DataAccessBinding.resource_id == resource.resource_id, DataAccessBinding.principal_id == principal.principal_id, DataAccessBinding.entitlement_id == entitlement.id, DataAccessBinding.status.in_(tuple(BINDING_CURRENT_STATUSES)), ).with_for_update(), ) if existing is not None and existing.source_request_item_id != item.id: return self._callback_refusal_compile_result( "active_binding_conflict", resource_id=str(item.resource_id), binding_id=str(existing.id), ) compiled.append((item, self.compile_bigquery_dataset_service_account_read_dry_run( resource, principal, entitlement, ))) return {"refusal_reason": None, "refusal_detail": {}, "effects": compiled} async def _compile_approved_personal_callback_effects( self, db: AsyncSession, *, request: DataAccessRequest, requester: IdentityPerson, items: Sequence[DataAccessRequestItem], source_payload: dict[str, Any], entitlement: DataAccessEntitlement, ) -> dict[str, Any]: if request.principal_id is None: return self._callback_refusal_compile_result("principal_snapshot_missing") try: principal = await self.validate_google_user_principal( db, principal_id=request.principal_id, requester=requester, ) except (DataAccessPrincipalValidationError, DataAccessSubmissionError) as exc: return self._callback_refusal_compile_result( "principal_stale", detail=str(exc), ) if ( request.principal_subject_snapshot != principal.subject_snapshot or request.principal_email_snapshot != principal.email_snapshot ): return self._callback_refusal_compile_result( "principal_context_changed", principal_id=str(principal.principal_id), ) semantic = source_payload.get("semantic_intent") if not isinstance(semantic, dict) or any(( semantic.get("principal_type") != c.PRINCIPAL_TYPE_GOOGLE_USER, semantic.get("principal_id") != str(principal.principal_id), semantic.get("principal_subject_snapshot") != principal.subject_snapshot, semantic.get("principal_email_snapshot") != principal.email_snapshot, semantic.get("source_version") != principal.source_version, semantic.get("source_etag") != principal.source_etag, )): return self._callback_refusal_compile_result( "principal_record_stale", principal_id=str(principal.principal_id), ) previews, preview_error = self._callback_compiled_preview_index(source_payload) if preview_error is not None: return self._callback_refusal_compile_result(preview_error) compiled: list[tuple[DataAccessRequestItem, DataAccessDryRunCompiledEffect]] = [] for item in items: preview = previews.get(item.resource_id) if ( preview is None or preview.get("compiler_version") != DATA_ACCESS_COMPILER_VERSION or preview.get("entitlement_version_snapshot") != request.entitlement_version_snapshot ): return self._callback_refusal_compile_result( "compiler_version_stale", resource_id=str(item.resource_id), ) try: resource = await self.validate_requestable_bigquery_dataset_resource( db, item.resource_id, require_resource_owner_authority=True, ) except DataAccessResourceValidationError as exc: return self._callback_refusal_compile_result( f"resource_{exc.reason}", resource_id=str(item.resource_id), ) stale = self._stale_resource_detail(item, resource) if stale is not None: reason, detail = stale return self._callback_refusal_compile_result(reason, **detail) existing = await db.scalar( select(DataAccessBinding).where( DataAccessBinding.resource_id == resource.resource_id, DataAccessBinding.principal_id == principal.principal_id, DataAccessBinding.entitlement_id == entitlement.id, DataAccessBinding.status.in_(tuple(BINDING_CURRENT_STATUSES)), ).with_for_update(), ) if existing is not None and existing.source_request_item_id != item.id: return self._callback_refusal_compile_result( "active_binding_conflict", resource_id=str(item.resource_id), binding_id=str(existing.id), ) compiled.append((item, self.compile_bigquery_dataset_personal_read_dry_run( resource, principal, entitlement, ))) return {"refusal_reason": None, "refusal_detail": {}, "effects": compiled} def _callback_compiled_preview_index( self, source_payload: dict[str, Any], ) -> tuple[dict[uuid.UUID, dict[str, Any]], str | None]: raw_previews = source_payload.get("compiled_preview") if not isinstance(raw_previews, list) or not raw_previews: return {}, "compiler_version_missing" previews: dict[uuid.UUID, dict[str, Any]] = {} for raw_preview in raw_previews: if not isinstance(raw_preview, dict): return {}, "compiled_preview_invalid" try: resource_id = uuid.UUID(str(raw_preview.get("resource_id") or "")) except ValueError: return {}, "compiled_preview_invalid" if resource_id in previews: return {}, "compiled_preview_duplicate_resource" previews[resource_id] = raw_preview return previews, None def _stale_resource_detail( self, item: DataAccessRequestItem, resource: ValidatedRequestableDatasetResource, ) -> tuple[str, dict[str, Any]] | None: owner_snapshot = _uuid_snapshot( resource.owner_person_id, field="resource_owner_person_id", ) checks = { "project_id": ( resource.project_id, item.resource_project_id_snapshot, ), "dataset_id": ( resource.dataset_id, item.resource_dataset_id_snapshot, ), "source_version": ( resource.source_version, item.resource_source_version_snapshot, ), "source_etag": ( resource.source_etag, item.resource_source_etag_snapshot, ), "owner_person_id": ( str(owner_snapshot) if owner_snapshot is not None else None, ( str(item.resource_owner_person_id_snapshot) if item.resource_owner_person_id_snapshot is not None else None ), ), } for field, (current, snapshot) in checks.items(): if current != snapshot: reason = ( DATA_ACCESS_STALENESS_REASON_RESOURCE_CATALOG_STALE if field in {"source_version", "source_etag"} else DATA_ACCESS_STALENESS_REASON_RESOURCE_CONTEXT_CHANGED ) return ( reason, { "line_no": item.line_no, "resource_id": str(item.resource_id), "field": field, "current": current, "snapshot": snapshot, }, ) return None def _callback_refusal_compile_result( self, reason: str, **detail: Any, ) -> dict[str, Any]: return { "refusal_reason": reason, "refusal_detail": detail, "effects": [], } async def _request_effect_counts( self, db: AsyncSession, request: DataAccessRequest, ) -> tuple[int, int]: binding_count = int( await db.scalar( select(func.count()) .select_from(DataAccessBinding) .where(DataAccessBinding.source_request_id == request.id), ) or 0, ) effect_count = int( await db.scalar( select(func.count()) .select_from(DataAccessCompiledEffect) .join( DataAccessBinding, DataAccessBinding.id == DataAccessCompiledEffect.binding_id, ) .where(DataAccessBinding.source_request_id == request.id), ) or 0, ) return binding_count, effect_count async def _binding_for_update( self, db: AsyncSession, binding_id: uuid.UUID, ) -> DataAccessBinding | None: return ( await db.scalars( select(DataAccessBinding) .where(DataAccessBinding.id == binding_id) .with_for_update(), ) ).one_or_none() async def _resource_for_update( self, db: AsyncSession, resource_id: uuid.UUID, ) -> DataAccessResource | None: return ( await db.scalars( select(DataAccessResource) .where(DataAccessResource.id == resource_id) .with_for_update(), ) ).one_or_none() async def _binding_resource( self, db: AsyncSession, binding: DataAccessBinding, ) -> DataAccessResource: resource = await db.get(DataAccessResource, binding.resource_id) if resource is None: raise AppException( code="DATA_ACCESS_BINDING_RESOURCE_MISSING", message="Binding resource was not found.", status_code=409, field="binding_id", ) return resource async def _binding_entitlement( self, db: AsyncSession, binding: DataAccessBinding, ) -> DataAccessEntitlement: entitlement = await db.get(DataAccessEntitlement, binding.entitlement_id) if entitlement is None: raise AppException( code="DATA_ACCESS_BINDING_ENTITLEMENT_MISSING", message="Binding entitlement was not found.", status_code=409, field="binding_id", ) return entitlement async def _authorize_revoke_actor( self, db: AsyncSession, *, actor: IdentityPerson, binding: DataAccessBinding, resource: DataAccessResource, ) -> str: role_names = await self._requester_role_names(db, actor) if has_role(role_names, GLOBAL_ADMIN_ROLE): return "global_admin" if ( binding.principal_type_snapshot == c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT and binding.principal_id is not None ): technical_owner = await db.scalar( select(DataAccessServiceAccountProfile.principal_id).where( DataAccessServiceAccountProfile.principal_id == binding.principal_id, DataAccessServiceAccountProfile.technical_owner_person_id == actor.id, ).limit(1), ) if technical_owner is not None: return "technical_owner" if ( binding.principal_type_snapshot == c.PRINCIPAL_TYPE_GOOGLE_USER and binding.principal_subject_snapshot == str(actor.id) ): return "personal_principal" group_owner = await db.scalar( select(EmailGroupOwner.id).where( EmailGroupOwner.email_group_id == binding.principal_email_group_id, EmailGroupOwner.person_id == actor.id, ), ) if group_owner is not None: return "group_owner" if ( resource.resource_owner_authority_enabled and _uuid_or_none(resource.owner_person_id) == actor.id ): return "resource_owner" raise DataAccessRevokeAuthorizationError( actor_person_id=actor.id, binding_id=binding.id, ) def _mark_binding_revoked_pending( self, binding: DataAccessBinding, *, reason: str, ) -> None: binding.status = BINDING_STATUS_REVOKED_PENDING binding.reconcile_status = RECONCILE_STATUS_PENDING binding.last_reconcile_error = None binding.revoke_reason = reason binding.revoked_at = datetime.now(UTC) def _compiled_effect_from_binding( self, *, binding: DataAccessBinding, resource: DataAccessResource, entitlement: DataAccessEntitlement, desired_state: str, managed_reason: str, provider_member: str | None = None, reference_effect: DataAccessCompiledEffect | None = None, ) -> DataAccessCompiledEffect: return DataAccessCompiledEffect( binding_id=binding.id, resource_id=binding.resource_id, principal_email_group_id=binding.principal_email_group_id, principal_id=binding.principal_id, principal_type_snapshot=binding.principal_type_snapshot, principal_subject_snapshot=binding.principal_subject_snapshot, entitlement_id=binding.entitlement_id, provider=c.PROVIDER_BIGQUERY, resource_type=( reference_effect.resource_type if reference_effect is not None else resource.resource_type ), target_project_id=( reference_effect.target_project_id if reference_effect is not None else resource.project_id ), target_dataset_id=( reference_effect.target_dataset_id if reference_effect is not None else resource.dataset_id ), target_table_id=( reference_effect.target_table_id if reference_effect is not None else resource.table_id ), provider_member=( provider_member or _provider_member_for_principal_binding(binding) ), provider_access_kind=( reference_effect.provider_access_kind if reference_effect is not None else ( "userByEmail" if binding.principal_type_snapshot in { c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT, c.PRINCIPAL_TYPE_GOOGLE_USER, } else "groupByEmail" ) ), provider_role=c.BIGQUERY_DATA_VIEWER_ROLE, desired_state=desired_state, entitlement_version_snapshot=( reference_effect.entitlement_version_snapshot if reference_effect is not None else entitlement.version ), resource_source_version_snapshot=( reference_effect.resource_source_version_snapshot if reference_effect is not None else resource.source_version ), resource_source_etag_snapshot=( reference_effect.resource_source_etag_snapshot if reference_effect is not None else resource.source_etag ), compiler_version=DATA_ACCESS_COMPILER_VERSION, managed_reason=managed_reason, already_present_at_first_apply=False, status=EFFECT_STATUS_PENDING_RECONCILE, ) async def _system_group_rename_bridge( self, db: AsyncSession, group: EmailGroup, ) -> tuple[EmailGroup | None, str | None]: if group.status != "deleted" or not group.google_group_id: return None, None active_same_google_id = ( await db.scalars( select(EmailGroup).where( EmailGroup.status == "active", EmailGroup.google_group_id == group.google_group_id, ), ) ).all() if len(active_same_google_id) > 1: return None, "group_rename_bridge_ambiguous" if not active_same_google_id: return None, None return active_same_google_id[0], None async def _active_group_invalid_revoke_reason( self, db: AsyncSession, group: EmailGroup, ) -> str | None: if group.status != "active": return None if not group.google_group_id: return "group_invalid" active_same_google_id = ( await db.scalars( select(EmailGroup.id).where( EmailGroup.status == "active", EmailGroup.google_group_id == group.google_group_id, ), ) ).all() if len(active_same_google_id) != 1: return "group_duplicate_active_google_group_id" return None async def _bridge_renamed_group_bindings( self, db: AsyncSession, *, old_group: EmailGroup, new_group: EmailGroup, bindings: list[DataAccessBinding], trace_id: str | None, ) -> DataAccessRevokeResult: effects: list[DataAccessCompiledEffect] = [] renamed_binding_count = 0 superseded_binding_count = 0 new_principal_id = await self._group_data_access_principal_id(db, new_group) for binding in bindings: resource = await self._binding_resource(db, binding) entitlement = await self._binding_entitlement(db, binding) reference_effect = await self._grant_effect_snapshot_for_binding( db, binding, ) old_provider_member = ( f"{c.BIGQUERY_GROUP_MEMBER_PREFIX}{binding.principal_email_snapshot}" ) new_email = new_group.email.strip().lower() current_new_binding = await self._current_semantic_binding( db, resource_id=binding.resource_id, principal_id=new_principal_id, principal_email_group_id=new_group.id, entitlement_id=binding.entitlement_id, ) if current_new_binding is not None and current_new_binding.id != binding.id: self._mark_binding_revoked_pending( binding, reason="group_rename_bridge_superseded", ) absent = self._compiled_effect_from_binding( binding=binding, resource=resource, entitlement=entitlement, desired_state=EFFECT_DESIRED_ABSENT, managed_reason=EFFECT_MANAGED_REASON_GROUP_RENAME_BRIDGE, provider_member=old_provider_member, reference_effect=reference_effect, ) bridge_effects = [absent] if current_new_binding.status == BINDING_STATUS_ACTIVE: present = self._compiled_effect_from_binding( binding=current_new_binding, resource=resource, entitlement=entitlement, desired_state=EFFECT_DESIRED_PRESENT, managed_reason=EFFECT_MANAGED_REASON_GROUP_RENAME_BRIDGE, provider_member=(f"{c.BIGQUERY_GROUP_MEMBER_PREFIX}{new_email}"), reference_effect=reference_effect, ) bridge_effects.append(present) db.add_all(bridge_effects) effects.extend(bridge_effects) superseded_binding_count += 1 continue binding.principal_email_group_id = new_group.id binding.principal_id = new_principal_id binding.principal_type_snapshot = c.PRINCIPAL_TYPE_GOOGLE_GROUP binding.principal_subject_snapshot = new_group.google_group_id binding.principal_google_group_id_snapshot = new_group.google_group_id binding.principal_email_snapshot = new_email absent = self._compiled_effect_from_binding( binding=binding, resource=resource, entitlement=entitlement, desired_state=EFFECT_DESIRED_ABSENT, managed_reason=EFFECT_MANAGED_REASON_GROUP_RENAME_BRIDGE, provider_member=old_provider_member, reference_effect=reference_effect, ) present = self._compiled_effect_from_binding( binding=binding, resource=resource, entitlement=entitlement, desired_state=EFFECT_DESIRED_PRESENT, managed_reason=EFFECT_MANAGED_REASON_GROUP_RENAME_BRIDGE, reference_effect=reference_effect, ) db.add_all([absent, present]) effects.extend([absent, present]) renamed_binding_count += 1 await db.flush() await self._audit_compile_output( db, trace_id=trace_id or f"data_access_group_rename:{old_group.id}", actor_type="system", actor_user_id=None, resource_type="email_group", resource_id=str(old_group.id), phase="group_rename_bridge", effects=effects, request_id=None, detail_extra={ "old_email_group_id": str(old_group.id), "new_email_group_id": str(new_group.id), "google_group_id": old_group.google_group_id, "renamed_binding_count": renamed_binding_count, "superseded_binding_count": superseded_binding_count, }, ) await AuditService(db).log( trace_id=trace_id or f"data_access_group_rename:{old_group.id}", actor_type="system", actor_user_id=None, action="data_access.binding.group_rename_bridged", status="success", resource_type="email_group", resource_id=str(old_group.id), detail={ "old_email_group_id": str(old_group.id), "new_email_group_id": str(new_group.id), "google_group_id": old_group.google_group_id, "renamed_binding_count": renamed_binding_count, "superseded_binding_count": superseded_binding_count, "effect_count": len(effects), }, ) await db.flush() return DataAccessRevokeResult( status=BINDING_STATUS_ACTIVE, reason="group_rename_bridge", authorized_as="system", renamed_binding_count=renamed_binding_count, revoked_pending_count=superseded_binding_count, effect_count=len(effects), ) async def _current_semantic_binding( self, db: AsyncSession, *, resource_id: uuid.UUID, principal_id: uuid.UUID | None, principal_email_group_id: uuid.UUID, entitlement_id: uuid.UUID, ) -> DataAccessBinding | None: principal_match = ( DataAccessBinding.principal_id == principal_id if principal_id is not None else DataAccessBinding.principal_email_group_id == principal_email_group_id ) return ( await db.scalars( select(DataAccessBinding) .where( DataAccessBinding.resource_id == resource_id, principal_match, DataAccessBinding.entitlement_id == entitlement_id, DataAccessBinding.status.in_(tuple(BINDING_CURRENT_STATUSES)), ) .with_for_update(), ) ).one_or_none() async def _grant_effect_snapshot_for_binding( self, db: AsyncSession, binding: DataAccessBinding, ) -> DataAccessCompiledEffect | None: return await db.scalar( select(DataAccessCompiledEffect) .where( DataAccessCompiledEffect.binding_id == binding.id, DataAccessCompiledEffect.desired_state == EFFECT_DESIRED_PRESENT, ) .order_by(DataAccessCompiledEffect.created_at.desc()) .limit(1), ) async def _audit_system_revoke_blocked( self, db: AsyncSession, *, group: EmailGroup, reason: str, trace_id: str | None, ) -> DataAccessRevokeResult: await AuditService(db).log( trace_id=trace_id or f"data_access_group_revoke_blocked:{group.id}", actor_type="system", actor_user_id=None, action="data_access.binding.system_revoke_blocked", status="failure", resource_type="email_group", resource_id=str(group.id), detail={ "email_group_id": str(group.id), "google_group_id": group.google_group_id, "reason": reason, }, ) await db.flush() return DataAccessRevokeResult( status="blocked", reason=reason, authorized_as="system", ) async def _load_read_v1_entitlement( self, db: AsyncSession, ) -> DataAccessEntitlement: entitlement = await db.scalar( select(DataAccessEntitlement).where( DataAccessEntitlement.domain == c.DOMAIN_DATA_ACCESS, DataAccessEntitlement.provider == c.PROVIDER_BIGQUERY, DataAccessEntitlement.resource_type == c.RESOURCE_TYPE_DATASET, DataAccessEntitlement.entitlement == c.ENTITLEMENT_READ, DataAccessEntitlement.version == ENTITLEMENT_VERSION_V1, DataAccessEntitlement.requestable.is_(True), DataAccessEntitlement.status == ENTITLEMENT_STATUS_ACTIVE, ), ) if entitlement is None or not entitlement.is_canonical_read_v1: raise DataAccessSubmissionError( "entitlement_not_configured", "Data-access read:v1 entitlement is not configured.", field="entitlement", status_code=503, ) return entitlement async def _data_access_feishu_client( self, db: AsyncSession, ) -> FeishuApprovalClient: config = await get_feishu_config(session=db) return FeishuApprovalClient( app_id=str(config.get("app_id") or ""), app_secret=str(config.get("app_secret") or ""), ) async def _requester_role_names( self, db: AsyncSession, requester: IdentityPerson, ) -> list[str]: override = requester.__dict__.get("_role_names_override") if override is not None: return list(override) rows = ( await db.scalars( select(AuthRole.name) .join(AuthUserRole, AuthUserRole.role_id == AuthRole.id) .where(AuthUserRole.user_id == requester.id), ) ).all() return normalize_role_names(rows) async def _requester_member_group_ids( self, db: AsyncSession, requester: IdentityPerson, ) -> set[uuid.UUID]: override = requester.__dict__.get("_data_access_member_group_ids_override") if override is not None: return { value if isinstance(value, uuid.UUID) else uuid.UUID(str(value)) for value in override } try: result = await EmailGroupService(db).list_member_groups_for_person( person=requester, ) except Exception: # noqa: BLE001 # Additive enrichment on top of group-ownership: never let a flaky # Google Directory lookup turn a valid owner's request submission # into a 500. Degrade to owner-only (empty member set). logger.warning( "data-access member-group lookup failed for requester=%s; degrading to owner-only", requester.id, exc_info=True, ) return set() group_ids: set[uuid.UUID] = set() for item in result.groups: if item.group_id is not None: group_ids.add(item.group_id) return group_ids async def _active_rename_bridge( self, db: AsyncSession, deleted_group: EmailGroup, ) -> EmailGroup: if not deleted_group.google_group_id: raise DataAccessPrincipalValidationError( "missing_google_group_id", "Deleted Google Group principal has no rename anchor.", principal=deleted_group.id, ) active_same_google_id = ( await db.scalars( select(EmailGroup).where( EmailGroup.status == "active", EmailGroup.google_group_id == deleted_group.google_group_id, ), ) ).all() if not active_same_google_id: raise DataAccessPrincipalValidationError( "deleted_group", "Deleted Google Group has no active same-google-id bridge.", principal=deleted_group.id, ) if len(active_same_google_id) > 1: raise DataAccessPrincipalValidationError( "duplicate_active_google_group_id", "Google group id resolves to multiple active groups.", principal=deleted_group.id, ) return active_same_google_id[0] def _normalised_aliases(group: EmailGroup) -> set[str]: return { alias.strip().lower() for alias in (group.aliases or []) if isinstance(alias, str) and alias.strip() } def _unique_groups(groups: list[EmailGroup]) -> list[EmailGroup]: by_id: dict[uuid.UUID, EmailGroup] = {} for group in groups: by_id[group.id] = group return list(by_id.values()) def _clean_required_multiline( value: str, *, field: str, max_length: int, ) -> str: cleaned = str(value or "").strip() if not cleaned: raise DataAccessFeishuPayloadError( "missing_required_field", f"{field} is required for data-access Feishu submission.", field=field, ) if len(cleaned) > max_length: raise DataAccessFeishuPayloadError( "field_too_long", f"{field} must be at most {max_length} characters.", field=field, ) return cleaned.replace("\r\n", "\n").replace("\r", "\n") def _clean_required_single_line( value: str, *, field: str, max_length: int, ) -> str: cleaned = str(value or "").strip() if not cleaned: raise DataAccessFeishuPayloadError( "missing_required_field", f"{field} is required for data-access Feishu submission.", field=field, ) if len(cleaned) > max_length: raise DataAccessFeishuPayloadError( "field_too_long", f"{field} must be at most {max_length} characters.", field=field, ) if "\n" in cleaned or "\r" in cleaned: raise DataAccessFeishuPayloadError( "invalid_single_line_field", f"{field} must be a single line.", field=field, ) return cleaned def _clean_optional_single_line( value: str | None, *, field: str, max_length: int, ) -> str | None: if value is None: return None cleaned = str(value).strip() if not cleaned: return None if len(cleaned) > max_length: raise DataAccessSubmissionError( "field_too_long", f"{field} must be at most {max_length} characters.", field=field, ) if "\n" in cleaned or "\r" in cleaned: raise DataAccessSubmissionError( "invalid_single_line_field", f"{field} must be a single line.", field=field, ) return cleaned def _normalise_service_account_email(value: str) -> str: email = str(value or "").strip().lower() if not email: raise DataAccessSubmissionError( "invalid_service_account_email", "Enter a valid GCP service-account email.", field="service_account_email", ) if len(email) > DATA_ACCESS_SERVICE_ACCOUNT_EMAIL_MAX_LENGTH: raise DataAccessSubmissionError( "invalid_service_account_email", "Enter a valid GCP service-account email.", field="service_account_email", ) if "\n" in email or "\r" in email: raise DataAccessSubmissionError( "invalid_service_account_email", "Enter a valid GCP service-account email.", field="service_account_email", ) if not _is_valid_service_account_email(email): raise DataAccessSubmissionError( "invalid_service_account_email", "Enter a valid GCP service-account email.", field="service_account_email", ) return email def _normalise_personal_work_email(value: str) -> str: email = str(value or "").strip().lower() if ( not email or len(email) > DATA_ACCESS_SERVICE_ACCOUNT_EMAIL_MAX_LENGTH or "\n" in email or "\r" in email or not re.fullmatch(r"[^@\s]+@[^@\s]+\.[^@\s]+", email) ): raise DataAccessSubmissionError( "invalid_personal_work_email", "Your active work account needs a valid email before requesting data access.", field="requester.work_email", status_code=422, ) return email def _is_valid_service_account_email(value: str) -> bool: return bool(re.fullmatch(c.GCP_SERVICE_ACCOUNT_EMAIL_PATTERN, value)) def _self_service_principal_etag(email: str) -> str: # The email is the user-supplied provider selector. Keep a deterministic # source fingerprint for callback staleness checks without inventing a # workload catalog or an SA technical owner. return f"self-service:{uuid.uuid5(uuid.NAMESPACE_URL, email).hex}" def _personal_principal_etag(subject: str, email: str) -> str: return f"identity-person:{uuid.uuid5(uuid.NAMESPACE_URL, f'{subject}|{email}').hex}" def _provider_member_for_principal_binding(binding: DataAccessBinding) -> str: return provider_member_for_binding(binding) def _clean_feishu_dataset_description(value: str | None) -> str | None: if value is None: return None cleaned = " ".join(str(value).split()) if not cleaned: return None if len(cleaned) <= DATA_ACCESS_FEISHU_DATASET_DESCRIPTION_MAX_LENGTH: return cleaned return cleaned[: DATA_ACCESS_FEISHU_DATASET_DESCRIPTION_MAX_LENGTH - 3].rstrip() + "..." def _is_valid_detail_url(value: str) -> bool: if value.startswith("/"): return not value.startswith("//") parsed = urlparse(value) return parsed.scheme in {"http", "https"} and bool(parsed.netloc) def _dataset_count_label(count: int) -> str: noun = "BigQuery dataset" if count == 1 else "BigQuery datasets" return f"{count} {noun}" def _feishu_request_summary( payload: ( DataAccessFeishuSubmissionPayload | DataAccessServiceAccountFeishuSubmissionPayload | DataAccessPersonalFeishuSubmissionPayload ), ) -> str: if isinstance(payload, DataAccessServiceAccountFeishuSubmissionPayload): return ( f"GCP service account {payload.principal.email_snapshot} is requesting " f"read access to {_dataset_count_label(payload.dataset_count)}." ) if isinstance(payload, DataAccessPersonalFeishuSubmissionPayload): return ( f"Personal work account {payload.principal.email_snapshot} is requesting " f"read access to {_dataset_count_label(payload.dataset_count)}." ) return ( f"Google Group {payload.principal_email_snapshot} is requesting " f"read access to {_dataset_count_label(payload.dataset_count)}." ) def _human_provider_role(provider_role: str) -> str: if provider_role == c.BIGQUERY_DATA_VIEWER_ROLE: return "BigQuery data viewer" return provider_role def _human_desired_state(desired_state: str) -> str: if desired_state == EFFECT_DESIRED_PRESENT: return "grant access" if desired_state == EFFECT_DESIRED_ABSENT: return "remove access" return desired_state.replace("_", " ") def _human_owner_check(dataset: DataAccessFeishuDatasetContext) -> str: status = dataset.owner_resolution_status.replace("_", " ") if dataset.owner_authority_enabled: return status return f"{status}; owner authority disabled" def _dataset_locator(project_id: str, dataset_id: str) -> str: return f"{project_id}.{dataset_id}" def _resource_locator(dataset: DataAccessFeishuDatasetContext) -> str: locator = _dataset_locator(dataset.project_id, dataset.dataset_id) return f"{locator}.{dataset.table_id}" if dataset.table_id else locator def _payload_resource_type( resources: Sequence[DataAccessFeishuDatasetContext], ) -> str: resource_types = {resource.resource_type for resource in resources} if len(resource_types) == 1: return next(iter(resource_types)) return "mixed" def _compiled_preview_by_resource_id( payload: DataAccessFeishuSubmissionPayload, ) -> dict[str, DataAccessFeishuCompiledPreview]: return {str(preview.resource_id): preview for preview in payload.compiled_preview} def _feishu_requested_access( payload: DataAccessFeishuSubmissionPayload, ) -> str: roles = sorted( {_human_provider_role(preview.provider_role) for preview in payload.compiled_preview} ) role_label = ", ".join(roles) if roles else "BigQuery dataset access" return f"Grant {role_label} to Google Group {payload.principal_email_snapshot}." def _feishu_dataset_list(payload: DataAccessFeishuSubmissionPayload) -> str: preview_by_resource_id = _compiled_preview_by_resource_id(payload) lines: list[str] = [] for index, dataset in enumerate(payload.datasets, start=1): lines.append( f"{index}. {_resource_locator(dataset)}", ) if dataset.description: lines.append(f" Description: {dataset.description}") preview = preview_by_resource_id.get(str(dataset.resource_id)) if preview is not None: lines.append( " Planned change: " f"{_human_desired_state(preview.desired_state)} as " f"{_human_provider_role(preview.provider_role)}", ) lines.append(f" Catalog version: {dataset.source_version}") lines.append(f" Owner check: {_human_owner_check(dataset)}") return "\n".join(lines) def _resolve_feishu_widget_spec( field_name: str, spec: Any, ) -> tuple[str, str]: default_type = DATA_ACCESS_FEISHU_FORM_DEFAULT_TYPES[field_name] if spec is None: raise DataAccessSubmissionError( "missing_payload_widget", f"Data-access Feishu payload_schema is missing {field_name}.", field=f"payload_schema.{field_name}", status_code=503, ) if isinstance(spec, str): widget_id = spec.strip() widget_type = default_type elif isinstance(spec, dict): widget_id = str( spec.get("id") or spec.get("widget_id") or spec.get("field_id") or "", ).strip() widget_type = str( spec.get("type") or spec.get("widget_type") or default_type, ).strip() else: raise DataAccessSubmissionError( "invalid_payload_widget", f"Data-access Feishu widget mapping for {field_name} is invalid.", field=f"payload_schema.{field_name}", status_code=503, ) if not widget_id or not widget_type: raise DataAccessSubmissionError( "invalid_payload_widget", f"Data-access Feishu widget mapping for {field_name} is incomplete.", field=f"payload_schema.{field_name}", status_code=503, ) return widget_id, widget_type def _feishu_form_value(value: Any) -> str: if value is None: return "" if isinstance(value, str): return value if isinstance(value, int | float): return str(value) return json.dumps(value, ensure_ascii=False, sort_keys=True) def _resolve_feishu_approver_node_id( payload_schema: dict[str, Any] | None, node_key: str, ) -> str | None: if not isinstance(payload_schema, dict): return None approval_process = payload_schema.get("approval_process") if not isinstance(approval_process, dict): return None direct = str( approval_process.get(f"{node_key}_node_id") or approval_process.get(node_key) or "", ).strip() if direct: return direct nodes = approval_process.get("approver_nodes") if isinstance(nodes, dict): approver_node = nodes.get(node_key) if isinstance(approver_node, str): node_id = approver_node.strip() elif isinstance(approver_node, dict): node_id = str( approver_node.get("id") or approver_node.get("node_id") or approver_node.get("custom_id") or "", ).strip() else: node_id = "" return node_id or None if isinstance(nodes, list): for node in nodes: if not isinstance(node, dict): continue key = str( node.get("key") or node.get("name") or node.get("canonical_node") or "", ).strip() if key != node_key: continue node_id = str( node.get("id") or node.get("node_id") or node.get("custom_id") or "", ).strip() return node_id or None return None def _resolve_feishu_data_owner_review_node_id( payload_schema: dict[str, Any] | None, ) -> str | None: return _resolve_feishu_approver_node_id(payload_schema, "data_owner_review") def _merge_feishu_approver_payloads(*payloads: dict[str, Any]) -> dict[str, Any]: merged: dict[str, list[dict[str, Any]]] = { "node_approver_user_id_list": [], "node_approver_open_id_list": [], } for payload in payloads: for key in merged: values = payload.get(key) if isinstance(values, list): merged[key].extend(value for value in values if isinstance(value, dict)) return {key: value for key, value in merged.items() if value} def _approval_source_payload( *, request_id: uuid.UUID, payload: ( DataAccessFeishuSubmissionPayload | DataAccessServiceAccountFeishuSubmissionPayload | DataAccessPersonalFeishuSubmissionPayload ), form_payload: str, instance: dict[str, Any], approver_payload: dict[str, Any] | None = None, cc_payload: dict[str, Any] | None = None, cc_result: dict[str, Any] | None = None, ) -> dict[str, Any]: fields = payload.to_audit_fields() feishu_payload: dict[str, Any] = { "instance_code": instance.get("instance_code"), "instance_status": instance.get("instance_status", "PENDING"), "submitted_form": json.loads(form_payload), } if approver_payload: feishu_payload["submitted_approvers"] = approver_payload if cc_payload: feishu_payload["submitted_cc"] = cc_payload if cc_result: feishu_payload["cc_result"] = cc_result return { "data_access_request_id": str(request_id), "approval_flow_key": payload.flow_key, "approval_code": payload.external_code, "payload_version": payload.payload_version, "semantic_intent": fields["semantic_intent"], "dataset_count": payload.dataset_count, "business_reason": payload.business_reason, "detail_url": payload.detail_url, "dataset_owner_context": fields["dataset_owner_context"], "compiled_preview": fields["compiled_preview"], "feishu": feishu_payload, } def _uuid_snapshot(value: str | uuid.UUID | None, *, field: str) -> uuid.UUID | None: if value is None: return None try: return value if isinstance(value, uuid.UUID) else uuid.UUID(str(value)) except ValueError as exc: raise DataAccessSubmissionError( "invalid_uuid_snapshot", f"{field} must be a UUID snapshot.", field=field, ) from exc def _clean_revoke_reason(value: str) -> str: clean = value.strip() if not clean: return "manual_revoke" if len(clean) > 64: raise AppException( code="DATA_ACCESS_REVOKE_REASON_TOO_LONG", message="Data-access revoke reason must be at most 64 characters.", status_code=422, field="reason", ) return clean def _uuid_or_none(value: str | uuid.UUID | None) -> uuid.UUID | None: if value is None or isinstance(value, uuid.UUID): return value try: return uuid.UUID(str(value)) except ValueError: return None