diff --git a/pyproject.toml b/pyproject.toml index 5b776df..599487c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,13 +4,13 @@ build-backend = "setuptools.build_meta" [project] name = "govoplan-datasources" -version = "0.1.19" +version = "0.1.20" description = "Governed datasource catalogue, staging, and materialization lifecycle for GovOPlaN." readme = "README.md" requires-python = ">=3.12" license = "AGPL-3.0-or-later" authors = [{ name = "GovOPlaN" }] -dependencies = ["govoplan-core>=0.1.18"] +dependencies = ["govoplan-core>=0.1.20"] [tool.setuptools.packages.find] where = ["src"] diff --git a/src/govoplan_datasources/__init__.py b/src/govoplan_datasources/__init__.py index 9c78b7b..727e364 100644 --- a/src/govoplan_datasources/__init__.py +++ b/src/govoplan_datasources/__init__.py @@ -1,3 +1,3 @@ """GovOPlaN Datasources module.""" -__version__ = "0.1.19" +__version__ = "0.1.20" diff --git a/src/govoplan_datasources/backend/db/models.py b/src/govoplan_datasources/backend/db/models.py index 02791ab..b2a856b 100644 --- a/src/govoplan_datasources/backend/db/models.py +++ b/src/govoplan_datasources/backend/db/models.py @@ -109,6 +109,10 @@ class DatasourceRecord(Base, TimestampMixin): ) privacy_profile_ref: Mapped[str | None] = mapped_column(String(500), nullable=True) retention_policy_ref: Mapped[str | None] = mapped_column(String(500), nullable=True) + access_policy_ref: Mapped[str | None] = mapped_column(String(500), nullable=True) + visibility_policy: Mapped[dict[str, Any]] = mapped_column( + JSON, default=dict, nullable=False + ) hold_refs: Mapped[list[str]] = mapped_column(JSON, default=list, nullable=False) publication_state: Mapped[str] = mapped_column( String(50), default="draft", nullable=False, index=True diff --git a/src/govoplan_datasources/backend/manifest.py b/src/govoplan_datasources/backend/manifest.py index 6bb5827..effcdf4 100644 --- a/src/govoplan_datasources/backend/manifest.py +++ b/src/govoplan_datasources/backend/manifest.py @@ -12,6 +12,7 @@ from govoplan_core.core.datasources import ( CAPABILITY_DATASOURCE_LIFECYCLE, CAPABILITY_DATASOURCE_ORIGINS, CAPABILITY_DATASOURCE_PUBLICATION, + CAPABILITY_POLICY_DATASOURCE_VISIBILITY, datasource_artifact_backend_provider, ) from govoplan_core.core.module_guards import ( @@ -62,7 +63,7 @@ from govoplan_datasources.backend.payloads import ExternalArtifactPayloadBackend MODULE_ID = "datasources" MODULE_NAME = "Datasources" -MODULE_VERSION = "0.1.19" +MODULE_VERSION = "0.1.20" DATASOURCE_INTERFACE_VERSION = "0.2.0" ARCHITECTURE = ModuleArchitectureDeclaration( @@ -256,6 +257,7 @@ manifest = ModuleManifest( CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_DATASOURCE_ORIGINS, CAPABILITY_DATASOURCE_ARTIFACT_BACKENDS, + CAPABILITY_POLICY_DATASOURCE_VISIBILITY, ), provides_interfaces=( ModuleInterfaceProvider( @@ -293,6 +295,12 @@ manifest = ModuleManifest( version_max_exclusive="2.0.0", optional=True, ), + ModuleInterfaceRequirement( + name="policy.datasource_visibility", + version_min="1.0.0", + version_max_exclusive="2.0.0", + optional=True, + ), ), permissions=PERMISSIONS, role_templates=ROLE_TEMPLATES, @@ -547,10 +555,40 @@ manifest = ModuleManifest( "datasources.field.freshness-policy", "datasources.field.quality-policy", "datasources.field.retention-policy", + "datasources.field.access-policy", + "datasources.field.visibility-policy", "datasources.field.transfer-agreement", ], }, ), + DocumentationTopic( + id="datasources.visibility", + title="Datasource and field visibility", + summary="Apply source, materialization, field, and row controls before protected rows leave Datasources.", + body=( + "A local visibility policy can restrict discovery and reads by account, membership, identity, group, role, service account, or authentication method. " + "Materialization ACLs additionally protect current and frozen revisions. Field rules classify a field and either replace its value with null or omit the field unless an allow ACL matches. Row filters compare a configured field with a principal account, membership, identity, service-account, group, role, or function-assignment claim; all configured filters and all policy overlays must allow a row. " + "Datasources applies these controls before rows leave its provider boundary and returns an opaque permitted-view fingerprint bound to the base revision, effective policies, and relevant principal facts. Denied and filtered reads emit audit evidence containing outcomes and policy diagnostics, never protected values, provider locators, arbitrary metadata, or provenance bodies. " + "An optional access-policy reference delegates hierarchical overlays to Policy. Policy can only tighten local behavior. An unresolved reference, unavailable configured provider, or malformed policy fails closed. Without Policy and without a reference, the local visibility policy remains fully usable. Frozen reads enforce the current local policy together with a distinct restrictive policy captured in the frozen governance snapshot." + ), + layer="available", + documentation_types=("admin", "user"), + audience=("operator", "module_admin", "data_steward", "auditor"), + related_modules=("policy", "access", "audit", "connectors"), + order=72, + metadata={ + "seed": True, + "help_contexts": [ + "datasources.field.access-policy", + "datasources.field.visibility-policy", + "datasources.preview", + ], + "limitations": [ + "Row filters support deterministic equality or membership against principal claims; arbitrary expressions are not executed.", + "Live filtered previews scan at most 10,000 origin rows and disclose a structured limitation diagnostic if the origin is larger.", + ], + }, + ), DocumentationTopic( id="datasources.quality-gates", title="Validate staged and produced Datasource revisions", @@ -585,7 +623,7 @@ manifest = ModuleManifest( "dataflow", "workflow_engine", ), - order=72, + order=73, metadata={ "seed": True, "help_contexts": [ @@ -624,7 +662,7 @@ manifest = ModuleManifest( "reporting", "audit", ), - order=73, + order=74, metadata={ "seed": True, "help_contexts": [ diff --git a/src/govoplan_datasources/backend/migrations/versions/c9e3a6f1d4b8_v0120_datasource_visibility.py b/src/govoplan_datasources/backend/migrations/versions/c9e3a6f1d4b8_v0120_datasource_visibility.py new file mode 100644 index 0000000..9768a33 --- /dev/null +++ b/src/govoplan_datasources/backend/migrations/versions/c9e3a6f1d4b8_v0120_datasource_visibility.py @@ -0,0 +1,38 @@ +"""v0.1.20 datasource visibility policy + +Revision ID: c9e3a6f1d4b8 +Revises: b8d2f5a0c3e7 +Create Date: 2026-08-21 20:15:00.000000 +""" + +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "c9e3a6f1d4b8" +down_revision = "b8d2f5a0c3e7" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column( + "datasource_catalogue", + sa.Column("access_policy_ref", sa.String(length=500), nullable=True), + ) + op.add_column( + "datasource_catalogue", + sa.Column( + "visibility_policy", + sa.JSON(), + server_default=sa.text("'{}'"), + nullable=False, + ), + ) + + +def downgrade() -> None: + op.drop_column("datasource_catalogue", "visibility_policy") + op.drop_column("datasource_catalogue", "access_policy_ref") diff --git a/src/govoplan_datasources/backend/router.py b/src/govoplan_datasources/backend/router.py index 26cf775..4676d49 100644 --- a/src/govoplan_datasources/backend/router.py +++ b/src/govoplan_datasources/backend/router.py @@ -1,5 +1,7 @@ from __future__ import annotations +import hashlib +import json from collections.abc import Mapping from fastapi import APIRouter, Depends, HTTPException, Query, status @@ -370,7 +372,17 @@ def api_update_datasource_governance( action="datasources.governance.updated", object_type="datasource", object_id=item.ref, - details={"governance": item.governance.to_dict()}, + details={ + "classification": item.governance.classification, + "publication_state": item.governance.publication_state, + "access_policy_ref": item.governance.access_policy_ref, + "visibility_policy_configured": bool( + item.governance.visibility_policy + ), + "visibility_policy_hash": _mapping_hash( + item.governance.visibility_policy + ), + }, ) session.commit() return _datasource_response(item) @@ -580,6 +592,7 @@ def _datasource_response(item: DatasourceDescriptor) -> DatasourceResponse: name=field.name, data_type=field.data_type, nullable=field.nullable, + classification=field.classification, ) for field in item.schema ], @@ -610,6 +623,7 @@ def _materialization_response( name=field.name, data_type=field.data_type, nullable=field.nullable, + classification=field.classification, ) for field in item.schema ], @@ -643,6 +657,7 @@ def _stage_response(item: DatasourceStage) -> DatasourceStageResponse: name=field.name, data_type=field.data_type, nullable=field.nullable, + classification=field.classification, ) for field in item.schema ], @@ -673,6 +688,7 @@ def _origin_response(item: DatasourceOrigin) -> DatasourceOriginResponse: name=field.name, data_type=field.data_type, nullable=field.nullable, + classification=field.classification, ) for field in item.schema ], @@ -715,6 +731,13 @@ def _safe_mapping(value: object) -> Mapping[str, object]: return value if isinstance(value, Mapping) else {} +def _mapping_hash(value: Mapping[str, object]) -> str | None: + if not value: + return None + encoded = json.dumps(value, sort_keys=True, separators=(",", ":"), default=str) + return hashlib.sha256(encoded.encode("utf-8")).hexdigest() + + def _audit( session: Session, principal: ApiPrincipal, diff --git a/src/govoplan_datasources/backend/schemas.py b/src/govoplan_datasources/backend/schemas.py index 6de1ae7..2fb7d30 100644 --- a/src/govoplan_datasources/backend/schemas.py +++ b/src/govoplan_datasources/backend/schemas.py @@ -43,6 +43,8 @@ class DatasourceGovernancePayload(BaseModel): classification: str = Field(default="internal", min_length=1, max_length=80) privacy_profile_ref: str | None = Field(default=None, max_length=500) retention_policy_ref: str | None = Field(default=None, max_length=500) + access_policy_ref: str | None = Field(default=None, max_length=500) + visibility_policy: dict[str, Any] = Field(default_factory=dict) hold_refs: list[str] = Field(default_factory=list, max_length=100) publication_state: str = Field(default="draft", min_length=1, max_length=50) transfer_agreement_ref: str | None = Field(default=None, max_length=500) @@ -70,6 +72,7 @@ class DatasourceFieldResponse(BaseModel): name: str data_type: str nullable: bool + classification: str = "internal" class DatasourceResponse(BaseModel): diff --git a/src/govoplan_datasources/backend/service.py b/src/govoplan_datasources/backend/service.py index 3f6170a..7fb24e2 100644 --- a/src/govoplan_datasources/backend/service.py +++ b/src/govoplan_datasources/backend/service.py @@ -33,7 +33,9 @@ from govoplan_core.core.datasources import ( DatasourceStageInput, DatasourceUnavailableError, DatasourceValidationError, + DatasourceVisibilityPolicyRequest, datasource_origins, + datasource_visibility_policy_provider, ) from govoplan_core.db.base import utcnow from govoplan_datasources.backend.db.models import ( @@ -62,6 +64,14 @@ from govoplan_datasources.backend.tabular import ( infer_schema, normalize_rows, ) +from govoplan_datasources.backend.visibility import ( + VisibilityPlan, + apply_visibility, + normalize_visibility_policy, + visibility_diagnostics, + visibility_plan, + visible_schema, +) CATALOGUE_READ_SCOPE = "datasources:catalogue:read" @@ -335,7 +345,15 @@ class SqlDatasourceProvider: DatasourceGovernanceReferenceRecord.reference == cleaned, ) ) - return tuple(_datasource_dto(item) for item in db.scalars(statement)) + result: list[DatasourceDescriptor] = [] + for item in db.scalars(statement): + if has_scope(api_principal, ADMIN_SCOPE) or self._can_discover( + db, + api_principal, + item=item, + ): + result.append(_datasource_dto(item)) + return tuple(result) def get_datasource( self, @@ -350,7 +368,15 @@ class SqlDatasourceProvider: tenant_id=api_principal.tenant_id, datasource_ref=datasource_ref, ) - return _datasource_dto(item) if item is not None else None + if item is None: + return None + if not has_scope(api_principal, ADMIN_SCOPE) and not self._can_discover( + db, + api_principal, + item=item, + ): + return None + return _datasource_dto(item) def read_datasource( self, @@ -374,41 +400,67 @@ class SqlDatasourceProvider: and not request.materialization_ref and request.consistency == "current" ) - if read_live: - if not item.provider_ref: - raise DatasourceUnavailableError( - "This datasource has no live origin." + try: + if read_live: + if not item.provider_ref: + raise DatasourceUnavailableError( + "This datasource has no live origin." + ) + result = self._read_live( + db, + api_principal, + item=item, + request=request, + limit=limit, + offset=offset, + columns=columns, ) - result = self._read_live( + else: + result = self._read_materialized( + db, + api_principal, + item=item, + request=request, + limit=limit, + offset=offset, + columns=columns, + ) + except DatasourceAccessError: + _audit_visibility_read( db, api_principal, item=item, - request=request, - limit=limit, - offset=offset, - columns=columns, + outcome="denied", + consistency=request.consistency, ) - if ( - request.expected_fingerprint - and request.expected_fingerprint != result.datasource.fingerprint - ): - raise DatasourceValidationError( - "The live datasource fingerprint changed; refresh the consuming definition." - ) - return result - - return self._read_materialized( - db, - item=item, - request=request, - limit=limit, - offset=offset, - columns=columns, + raise + if ( + request.expected_fingerprint + and request.expected_fingerprint != result.datasource.fingerprint + ): + raise DatasourceValidationError( + "The datasource permitted-view fingerprint changed; refresh the consuming definition." + ) + visibility_evidence = tuple( + item + for item in result.diagnostics + if item.code.startswith("datasource.visibility_") ) + if visibility_evidence: + _audit_visibility_read( + db, + api_principal, + item=item, + outcome="filtered", + consistency=request.consistency, + diagnostics=visibility_evidence, + ) + return result def _read_materialized( self, session: Session, + principal: ApiPrincipal, *, item: DatasourceRecord, request: DatasourceReadRequest, @@ -428,41 +480,114 @@ class SqlDatasourceProvider: else "This datasource has no materialized state." ) raise DatasourceUnavailableError(message) - if ( - request.expected_fingerprint - and request.expected_fingerprint != materialization.fingerprint - ): - raise DatasourceValidationError( - "The datasource fingerprint changed; refresh the consuming definition." - ) - _validate_columns(materialization.schema_, columns) + source_schema = _fields(materialization.schema_) + snapshot_governance = DatasourceGovernance.from_mapping( + materialization.governance_snapshot_ + ) + snapshot_policies = ( + (snapshot_governance.visibility_policy,) + if snapshot_governance.visibility_policy + and snapshot_governance.visibility_policy + != _datasource_governance(item).visibility_policy + else () + ) + plan = self._visibility_plan( + session, + principal, + item=item, + request=request, + materialized=True, + schema=source_schema, + additional_policies=snapshot_policies, + ) + permitted_schema = visible_schema(source_schema, plan) + _validate_field_names(permitted_schema, columns) payload = payload_for_materialization(session, materialization) - if payload is None: - window = materialization.rows[offset : offset + limit] - else: - backend = self._payload_backends.require(payload.backend) - backend.verify(session, payload) - window = backend.read_rows( - session, - payload, - offset=offset, - limit=limit, + if plan.applied: + if payload is None: + scanned_rows = tuple(materialization.rows[:MAX_STAGE_ROWS]) + else: + backend = self._payload_backends.require(payload.backend) + backend.verify(session, payload) + scanned_rows = tuple( + backend.read_rows( + session, + payload, + offset=0, + limit=MAX_STAGE_ROWS, + ) + ) + permitted_rows, filtered_rows = apply_visibility( + scanned_rows, + principal=principal.principal, + plan=plan, + columns=columns, ) - rows = tuple(_select_columns(row, columns) for row in window) + candidate_rows = permitted_rows[offset : offset + limit] + rows = _bounded_result_rows( + candidate_rows, + max_bytes=request.max_bytes, + ) + byte_limited = len(rows) < len(candidate_rows) + total_rows = len(permitted_rows) + scan_limited = len(scanned_rows) < materialization.row_count + else: + if payload is None: + window = materialization.rows[offset : offset + limit] + else: + backend = self._payload_backends.require(payload.backend) + backend.verify(session, payload) + window = backend.read_rows( + session, + payload, + offset=offset, + limit=limit, + ) + rows = tuple(_select_columns(row, columns) for row in window) + scanned_rows = rows + total_rows = materialization.row_count + filtered_rows = 0 + scan_limited = False + byte_limited = False + view_fingerprint = plan.view_fingerprint(materialization.fingerprint) descriptor = _datasource_dto(item) + result_descriptor = replace( + descriptor, + schema=permitted_schema, + schema_version=str(materialization.schema_version), + fingerprint=view_fingerprint, + row_count=total_rows, + byte_count=encoded_size(rows), + ) + materialization_dto = replace( + _materialization_dto(materialization), + schema=permitted_schema, + fingerprint=view_fingerprint, + row_count=total_rows, + byte_count=encoded_size(rows), + ) + if plan.applied: + result_descriptor = _opaque_descriptor(result_descriptor) + materialization_dto = _opaque_materialization(materialization_dto) return DatasourceReadResult( - datasource=replace( - descriptor, - schema=_fields(materialization.schema_), - schema_version=str(materialization.schema_version), - fingerprint=materialization.fingerprint, - row_count=materialization.row_count, - byte_count=materialization.byte_count, - ), + datasource=result_descriptor, rows=rows, - total_rows=materialization.row_count, - truncated=offset + len(rows) < materialization.row_count, - materialization=_materialization_dto(materialization), + total_rows=total_rows, + truncated=( + scan_limited or byte_limited or offset + len(rows) < total_rows + ), + materialization=materialization_dto, + returned_bytes=encoded_size(rows), + effective_row_limit=limit, + effective_byte_limit=request.max_bytes, + effective_timeout_ms=request.timeout_ms, + diagnostics=visibility_diagnostics( + plan, + scanned_rows=len(scanned_rows), + permitted_rows=total_rows, + filtered_rows=filtered_rows, + scan_limited=scan_limited, + ), ) def list_materializations( @@ -486,7 +611,46 @@ class SqlDatasourceProvider: ) .order_by(DatasourceMaterializationRecord.revision.desc()) ) - return tuple(_materialization_dto(row) for row in db.scalars(statement)) + result: list[DatasourceMaterialization] = [] + for row in db.scalars(statement): + schema = _fields(row.schema_) + snapshot = DatasourceGovernance.from_mapping(row.governance_snapshot_) + additional_policies = ( + (snapshot.visibility_policy,) + if snapshot.visibility_policy + and snapshot.visibility_policy + != _datasource_governance(item).visibility_policy + else () + ) + try: + plan = self._visibility_plan( + db, + api_principal, + item=item, + request=DatasourceReadRequest( + datasource_ref=datasource_ref, + materialization_ref=_materialization_ref(row.id), + consistency="frozen", + ), + materialized=True, + schema=schema, + additional_policies=additional_policies, + ) + except DatasourceAccessError: + continue + dto = _materialization_dto(row) + if plan.applied: + dto = _opaque_materialization( + replace( + dto, + schema=visible_schema(schema, plan), + fingerprint=plan.view_fingerprint(row.fingerprint), + row_count=None, + byte_count=None, + ) + ) + result.append(dto) + return tuple(result) def list_stages( self, @@ -551,6 +715,13 @@ class SqlDatasourceProvider: ) rows = normalize_rows(stage.rows) schema = infer_schema(rows) + governance = replace( + governance, + visibility_policy=normalize_visibility_policy( + governance.visibility_policy, + schema=schema, + ), + ) fingerprint = fingerprint_rows(rows, schema) validation = validate_stage( rows=rows, @@ -919,6 +1090,26 @@ class SqlDatasourceProvider: raise DatasourceUnavailableError( "No connector origin provider is currently available." ) + plan = self._visibility_plan( + session, + principal, + item=item, + request=request, + materialized=False, + schema=_fields(item.schema_), + ) + if plan.applied: + return self._read_live_visible( + session, + principal, + provider=provider, + item=item, + request=request, + plan=plan, + limit=limit, + offset=offset, + columns=columns, + ) try: result = provider.read_origin( session, @@ -962,6 +1153,203 @@ class SqlDatasourceProvider: diagnostics=result.diagnostics, ) + def _read_live_visible( + self, + session: Session, + principal: ApiPrincipal, + *, + provider: object, + item: DatasourceRecord, + request: DatasourceReadRequest, + plan: VisibilityPlan, + limit: int, + offset: int, + columns: tuple[str, ...], + ) -> DatasourceReadResult: + rows: list[Mapping[str, object]] = [] + first_result = None + last_result = None + base_fingerprint: str | None = None + elapsed_ms = 0 + while len(rows) < MAX_STAGE_ROWS: + page_limit = min(MAX_READ_ROWS, MAX_STAGE_ROWS - len(rows)) + try: + page = provider.read_origin( + session, + principal, + request=DatasourceOriginReadRequest( + origin_ref=str(item.provider_ref), + limit=page_limit, + offset=len(rows), + expected_fingerprint=base_fingerprint, + max_bytes=request.max_bytes, + timeout_ms=request.timeout_ms, + ), + ) + except DatasourceError: + raise + except Exception as exc: + raise DatasourceUnavailableError( + f"The datasource origin could not be read: {exc}" + ) from exc + if first_result is None: + first_result = page + base_fingerprint = page.origin.fingerprint + last_result = page + elapsed_ms += page.elapsed_ms + rows.extend(page.rows) + if not page.truncated or not page.rows: + break + if first_result is None or last_result is None: + raise DatasourceUnavailableError( + "The datasource origin returned no result." + ) + source_schema = first_result.origin.schema + plan = self._visibility_plan( + session, + principal, + item=item, + request=request, + materialized=False, + schema=source_schema, + ) + permitted_schema = visible_schema(source_schema, plan) + _validate_field_names(permitted_schema, columns) + permitted, filtered_rows = apply_visibility( + rows, + principal=principal.principal, + plan=plan, + columns=columns, + ) + window = permitted[offset : offset + limit] + window = _bounded_result_rows(window, max_bytes=request.max_bytes) + scan_limited = bool( + last_result.truncated + and (len(rows) >= MAX_STAGE_ROWS or not last_result.rows) + ) + total_rows = len(permitted) + view_fingerprint = plan.view_fingerprint(first_result.origin.fingerprint) + descriptor = replace( + _datasource_dto(item), + schema=permitted_schema, + schema_version=first_result.origin.schema_version, + fingerprint=view_fingerprint, + row_count=total_rows, + byte_count=encoded_size(window), + updated_at=first_result.origin.updated_at, + ) + return DatasourceReadResult( + datasource=_opaque_descriptor(descriptor), + rows=window, + total_rows=total_rows, + truncated=( + scan_limited + or offset + len(window) < total_rows + or len(window) < min(limit, max(0, total_rows - offset)) + ), + returned_bytes=encoded_size(window), + elapsed_ms=elapsed_ms, + effective_row_limit=limit, + effective_byte_limit=request.max_bytes, + effective_timeout_ms=request.timeout_ms, + diagnostics=( + *first_result.diagnostics, + *visibility_diagnostics( + plan, + scanned_rows=len(rows), + permitted_rows=total_rows, + filtered_rows=filtered_rows, + scan_limited=scan_limited, + ), + ), + ) + + def _can_discover( + self, + session: Session, + principal: ApiPrincipal, + *, + item: DatasourceRecord, + ) -> bool: + try: + self._visibility_plan( + session, + principal, + item=item, + request=DatasourceReadRequest(datasource_ref=_datasource_ref(item.id)), + materialized=False, + schema=_fields(item.schema_), + action="discover", + ) + except (DatasourceAccessError, DatasourceValidationError): + return False + return True + + def _visibility_plan( + self, + session: Session, + principal: ApiPrincipal, + *, + item: DatasourceRecord, + request: DatasourceReadRequest, + materialized: bool, + schema: Sequence[DatasourceField], + action: str = "read", + additional_policies: Sequence[Mapping[str, object]] = (), + ) -> VisibilityPlan: + governance = _datasource_governance(item) + policies: list[Mapping[str, object]] = [] + if governance.visibility_policy: + policies.append(governance.visibility_policy) + policies.extend(additional_policies) + decision_refs: list[str] = [] + policy_provider = datasource_visibility_policy_provider(self._registry) + if policy_provider is None: + if governance.access_policy_ref: + raise DatasourceAccessError( + "The configured datasource access policy provider is unavailable." + ) + else: + try: + decision = policy_provider.decide_datasource_visibility( + session, + request=DatasourceVisibilityPolicyRequest( + tenant_id=principal.tenant_id, + datasource_ref=_datasource_ref(item.id), + principal=principal.principal, + action=cast(Any, action), + classification=governance.classification, + policy_ref=governance.access_policy_ref, + consistency=request.consistency, + materialization_ref=request.materialization_ref, + ), + ) + except DatasourceError: + raise + except Exception as exc: + raise DatasourceAccessError( + "Datasource policy evaluation is unavailable." + ) from exc + if not decision.allowed: + raise DatasourceAccessError( + decision.reason or "Datasource policy denied access." + ) + policies.extend(decision.policies) + if decision.decision_ref: + decision_refs.append(decision.decision_ref) + try: + return visibility_plan( + principal=principal.principal, + policies=policies, + materialized=materialized, + schema=schema, + decision_refs=decision_refs, + ) + except DatasourceValidationError as exc: + raise DatasourceAccessError( + "Datasource visibility policy is invalid and access was denied." + ) from exc + def _required_origin( self, session: Session, @@ -1422,6 +1810,8 @@ def _datasource_governance(item: DatasourceRecord) -> DatasourceGovernance: "classification": item.classification, "privacy_profile_ref": item.privacy_profile_ref, "retention_policy_ref": item.retention_policy_ref, + "access_policy_ref": item.access_policy_ref, + "visibility_policy": item.visibility_policy, "hold_refs": item.hold_refs, "publication_state": item.publication_state, "transfer_agreement_ref": item.transfer_agreement_ref, @@ -1453,6 +1843,13 @@ def _apply_datasource_governance( item.classification = governance.classification item.privacy_profile_ref = governance.privacy_profile_ref item.retention_policy_ref = governance.retention_policy_ref + item.access_policy_ref = governance.access_policy_ref + item.visibility_policy = dict( + normalize_visibility_policy( + governance.visibility_policy, + schema=_fields(item.schema_), + ) + ) item.hold_refs = list(governance.hold_refs) item.publication_state = governance.publication_state item.transfer_agreement_ref = governance.transfer_agreement_ref @@ -1569,6 +1966,7 @@ def _fields(payload: Sequence[Mapping[str, object]]) -> tuple[DatasourceField, . name=str(item.get("name", "")), data_type=str(item.get("data_type", "unknown")), nullable=bool(item.get("nullable", True)), + classification=str(item.get("classification") or "internal"), ) for item in payload ) @@ -1617,6 +2015,109 @@ def _validate_columns( ) +def _validate_field_names( + schema: Sequence[DatasourceField], + columns: tuple[str, ...], +) -> None: + _validate_columns( + tuple({"name": field.name} for field in schema), + columns, + ) + + +def _bounded_result_rows( + rows: Sequence[Mapping[str, object]], + *, + max_bytes: int, +) -> tuple[Mapping[str, object], ...]: + result: list[Mapping[str, object]] = [] + byte_limit = max(1, int(max_bytes)) + for row in rows: + candidate = (*result, row) + if encoded_size(candidate) > byte_limit: + break + result.append(row) + return tuple(result) + + +def _opaque_descriptor(item: DatasourceDescriptor) -> DatasourceDescriptor: + return replace( + item, + provider=None, + provider_ref=None, + provenance={"visibility": "policy_filtered"}, + metadata={}, + governance=DatasourceGovernance( + classification=item.governance.classification, + publication_state=item.governance.publication_state, + ), + ) + + +def _opaque_materialization( + item: DatasourceMaterialization, +) -> DatasourceMaterialization: + return replace( + item, + provenance={"visibility": "policy_filtered"}, + metadata={}, + governance=DatasourceGovernance( + classification=item.governance.classification, + publication_state=item.governance.publication_state, + ), + ) + + +def _audit_visibility_read( + session: Session, + principal: ApiPrincipal, + *, + item: DatasourceRecord, + outcome: str, + consistency: str, + diagnostics: Sequence[object] = (), +) -> None: + visibility_summary: dict[str, object] = {} + for diagnostic in diagnostics: + if getattr(diagnostic, "code", None) != "datasource.visibility_applied": + continue + details = getattr(diagnostic, "details", {}) + if isinstance(details, Mapping): + visibility_summary = { + key: details[key] + for key in ( + "policy_count", + "field_rule_count", + "row_filter_count", + "scanned_rows", + "permitted_rows", + "filtered_rows", + ) + if key in details + } + audit_event( + session, + tenant_id=principal.tenant_id, + user_id=getattr(principal.user, "id", None) or principal.account_id, + api_key_id=principal.api_key_id, + action=f"datasource.read.{outcome}", + object_type="datasource", + object_id=item.id, + details={ + "outcome": outcome, + "consistency": consistency, + "policy_ref_configured": bool(item.access_policy_ref), + "local_policy_configured": bool(item.visibility_policy), + "diagnostic_codes": [ + str(getattr(diagnostic, "code", "")) + for diagnostic in diagnostics + if getattr(diagnostic, "code", None) + ], + "visibility_summary": visibility_summary, + }, + ) + + def _select_columns( row: Mapping[str, object], columns: tuple[str, ...], diff --git a/src/govoplan_datasources/backend/tabular.py b/src/govoplan_datasources/backend/tabular.py index 5841267..3443a17 100644 --- a/src/govoplan_datasources/backend/tabular.py +++ b/src/govoplan_datasources/backend/tabular.py @@ -145,6 +145,7 @@ def field_payload(field: DatasourceField) -> dict[str, object]: "name": field.name, "data_type": field.data_type, "nullable": field.nullable, + "classification": field.classification, } diff --git a/src/govoplan_datasources/backend/visibility.py b/src/govoplan_datasources/backend/visibility.py new file mode 100644 index 0000000..4ce1572 --- /dev/null +++ b/src/govoplan_datasources/backend/visibility.py @@ -0,0 +1,490 @@ +from __future__ import annotations + +import hashlib +import json +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from typing import cast + +from govoplan_core.core.access import PrincipalRef +from govoplan_core.core.datasources import ( + DatasourceAccessError, + DatasourceField, + DatasourceValidationError, +) +from govoplan_core.core.tabular_sources import TabularPreviewDiagnostic + + +_ACL_KEYS = frozenset( + { + "account_ids", + "membership_ids", + "identity_ids", + "group_ids", + "role_ids", + "service_account_ids", + "auth_methods", + } +) +_CLAIM_KEYS = frozenset( + { + "account_id", + "membership_id", + "identity_id", + "service_account_id", + "group_ids", + "role_ids", + "function_assignment_ids", + } +) +_POLICY_KEYS = frozenset({"source_acl", "materialization_acl", "fields", "row_filters"}) + + +@dataclass(frozen=True, slots=True) +class FieldRule: + name: str + classification: str + action: str + allowed: bool + + +@dataclass(frozen=True, slots=True) +class RowFilter: + field: str + claim: str + operator: str + allow_null: bool + + +@dataclass(frozen=True, slots=True) +class VisibilityPlan: + applied: bool + policy_hashes: tuple[str, ...] + field_rules: tuple[FieldRule, ...] + row_filters: tuple[RowFilter, ...] + principal_fingerprint: str + decision_refs: tuple[str, ...] = () + + def view_fingerprint(self, base_fingerprint: str) -> str: + if not self.applied: + return base_fingerprint + return _digest( + { + "base_fingerprint": base_fingerprint, + "policy_hashes": self.policy_hashes, + "principal_fingerprint": self.principal_fingerprint, + "decision_refs": self.decision_refs, + } + ) + + +def visibility_plan( + *, + principal: PrincipalRef, + policies: Sequence[Mapping[str, object]], + materialized: bool, + schema: Sequence[DatasourceField], + decision_refs: Sequence[str] = (), +) -> VisibilityPlan: + normalized = tuple(_normalize_policy(policy) for policy in policies if policy) + if not normalized: + return VisibilityPlan( + applied=False, + policy_hashes=(), + field_rules=(), + row_filters=(), + principal_fingerprint="", + ) + + schema_names = {field.name for field in schema} + field_rules: list[FieldRule] = [] + row_filters: list[RowFilter] = [] + for policy in normalized: + source_acl = policy.get("source_acl") + if isinstance(source_acl, Mapping) and not _acl_allows(source_acl, principal): + raise DatasourceAccessError("Datasource visibility policy denied access.") + materialization_acl = policy.get("materialization_acl") + if ( + materialized + and isinstance(materialization_acl, Mapping) + and not _acl_allows(materialization_acl, principal) + ): + raise DatasourceAccessError( + "Datasource materialization policy denied access." + ) + fields = policy.get("fields", {}) + if isinstance(fields, Mapping): + for name, raw_rule in fields.items(): + if name not in schema_names: + raise DatasourceValidationError( + f"Visibility policy references unknown field {name!r}." + ) + assert isinstance(raw_rule, Mapping) + field_rules.append( + FieldRule( + name=name, + classification=str(raw_rule["classification"]), + action=str(raw_rule["action"]), + allowed=_acl_allows(raw_rule["allow"], principal), + ) + ) + for raw_filter in policy.get("row_filters", ()): + assert isinstance(raw_filter, Mapping) + field = str(raw_filter["field"]) + if field not in schema_names: + raise DatasourceValidationError( + f"Visibility policy references unknown row-filter field {field!r}." + ) + row_filters.append( + RowFilter( + field=field, + claim=str(raw_filter["claim"]), + operator=str(raw_filter["operator"]), + allow_null=bool(raw_filter["allow_null"]), + ) + ) + + all_facts = _principal_policy_facts(principal) + relevant_fact_keys = _relevant_policy_fact_keys(normalized) + principal_payload = {key: all_facts[key] for key in sorted(relevant_fact_keys)} + return VisibilityPlan( + applied=True, + policy_hashes=tuple(_digest(policy) for policy in normalized), + field_rules=tuple(field_rules), + row_filters=tuple(row_filters), + principal_fingerprint=_digest(principal_payload), + decision_refs=tuple(sorted(set(decision_refs))), + ) + + +def normalize_visibility_policy( + policy: Mapping[str, object], + *, + schema: Sequence[DatasourceField] = (), +) -> Mapping[str, object]: + normalized = _normalize_policy(policy) + if schema: + known = {field.name for field in schema} + configured = set(cast(Mapping[str, object], normalized.get("fields", {}))) + configured.update( + str(item["field"]) + for item in cast( + Sequence[Mapping[str, object]], normalized.get("row_filters", ()) + ) + ) + unknown = sorted(configured - known) + if unknown: + raise DatasourceValidationError( + f"Visibility policy references unknown fields: {', '.join(unknown)}." + ) + return normalized + + +def visible_schema( + schema: Sequence[DatasourceField], + plan: VisibilityPlan, +) -> tuple[DatasourceField, ...]: + classifications: dict[str, str] = {} + omitted: set[str] = set() + for rule in plan.field_rules: + classifications[rule.name] = rule.classification + if not rule.allowed and rule.action == "omit": + omitted.add(rule.name) + return tuple( + DatasourceField( + name=field.name, + data_type=field.data_type, + nullable=field.nullable, + classification=classifications.get(field.name, field.classification), + ) + for field in schema + if field.name not in omitted + ) + + +def apply_visibility( + rows: Sequence[Mapping[str, object]], + *, + principal: PrincipalRef, + plan: VisibilityPlan, + columns: Sequence[str] = (), +) -> tuple[tuple[Mapping[str, object], ...], int]: + omitted = { + rule.name + for rule in plan.field_rules + if not rule.allowed and rule.action == "omit" + } + redacted = { + rule.name + for rule in plan.field_rules + if not rule.allowed and rule.action == "redact" + } + requested = tuple(dict.fromkeys(columns)) + result: list[Mapping[str, object]] = [] + filtered = 0 + for source_row in rows: + if not all( + _row_filter_allows(source_row, item, principal) for item in plan.row_filters + ): + filtered += 1 + continue + names = requested or tuple(source_row) + result.append( + { + name: None if name in redacted else source_row.get(name) + for name in names + if name not in omitted + } + ) + return tuple(result), filtered + + +def visibility_diagnostics( + plan: VisibilityPlan, + *, + scanned_rows: int, + permitted_rows: int, + filtered_rows: int, + scan_limited: bool, +) -> tuple[TabularPreviewDiagnostic, ...]: + if not plan.applied: + return () + diagnostics = [ + TabularPreviewDiagnostic( + severity="info", + code="datasource.visibility_applied", + message="Datasource visibility policy was applied before rows left the provider.", + details={ + "policy_count": len(plan.policy_hashes), + "field_rule_count": len(plan.field_rules), + "row_filter_count": len(plan.row_filters), + "scanned_rows": scanned_rows, + "permitted_rows": permitted_rows, + "filtered_rows": filtered_rows, + }, + ) + ] + if scan_limited: + diagnostics.append( + TabularPreviewDiagnostic( + severity="warning", + code="datasource.visibility_scan_limited", + message="The bounded visibility scan ended before the origin was exhausted.", + details={"scanned_rows": scanned_rows}, + ) + ) + return tuple(diagnostics) + + +def _normalize_policy(policy: Mapping[str, object]) -> dict[str, object]: + unknown = set(policy) - _POLICY_KEYS + if unknown: + raise DatasourceValidationError( + f"Unsupported visibility policy keys: {', '.join(sorted(unknown))}." + ) + result: dict[str, object] = {} + for name in ("source_acl", "materialization_acl"): + if name in policy: + result[name] = _normalize_acl(policy[name], path=name) + + raw_fields = policy.get("fields", {}) + if not isinstance(raw_fields, Mapping) or len(raw_fields) > 500: + raise DatasourceValidationError( + "Visibility policy fields must be a mapping of at most 500 fields." + ) + fields: dict[str, object] = {} + for raw_name, raw_rule in sorted(raw_fields.items(), key=lambda item: str(item[0])): + name = str(raw_name).strip() + if not name or not isinstance(raw_rule, Mapping): + raise DatasourceValidationError( + "Every visibility field rule needs a field name and mapping." + ) + unknown_rule = set(raw_rule) - {"classification", "action", "allow"} + if unknown_rule: + raise DatasourceValidationError( + f"Unsupported visibility field keys for {name!r}: {', '.join(sorted(unknown_rule))}." + ) + classification = str(raw_rule.get("classification") or "restricted").strip() + if not classification or len(classification) > 80: + raise DatasourceValidationError( + "Field classifications must contain 1 to 80 characters." + ) + action = str(raw_rule.get("action") or "redact").strip() + if action not in {"redact", "omit"}: + raise DatasourceValidationError( + "Field visibility action must be redact or omit." + ) + if "allow" not in raw_rule: + raise DatasourceValidationError( + f"Visibility field rule {name!r} needs an allow ACL." + ) + fields[name] = { + "classification": classification, + "action": action, + "allow": _normalize_acl(raw_rule["allow"], path=f"fields.{name}.allow"), + } + if fields: + result["fields"] = fields + + raw_filters = policy.get("row_filters", ()) + if not isinstance(raw_filters, Sequence) or isinstance(raw_filters, (str, bytes)): + raise DatasourceValidationError("Visibility row_filters must be a list.") + if len(raw_filters) > 50: + raise DatasourceValidationError( + "Visibility policy supports at most 50 row filters." + ) + filters: list[dict[str, object]] = [] + for raw_filter in raw_filters: + if not isinstance(raw_filter, Mapping): + raise DatasourceValidationError( + "Every visibility row filter must be a mapping." + ) + unknown_filter = set(raw_filter) - {"field", "claim", "operator", "allow_null"} + if unknown_filter: + raise DatasourceValidationError( + f"Unsupported row-filter keys: {', '.join(sorted(unknown_filter))}." + ) + field = str(raw_filter.get("field") or "").strip() + claim = str(raw_filter.get("claim") or "").strip() + operator = str(raw_filter.get("operator") or "equals").strip() + if not field: + raise DatasourceValidationError( + "Every visibility row filter needs a field." + ) + if claim not in _CLAIM_KEYS: + raise DatasourceValidationError( + f"Unsupported visibility row-filter claim {claim!r}." + ) + if operator not in {"equals", "in"}: + raise DatasourceValidationError("Row-filter operator must be equals or in.") + filters.append( + { + "field": field, + "claim": claim, + "operator": operator, + "allow_null": bool(raw_filter.get("allow_null", False)), + } + ) + if filters: + result["row_filters"] = filters + return result + + +def _normalize_acl(value: object, *, path: str) -> dict[str, list[str]]: + if not isinstance(value, Mapping): + raise DatasourceValidationError( + f"Visibility policy {path} must be an ACL mapping." + ) + unknown = set(value) - _ACL_KEYS + if unknown: + raise DatasourceValidationError( + f"Unsupported ACL selectors in {path}: {', '.join(sorted(unknown))}." + ) + result: dict[str, list[str]] = {} + for key in sorted(value): + raw_values = value[key] + if not isinstance(raw_values, Sequence) or isinstance(raw_values, (str, bytes)): + raise DatasourceValidationError( + f"Visibility ACL selector {path}.{key} must be a list." + ) + if len(raw_values) > 250: + raise DatasourceValidationError( + f"Visibility ACL selector {path}.{key} is limited to 250 entries." + ) + normalized = sorted( + {str(item).strip() for item in raw_values if str(item).strip()} + ) + result[key] = normalized + return result + + +def _acl_allows(acl: Mapping[str, object], principal: PrincipalRef) -> bool: + facts = _principal_policy_facts(principal) + return any(set(values) & set(facts.get(key, ())) for key, values in acl.items()) + + +def _principal_policy_facts(principal: PrincipalRef) -> dict[str, tuple[str, ...]]: + return { + "account_ids": _optional_tuple(principal.account_id), + "membership_ids": _optional_tuple(principal.membership_id), + "identity_ids": _optional_tuple(principal.identity_id), + "group_ids": tuple(sorted(principal.group_ids)), + "role_ids": tuple(sorted(principal.role_ids)), + "service_account_ids": _optional_tuple(principal.service_account_id), + "auth_methods": (principal.auth_method,), + "function_assignment_ids": tuple(sorted(principal.function_assignment_ids)), + } + + +def _claim_values(principal: PrincipalRef, claim: str) -> tuple[str, ...]: + facts = _principal_policy_facts(principal) + aliases = { + "account_id": "account_ids", + "membership_id": "membership_ids", + "identity_id": "identity_ids", + "service_account_id": "service_account_ids", + } + return facts.get(aliases.get(claim, claim), ()) + + +def _relevant_policy_fact_keys( + policies: Sequence[Mapping[str, object]], +) -> set[str]: + keys: set[str] = set() + aliases = { + "account_id": "account_ids", + "membership_id": "membership_ids", + "identity_id": "identity_ids", + "service_account_id": "service_account_ids", + } + for policy in policies: + for acl_name in ("source_acl", "materialization_acl"): + acl = policy.get(acl_name) + if isinstance(acl, Mapping): + keys.update(str(key) for key in acl) + fields = policy.get("fields") + if isinstance(fields, Mapping): + for rule in fields.values(): + if isinstance(rule, Mapping) and isinstance(rule.get("allow"), Mapping): + keys.update(str(key) for key in rule["allow"]) + for row_filter in policy.get("row_filters", ()): + if isinstance(row_filter, Mapping): + claim = str(row_filter.get("claim") or "") + keys.add(aliases.get(claim, claim)) + return keys + + +def _row_filter_allows( + row: Mapping[str, object], + item: RowFilter, + principal: PrincipalRef, +) -> bool: + value = row.get(item.field) + if value is None: + return item.allow_null + claims = set(_claim_values(principal, item.claim)) + if not claims: + return False + if isinstance(value, Sequence) and not isinstance(value, (str, bytes)): + values = {str(candidate) for candidate in value} + return bool(values & claims) + return str(value) in claims + + +def _optional_tuple(value: object | None) -> tuple[str, ...]: + return (str(value),) if value is not None and str(value) else () + + +def _digest(value: object) -> str: + encoded = json.dumps(value, sort_keys=True, separators=(",", ":"), default=str) + return hashlib.sha256(encoded.encode("utf-8")).hexdigest() + + +__all__ = [ + "VisibilityPlan", + "apply_visibility", + "normalize_visibility_policy", + "visibility_diagnostics", + "visibility_plan", + "visible_schema", +] diff --git a/tests/test_interface_documentation_contract.py b/tests/test_interface_documentation_contract.py index b050977..9d2fd5e 100644 --- a/tests/test_interface_documentation_contract.py +++ b/tests/test_interface_documentation_contract.py @@ -27,11 +27,17 @@ class DatasourcesInterfaceDocumentationContractTests(unittest.TestCase): lifecycle = topics["datasources.lifecycle"] governance = topics["datasources.governance"] quality = topics["datasources.quality-gates"] + visibility = topics["datasources.visibility"] reference = topics["datasources.reference.fields-and-consequences"] self.assertIn("datasources.staging", lifecycle.metadata["help_contexts"]) self.assertIn("datasources.field.authority-mode", governance.metadata["help_contexts"]) self.assertIn("datasources.staging.validation", quality.metadata["help_contexts"]) + self.assertIn( + "datasources.field.visibility-policy", + visibility.metadata["help_contexts"], + ) + self.assertIn("fails closed", visibility.body) self.assertIn("policy version and hash", quality.body) self.assertTrue(quality.metadata["limitations"]) self.assertIn("datasources.action.promote", reference.metadata["help_contexts"]) diff --git a/tests/test_migrations.py b/tests/test_migrations.py index 703afc9..0fc2a39 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -34,7 +34,7 @@ class DatasourceMigrationTests(unittest.TestCase): try: with engine.connect() as connection: self.assertIn( - "b8d2f5a0c3e7", + "c9e3a6f1d4b8", set(MigrationContext.configure(connection).get_current_heads()), ) catalogue_columns = { @@ -50,6 +50,8 @@ class DatasourceMigrationTests(unittest.TestCase): "publication_state", "owner_ref", "quality_policy", + "access_policy_ref", + "visibility_policy", "dependency_refs", }.issubset(catalogue_columns) ) diff --git a/tests/test_visibility.py b/tests/test_visibility.py new file mode 100644 index 0000000..b39a7d7 --- /dev/null +++ b/tests/test_visibility.py @@ -0,0 +1,468 @@ +from __future__ import annotations + +import unittest +from unittest.mock import patch + +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from govoplan_core.auth import ApiPrincipal +from govoplan_core.core.access import PrincipalRef +from govoplan_core.core.change_sequence import ChangeSequenceEntry +from govoplan_core.core.datasources import ( + CAPABILITY_DATASOURCE_ORIGINS, + CAPABILITY_POLICY_DATASOURCE_VISIBILITY, + DatasourceAccessError, + DatasourceGovernance, + DatasourceNotFoundError, + DatasourceField, + DatasourceOrigin, + DatasourceOriginReadResult, + DatasourceReadRequest, + DatasourceVisibilityPolicyDecision, +) +from govoplan_core.db.base import Base, utcnow +from govoplan_datasources.backend.db.models import ( + DatasourceGovernanceReferenceRecord, + DatasourceMaterializationRecord, + DatasourcePayloadRecord, + DatasourceRecord, +) +from govoplan_datasources.backend.service import ( + CATALOGUE_READ_SCOPE, + SqlDatasourceProvider, +) + + +SCHEMA = [ + {"name": "id", "data_type": "integer", "nullable": False}, + {"name": "owner_id", "data_type": "string", "nullable": False}, + {"name": "secret", "data_type": "string", "nullable": False}, + {"name": "internal_note", "data_type": "string", "nullable": True}, +] +ROWS = [ + { + "id": 1, + "owner_id": "account-1", + "secret": "protected-one", + "internal_note": "hidden-one", + }, + { + "id": 2, + "owner_id": "account-2", + "secret": "protected-two", + "internal_note": "hidden-two", + }, +] + + +def principal( + *, + tenant_id: str = "tenant-1", + account_id: str = "account-1", + role_ids: tuple[str, ...] = ("reader",), + auth_method: str = "session", + service_account_id: str | None = None, +) -> ApiPrincipal: + return ApiPrincipal( + principal=PrincipalRef( + account_id=account_id, + membership_id=(None if auth_method == "service_account" else "member-1"), + tenant_id=tenant_id, + scopes=frozenset({CATALOGUE_READ_SCOPE}), + role_ids=frozenset(role_ids), + auth_method=auth_method, + service_account_id=service_account_id, + ), + account=object(), + user=object(), + ) + + +class _PolicyProvider: + def __init__(self, decision: DatasourceVisibilityPolicyDecision) -> None: + self.decision = decision + self.requests = [] + + def decide_datasource_visibility(self, _session, *, request): + self.requests.append(request) + return self.decision + + +class _Registry: + def __init__(self, policy_provider: _PolicyProvider) -> None: + self.policy_provider = policy_provider + + def has_capability(self, name: str) -> bool: + return name == CAPABILITY_POLICY_DATASOURCE_VISIBILITY + + def capability(self, name: str) -> object: + if not self.has_capability(name): + raise KeyError(name) + return self.policy_provider + + +class _OriginProvider: + origin = DatasourceOrigin( + ref="secret:origin", + source_name="live_cases", + name="Live cases", + kind="database", + shape="tabular", + supported_modes=("live",), + provider="test", + schema=tuple( + DatasourceField( + name=str(field["name"]), + data_type=str(field["data_type"]), + nullable=bool(field["nullable"]), + ) + for field in SCHEMA + ), + fingerprint="live-base-fingerprint", + row_count=2, + ) + + def list_origins(self, _session, _principal, *, query="", limit=100): + del query + return (self.origin,)[:limit] + + def get_origin(self, _session, _principal, *, origin_ref): + return self.origin if origin_ref == self.origin.ref else None + + def read_origin(self, _session, _principal, *, request): + rows = ROWS[request.offset : request.offset + request.limit] + return DatasourceOriginReadResult( + origin=self.origin, + rows=tuple(rows), + total_rows=2, + truncated=request.offset + len(rows) < 2, + elapsed_ms=1, + ) + + +class _LiveRegistry: + def __init__(self) -> None: + self.origin_provider = _OriginProvider() + + def has_capability(self, name: str) -> bool: + return name == CAPABILITY_DATASOURCE_ORIGINS + + def capability(self, name: str) -> object: + if not self.has_capability(name): + raise KeyError(name) + return self.origin_provider + + +class DatasourceVisibilityTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all( + self.engine, + tables=[ + DatasourceRecord.__table__, + DatasourceGovernanceReferenceRecord.__table__, + DatasourcePayloadRecord.__table__, + DatasourceMaterializationRecord.__table__, + ChangeSequenceEntry.__table__, + ], + ) + self.Session = sessionmaker(bind=self.engine) + self.session = self.Session() + + def tearDown(self) -> None: + self.session.close() + Base.metadata.drop_all( + self.engine, + tables=[ + DatasourceMaterializationRecord.__table__, + ChangeSequenceEntry.__table__, + DatasourcePayloadRecord.__table__, + DatasourceGovernanceReferenceRecord.__table__, + DatasourceRecord.__table__, + ], + ) + self.engine.dispose() + + def _datasource( + self, + *, + policy: dict[str, object] | None = None, + access_policy_ref: str | None = None, + frozen: bool = False, + snapshot_policy: dict[str, object] | None = None, + ) -> DatasourceRecord: + item = DatasourceRecord( + tenant_id="tenant-1", + source_name="governed_cases", + name="Governed cases", + kind="upload", + mode="static", + shape="tabular", + status="active", + provider="private.provider", + provider_ref="secret:origin", + schema_=SCHEMA, + fingerprint="base-fingerprint", + row_count=2, + byte_count=250, + provenance_={"secret_locator": "private://rows"}, + metadata_={"credential_ref": "credential:secret"}, + access_policy_ref=access_policy_ref, + visibility_policy=policy or {}, + ) + self.session.add(item) + self.session.flush() + snapshot = DatasourceGovernance( + publication_state="internal", + visibility_policy=snapshot_policy or policy or {}, + access_policy_ref=access_policy_ref, + ) + materialization = DatasourceMaterializationRecord( + tenant_id="tenant-1", + datasource_id=item.id, + revision=1, + state="published", + schema_=SCHEMA, + rows=ROWS, + fingerprint="materialization-fingerprint", + row_count=2, + byte_count=250, + frozen_at=utcnow() if frozen else None, + frozen_label="Evidence" if frozen else None, + provenance_={"protected": "source details"}, + metadata_={"protected": "materialization details"}, + governance_snapshot_=snapshot.to_dict(), + ) + self.session.add(materialization) + self.session.flush() + item.current_materialization_id = materialization.id + self.session.flush() + return item + + def test_role_acl_filters_discovery_and_denies_reads(self) -> None: + item = self._datasource(policy={"source_acl": {"role_ids": ["reader"]}}) + provider = SqlDatasourceProvider() + + self.assertEqual(1, len(provider.list_datasources(self.session, principal()))) + denied = principal(role_ids=("other",)) + self.assertEqual((), provider.list_datasources(self.session, denied)) + self.assertIsNone( + provider.get_datasource( + self.session, + denied, + datasource_ref=f"datasource:{item.id}", + ) + ) + with self.assertRaises(DatasourceAccessError): + provider.read_datasource( + self.session, + denied, + request=DatasourceReadRequest(datasource_ref=f"datasource:{item.id}"), + ) + other_tenant = principal(tenant_id="tenant-2") + self.assertEqual((), provider.list_datasources(self.session, other_tenant)) + with self.assertRaises(DatasourceNotFoundError): + provider.read_datasource( + self.session, + other_tenant, + request=DatasourceReadRequest(datasource_ref=f"datasource:{item.id}"), + ) + + def test_rows_and_fields_are_filtered_into_an_opaque_permitted_view(self) -> None: + policy = { + "source_acl": {"role_ids": ["reader"]}, + "fields": { + "secret": { + "classification": "restricted", + "action": "redact", + "allow": {"role_ids": ["privileged"]}, + }, + "internal_note": { + "classification": "confidential", + "action": "omit", + "allow": {"role_ids": ["privileged"]}, + }, + }, + "row_filters": [ + {"field": "owner_id", "claim": "account_id", "operator": "equals"} + ], + } + item = self._datasource(policy=policy) + provider = SqlDatasourceProvider() + + with patch( + "govoplan_datasources.backend.service.audit_event" + ) as audit_event: + result = provider.read_datasource( + self.session, + principal(), + request=DatasourceReadRequest( + datasource_ref=f"datasource:{item.id}" + ), + ) + + self.assertEqual( + ({"id": 1, "owner_id": "account-1", "secret": None},), + result.rows, + ) + self.assertEqual( + ["id", "owner_id", "secret"], + [field.name for field in result.datasource.schema], + ) + self.assertEqual("restricted", result.datasource.schema[-1].classification) + self.assertNotEqual( + "materialization-fingerprint", result.datasource.fingerprint + ) + self.assertIsNone(result.datasource.provider) + self.assertIsNone(result.datasource.provider_ref) + self.assertNotIn("credential_ref", result.datasource.metadata) + self.assertEqual(1, result.total_rows) + self.assertEqual( + ["datasource.visibility_applied"], + [diagnostic.code for diagnostic in result.diagnostics], + ) + audit_details = audit_event.call_args.kwargs["details"] + self.assertEqual(2, audit_details["visibility_summary"]["field_rule_count"]) + self.assertNotIn("protected-one", str(audit_details)) + + repeated = provider.read_datasource( + self.session, + principal(), + request=DatasourceReadRequest( + datasource_ref=f"datasource:{item.id}", + expected_fingerprint=result.datasource.fingerprint, + ), + ) + self.assertEqual(result.datasource.fingerprint, repeated.datasource.fingerprint) + + def test_materialization_acl_supports_service_principals(self) -> None: + item = self._datasource( + policy={ + "source_acl": {"auth_methods": ["session", "service_account"]}, + "materialization_acl": {"service_account_ids": ["service:reporting"]}, + } + ) + provider = SqlDatasourceProvider() + + with self.assertRaises(DatasourceAccessError): + provider.read_datasource( + self.session, + principal(), + request=DatasourceReadRequest(datasource_ref=f"datasource:{item.id}"), + ) + result = provider.read_datasource( + self.session, + principal( + account_id="service-account", + auth_method="service_account", + service_account_id="service:reporting", + ), + request=DatasourceReadRequest(datasource_ref=f"datasource:{item.id}"), + ) + self.assertEqual(2, result.total_rows) + + def test_frozen_state_keeps_its_restrictive_local_policy(self) -> None: + snapshot_policy = {"source_acl": {"role_ids": ["evidence-reader"]}} + item = self._datasource( + policy={}, + frozen=True, + snapshot_policy=snapshot_policy, + ) + provider = SqlDatasourceProvider() + + with self.assertRaises(DatasourceAccessError): + provider.read_datasource( + self.session, + principal(role_ids=("reader",)), + request=DatasourceReadRequest( + datasource_ref=f"datasource:{item.id}", + consistency="frozen", + ), + ) + result = provider.read_datasource( + self.session, + principal(role_ids=("evidence-reader",)), + request=DatasourceReadRequest( + datasource_ref=f"datasource:{item.id}", + consistency="frozen", + ), + ) + self.assertEqual(2, result.total_rows) + + def test_external_policy_tightens_local_behavior_and_missing_provider_denies( + self, + ) -> None: + item = self._datasource(access_policy_ref="case-workers") + with self.assertRaises(DatasourceAccessError): + SqlDatasourceProvider().read_datasource( + self.session, + principal(), + request=DatasourceReadRequest(datasource_ref=f"datasource:{item.id}"), + ) + + external = _PolicyProvider( + DatasourceVisibilityPolicyDecision( + allowed=True, + policies=({"source_acl": {"role_ids": ["case-worker"]}},), + decision_ref="policy-decision:1", + ) + ) + provider = SqlDatasourceProvider(registry=_Registry(external)) + result = provider.read_datasource( + self.session, + principal(role_ids=("case-worker",)), + request=DatasourceReadRequest(datasource_ref=f"datasource:{item.id}"), + ) + self.assertEqual(2, result.total_rows) + self.assertEqual("case-workers", external.requests[-1].policy_ref) + + def test_denied_audit_contains_no_protected_values(self) -> None: + item = self._datasource(policy={"source_acl": {"role_ids": ["authorized"]}}) + provider = SqlDatasourceProvider() + with patch("govoplan_datasources.backend.service.audit_event") as audit_event: + with self.assertRaises(DatasourceAccessError): + provider.read_datasource( + self.session, + principal(role_ids=("denied",)), + request=DatasourceReadRequest( + datasource_ref=f"datasource:{item.id}" + ), + ) + + details = audit_event.call_args.kwargs["details"] + self.assertEqual("denied", details["outcome"]) + serialized = str(details) + self.assertNotIn("protected-one", serialized) + self.assertNotIn("hidden-one", serialized) + self.assertNotIn("credential:secret", serialized) + + def test_live_rows_are_filtered_before_origin_data_is_returned(self) -> None: + item = self._datasource( + policy={ + "source_acl": {"role_ids": ["reader"]}, + "row_filters": [ + {"field": "owner_id", "claim": "account_id"} + ], + } + ) + item.mode = "live" + self.session.flush() + + result = SqlDatasourceProvider(registry=_LiveRegistry()).read_datasource( + self.session, + principal(), + request=DatasourceReadRequest( + datasource_ref=f"datasource:{item.id}", + consistency="live", + ), + ) + + self.assertEqual(1, result.total_rows) + self.assertEqual("account-1", result.rows[0]["owner_id"]) + self.assertNotEqual("live-base-fingerprint", result.datasource.fingerprint) + self.assertIsNone(result.datasource.provider_ref) + + +if __name__ == "__main__": + unittest.main() diff --git a/webui/package.json b/webui/package.json index 3481c0c..726792a 100644 --- a/webui/package.json +++ b/webui/package.json @@ -1,6 +1,6 @@ { "name": "@govoplan/datasources-webui", - "version": "0.1.19", + "version": "0.1.20", "private": true, "type": "module", "main": "src/index.ts", diff --git a/webui/src/api/datasources.ts b/webui/src/api/datasources.ts index 86895fd..d7d1e49 100644 --- a/webui/src/api/datasources.ts +++ b/webui/src/api/datasources.ts @@ -28,6 +28,8 @@ export type DatasourceGovernance = { classification: string; privacy_profile_ref?: string | null; retention_policy_ref?: string | null; + access_policy_ref?: string | null; + visibility_policy: Record; hold_refs: string[]; publication_state: string; transfer_agreement_ref?: string | null; @@ -43,6 +45,7 @@ export type DatasourceField = { name: string; data_type: string; nullable: boolean; + classification: string; }; export type Datasource = { diff --git a/webui/src/features/datasources/DatasourcesPage.tsx b/webui/src/features/datasources/DatasourcesPage.tsx index 6326248..392431e 100644 --- a/webui/src/features/datasources/DatasourcesPage.tsx +++ b/webui/src/features/datasources/DatasourcesPage.tsx @@ -73,6 +73,7 @@ import { import { DATASOURCE_FIELDS_DOCUMENTATION, DATASOURCE_GOVERNANCE_DOCUMENTATION, + DATASOURCE_VISIBILITY_DOCUMENTATION, DATASOURCES_DOCUMENTATION, DATASOURCES_I18N } from "./interfacePatterns"; @@ -898,6 +899,7 @@ function GovernanceDialog({ const [draft, setDraft] = useState(null); const [freshness, setFreshness] = useState("{}"); const [quality, setQuality] = useState("{}"); + const [visibility, setVisibility] = useState("{}"); const [baselineKey, setBaselineKey] = useState(""); const [busy, setBusy] = useState(false); const [error, setError] = useState(""); @@ -908,14 +910,16 @@ function GovernanceDialog({ const nextDraft = structuredClone(datasource.governance); const nextFreshness = JSON.stringify(datasource.governance.freshness_policy, null, 2); const nextQuality = JSON.stringify(datasource.governance.quality_policy, null, 2); + const nextVisibility = JSON.stringify(datasource.governance.visibility_policy ?? {}, null, 2); setDraft(nextDraft); setFreshness(nextFreshness); setQuality(nextQuality); - setBaselineKey(JSON.stringify({ draft: nextDraft, freshness: nextFreshness, quality: nextQuality })); + setVisibility(nextVisibility); + setBaselineKey(JSON.stringify({ draft: nextDraft, freshness: nextFreshness, quality: nextQuality, visibility: nextVisibility })); setError(""); }, [datasource, open]); - const dirty = Boolean(open && draft && JSON.stringify({ draft, freshness, quality }) !== baselineKey); + const dirty = Boolean(open && draft && JSON.stringify({ draft, freshness, quality, visibility }) !== baselineKey); const save = async (): Promise => { if (!datasource || !draft) return false; @@ -925,7 +929,8 @@ function GovernanceDialog({ const updated = await updateDatasourceGovernance(settings, datasource.ref, { ...draft, freshness_policy: parseObject(freshness, "Freshness policy"), - quality_policy: parseObject(quality, "Quality policy") + quality_policy: parseObject(quality, "Quality policy"), + visibility_policy: parseObject(visibility, "Visibility policy") }); await onSaved(updated); return true; @@ -1020,6 +1025,9 @@ function GovernanceDialog({ setValue("retention_policy_ref", event.target.value || null)} /> + + setValue("access_policy_ref", event.target.value || null)} placeholder="Optional Policy module target" /> + setValue("transfer_agreement_ref", event.target.value || null)} /> @@ -1040,6 +1048,10 @@ function GovernanceDialog({ setValue("known_limits", values)} /> + +