"""Canonical constants for Helix data-access management V1. DAX-002 freezes the vocabulary before schema/API work starts. Keep these values as plain strings, matching the local constants style used by sibling domains, so DB rows, JSON payloads, and tests compare exact wire values. """ from __future__ import annotations from typing import Final # ── Canonical V1 vocabulary ──────────────────────────────────────────── DOMAIN_DATA_ACCESS: Final = "data_access" PROVIDER_BIGQUERY: Final = "bigquery" RESOURCE_TYPE_DATASET: Final = "dataset" RESOURCE_TYPE_TABLE: Final = "table" PRINCIPAL_TYPE_GOOGLE_GROUP: Final = "google_group" PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT: Final = "gcp_service_account" PRINCIPAL_TYPE_GOOGLE_USER: Final = "google_user" ENTITLEMENT_READ: Final = "read" APPROVAL_FLOW_DATA_ACCESS_GROUP_GRANT: Final = "data_access_group_grant" APPROVAL_FLOW_DATA_ACCESS_SERVICE_ACCOUNT_GRANT: Final = ( "data_access_service_account_grant" ) APPROVAL_FLOW_DATA_ACCESS_PERSONAL_GRANT: Final = "data_access_personal_grant" BIGQUERY_DATA_VIEWER_ROLE: Final = "roles/bigquery.dataViewer" BIGQUERY_GROUP_MEMBER_PREFIX: Final = "group:" BIGQUERY_SERVICE_ACCOUNT_MEMBER_PREFIX: Final = "serviceAccount:" BIGQUERY_USER_MEMBER_PREFIX: Final = "user:" # Typed by employees at request time and validated again by the BigQuery ACL # adapter. The normalised email is the self-service principal selector; an # optional observed catalog must not gate request authoring. GCP_SERVICE_ACCOUNT_EMAIL_PATTERN: Final = ( r"^[a-z][a-z0-9-]{0,62}@" r"[a-z][a-z0-9-]{4,28}[a-z0-9]\.iam\.gserviceaccount\.com$" ) GCP_SERVICE_ACCOUNT_SELF_SERVICE_SOURCE_VERSION: Final = "self_service_input:v1" GOOGLE_USER_SELF_SERVICE_SOURCE_VERSION: Final = "identity_person:v1" ALL_DOMAINS: Final = (DOMAIN_DATA_ACCESS,) ALL_PROVIDERS: Final = (PROVIDER_BIGQUERY,) ALL_RESOURCE_TYPES: Final = (RESOURCE_TYPE_DATASET, RESOURCE_TYPE_TABLE) ALL_PRINCIPAL_TYPES: Final = ( PRINCIPAL_TYPE_GOOGLE_GROUP, PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT, PRINCIPAL_TYPE_GOOGLE_USER, ) ALL_ENTITLEMENTS: Final = (ENTITLEMENT_READ,) ALL_APPROVAL_FLOWS: Final = ( APPROVAL_FLOW_DATA_ACCESS_GROUP_GRANT, APPROVAL_FLOW_DATA_ACCESS_SERVICE_ACCOUNT_GRANT, APPROVAL_FLOW_DATA_ACCESS_PERSONAL_GRANT, ) # ── Durable field-name contract ──────────────────────────────────────── FIELD_PRINCIPAL_EMAIL_GROUP_ID: Final = "principal_email_group_id" FIELD_PRINCIPAL_ID: Final = "principal_id" FIELD_PRINCIPAL_TYPE_SNAPSHOT: Final = "principal_type_snapshot" FIELD_PRINCIPAL_SUBJECT_SNAPSHOT: Final = "principal_subject_snapshot" FIELD_PRINCIPAL_GOOGLE_GROUP_ID_SNAPSHOT: Final = ( "principal_google_group_id_snapshot" ) FIELD_PRINCIPAL_EMAIL_SNAPSHOT: Final = "principal_email_snapshot" FIELD_OWNER_PERSON_ID: Final = "owner_person_id" FIELD_OWNER_SOURCE: Final = "owner_source" FIELD_SOURCE_VERSION: Final = "source_version" FIELD_SOURCE_ETAG: Final = "source_etag" SEMANTIC_PRINCIPAL_FIELD: Final = FIELD_PRINCIPAL_ID PROVIDER_MEMBER_EMAIL_FIELD: Final = FIELD_PRINCIPAL_EMAIL_SNAPSHOT RUNTIME_OWNER_AUTHORITY_FIELD: Final = FIELD_OWNER_PERSON_ID PRINCIPAL_IDENTITY_FIELDS: Final = ( FIELD_PRINCIPAL_ID, FIELD_PRINCIPAL_TYPE_SNAPSHOT, FIELD_PRINCIPAL_SUBJECT_SNAPSHOT, FIELD_PRINCIPAL_EMAIL_GROUP_ID, FIELD_PRINCIPAL_GOOGLE_GROUP_ID_SNAPSHOT, FIELD_PRINCIPAL_EMAIL_SNAPSHOT, ) RESOURCE_OWNER_FIELDS: Final = ( FIELD_OWNER_PERSON_ID, FIELD_OWNER_SOURCE, FIELD_SOURCE_VERSION, FIELD_SOURCE_ETAG, ) # ── Rejected aliases ─────────────────────────────────────────────────── FORBIDDEN_DOMAIN_ALIASES: Final = frozenset({ "cmdb_data_access", "bq_access", "bigquery_access", }) FORBIDDEN_PROVIDER_ALIASES: Final = frozenset({ "gcp", "google_cloud", "google", }) FORBIDDEN_RESOURCE_TYPE_ALIASES: Final = frozenset({ "project", "bigquery_table", "bigquery_project", }) FORBIDDEN_PRINCIPAL_TYPE_ALIASES: Final = frozenset({ "user", "person", "service_account", "group", "email_group", }) FORBIDDEN_ENTITLEMENT_ALIASES: Final = frozenset({ "viewer", "data_viewer", "bigquery_data_viewer", BIGQUERY_DATA_VIEWER_ROLE, "roles/bigquery.jobUser", "bigquery.jobUser", "job_user", }) FORBIDDEN_APPROVAL_FLOW_ALIASES: Final = frozenset({ "cloud_scope_request", "cmdb_data_access", "bigquery_access_request", }) # Primary email and aliases are useful lookup/input material, but not durable # semantic principal keys. Bindings/requests use ``email_groups.id``. FORBIDDEN_SEMANTIC_PRINCIPAL_FIELDS: Final = frozenset({ "email", "primary_email", "principal_email", "principal_email_snapshot", "group_email", "google_group_email", "alias", "aliases", "principal_alias", }) # Provider members come from the compiled primary-email snapshot, never from # aliases. Alias-derived IAM members would grant the same Google group through a # non-authoritative surface and make rename/revoke accounting ambiguous. FORBIDDEN_PROVIDER_MEMBER_EMAIL_FIELDS: Final = frozenset({ "alias", "aliases", "principal_alias", "principal_alias_snapshot", "google_group_alias", }) # Runtime authorization reads the projected resource row. Raw catalog/bootstrap # owner material is import input/provenance, not a second policy engine. FORBIDDEN_RUNTIME_OWNER_AUTHORITY_FIELDS: Final = frozenset({ "catalog_owner_person_id", "catalog_owner_email", "bootstrap_owner_person_id", "bootstrap_owner_email", "raw_catalog_owner", "raw_catalog_owner_person_id", "raw_bootstrap_owner", "raw_bootstrap_owner_person_id", "source_owner_person_id", "source_owner_email", }) FORBIDDEN_BY_AXIS: Final = { "domain": FORBIDDEN_DOMAIN_ALIASES, "provider": FORBIDDEN_PROVIDER_ALIASES, "resource_type": FORBIDDEN_RESOURCE_TYPE_ALIASES, "principal_type": FORBIDDEN_PRINCIPAL_TYPE_ALIASES, "entitlement": FORBIDDEN_ENTITLEMENT_ALIASES, "approval_flow": FORBIDDEN_APPROVAL_FLOW_ALIASES, "semantic_principal_field": FORBIDDEN_SEMANTIC_PRINCIPAL_FIELDS, "provider_member_email_field": FORBIDDEN_PROVIDER_MEMBER_EMAIL_FIELDS, "runtime_owner_authority_field": FORBIDDEN_RUNTIME_OWNER_AUTHORITY_FIELDS, } _CANONICAL_BY_AXIS: Final = { "domain": ALL_DOMAINS, "provider": ALL_PROVIDERS, "resource_type": ALL_RESOURCE_TYPES, "principal_type": ALL_PRINCIPAL_TYPES, "entitlement": ALL_ENTITLEMENTS, "approval_flow": ALL_APPROVAL_FLOWS, "semantic_principal_field": (SEMANTIC_PRINCIPAL_FIELD,), "provider_member_email_field": (PROVIDER_MEMBER_EMAIL_FIELD,), "runtime_owner_authority_field": (RUNTIME_OWNER_AUTHORITY_FIELD,), } def assert_canonical(axis: str, value: str) -> str: """Return ``value`` only when it is canonical for ``axis``. This helper is intentionally small and ValueError-based so Pydantic validators, service guards, and unit tests can reuse it without importing FastAPI or application exceptions. """ try: allowed = _CANONICAL_BY_AXIS[axis] except KeyError as exc: raise ValueError(f"unknown data-access canonical axis: {axis}") from exc forbidden = FORBIDDEN_BY_AXIS.get(axis, frozenset()) if value in allowed: return value if value in forbidden: raise ValueError( f"{value!r} is a forbidden data-access {axis} alias; " f"use one of {allowed!r}", ) raise ValueError( f"{value!r} is not a supported data-access {axis}; " f"use one of {allowed!r}", ) def assert_semantic_principal_field(value: str) -> str: return assert_canonical("semantic_principal_field", value) def assert_provider_member_email_field(value: str) -> str: return assert_canonical("provider_member_email_field", value) def assert_runtime_owner_authority_field(value: str) -> str: return assert_canonical("runtime_owner_authority_field", value)