diff --git a/pyproject.toml b/pyproject.toml index a21cbbc..164854b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,7 +10,10 @@ readme = "README.md" requires-python = ">=3.12" license = "AGPL-3.0-or-later" authors = [{ name = "GovOPlaN" }] -dependencies = ["govoplan-core>=0.1.14"] +dependencies = [ + "defusedxml>=0.7,<1", + "govoplan-core>=0.1.14", +] [tool.setuptools.packages.find] where = ["src"] diff --git a/src/govoplan_connectors/backend/db/models.py b/src/govoplan_connectors/backend/db/models.py index fde304f..36b611d 100644 --- a/src/govoplan_connectors/backend/db/models.py +++ b/src/govoplan_connectors/backend/db/models.py @@ -4,7 +4,17 @@ import uuid from datetime import datetime from typing import Any -from sqlalchemy import DateTime, Index, Integer, JSON, String, Text, UniqueConstraint +from sqlalchemy import ( + DateTime, + ForeignKey, + Index, + Integer, + JSON, + LargeBinary, + String, + Text, + UniqueConstraint, +) from sqlalchemy.orm import Mapped, mapped_column from govoplan_core.db.base import Base, TimestampMixin @@ -41,4 +51,213 @@ class ConnectorTabularSource(Base, TimestampMixin): deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True) -__all__ = ["ConnectorTabularSource", "new_uuid"] +class ConnectorSanctionsAcquisitionRun(Base, TimestampMixin): + __tablename__ = "connector_sanctions_acquisition_runs" + __table_args__ = ( + Index( + "ix_connector_sanctions_run_health", + "tenant_id", + "provider_id", + "status", + "started_at", + ), + ) + + id: Mapped[str] = mapped_column( + String(36), + primary_key=True, + default=new_uuid, + ) + tenant_id: Mapped[str] = mapped_column( + String(36), + nullable=False, + index=True, + ) + provider_id: Mapped[str] = mapped_column( + String(100), + nullable=False, + index=True, + ) + source_id: Mapped[str] = mapped_column( + String(200), + nullable=False, + index=True, + ) + status: Mapped[str] = mapped_column( + String(40), + default="running", + nullable=False, + index=True, + ) + attempt_count: Mapped[int] = mapped_column( + Integer, + default=0, + nullable=False, + ) + request_evidence: Mapped[dict[str, Any]] = mapped_column( + JSON, + default=dict, + nullable=False, + ) + response_evidence: Mapped[dict[str, Any]] = mapped_column( + JSON, + default=dict, + nullable=False, + ) + started_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + nullable=False, + index=True, + ) + finished_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), + nullable=True, + ) + snapshot_id: Mapped[str | None] = mapped_column( + String(36), + nullable=True, + index=True, + ) + error: Mapped[str | None] = mapped_column( + Text, + nullable=True, + ) + created_by: Mapped[str | None] = mapped_column( + String(255), + nullable=True, + index=True, + ) + + +class ConnectorSanctionsSnapshot(Base, TimestampMixin): + __tablename__ = "connector_sanctions_snapshots" + __table_args__ = ( + UniqueConstraint( + "connector_run_id", + name="uq_connector_sanctions_snapshot_run", + ), + Index( + "ix_connector_sanctions_snapshot_source", + "tenant_id", + "provider_id", + "acquired_at", + ), + Index( + "ix_connector_sanctions_snapshot_version", + "provider_id", + "source_id", + "source_version", + ), + ) + + id: Mapped[str] = mapped_column( + String(36), + primary_key=True, + default=new_uuid, + ) + tenant_id: Mapped[str] = mapped_column( + String(36), + nullable=False, + index=True, + ) + provider_id: Mapped[str] = mapped_column( + String(100), + nullable=False, + index=True, + ) + publisher: Mapped[str] = mapped_column( + String(300), + nullable=False, + ) + jurisdiction: Mapped[str] = mapped_column( + String(100), + nullable=False, + index=True, + ) + list_type: Mapped[str] = mapped_column( + String(100), + nullable=False, + index=True, + ) + source_id: Mapped[str] = mapped_column( + String(200), + nullable=False, + index=True, + ) + source_version: Mapped[str] = mapped_column( + String(255), + nullable=False, + index=True, + ) + publication_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), + nullable=True, + ) + effective_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), + nullable=True, + ) + acquired_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + nullable=False, + index=True, + ) + source_url: Mapped[str | None] = mapped_column( + String(1500), + nullable=True, + ) + content_type: Mapped[str] = mapped_column( + String(200), + nullable=False, + ) + byte_count: Mapped[int] = mapped_column( + Integer, + nullable=False, + ) + sha256: Mapped[str] = mapped_column( + String(64), + nullable=False, + index=True, + ) + signature_evidence: Mapped[dict[str, Any]] = mapped_column( + JSON, + default=dict, + nullable=False, + ) + parser_version: Mapped[str] = mapped_column( + String(100), + nullable=False, + ) + licence_notes: Mapped[str | None] = mapped_column( + Text, + nullable=True, + ) + trust_notes: Mapped[str | None] = mapped_column( + Text, + nullable=True, + ) + connector_run_id: Mapped[str] = mapped_column( + ForeignKey( + "connector_sanctions_acquisition_runs.id", + ondelete="RESTRICT", + ), + nullable=False, + index=True, + ) + transport_evidence: Mapped[dict[str, Any]] = mapped_column( + JSON, + default=dict, + nullable=False, + ) + raw_content: Mapped[bytes] = mapped_column( + LargeBinary, + nullable=False, + ) + + +__all__ = [ + "ConnectorSanctionsAcquisitionRun", + "ConnectorSanctionsSnapshot", + "ConnectorTabularSource", + "new_uuid", +] diff --git a/src/govoplan_connectors/backend/manifest.py b/src/govoplan_connectors/backend/manifest.py index 591568d..d811ed8 100644 --- a/src/govoplan_connectors/backend/manifest.py +++ b/src/govoplan_connectors/backend/manifest.py @@ -23,8 +23,20 @@ from govoplan_core.core.tabular_sources import ( CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER, CAPABILITY_CONNECTORS_TABULAR_SOURCES, ) +from govoplan_core.core.sanctions import ( + CAPABILITY_CONNECTORS_SANCTIONS_SNAPSHOTS, +) from govoplan_core.db.base import Base -from govoplan_connectors.backend.db.models import ConnectorTabularSource +from govoplan_connectors.backend.db.models import ( + ConnectorSanctionsAcquisitionRun, + ConnectorSanctionsSnapshot, + ConnectorTabularSource, +) +from govoplan_connectors.backend.sanctions_sources import ( + SANCTIONS_READ_SCOPE, + SANCTIONS_REFRESH_SCOPE, + SqlSanctionsSnapshotProvider, +) from govoplan_connectors.backend.tabular_sources import ( ADMIN_SCOPE, READ_SCOPE, @@ -40,6 +52,7 @@ MODULE_ID = "connectors" MODULE_VERSION = "0.1.14" TABULAR_SOURCE_INTERFACE_VERSION = "0.1.0" DATASOURCE_ORIGIN_INTERFACE_VERSION = "0.1.0" +SANCTIONS_SNAPSHOT_INTERFACE_VERSION = "1.0.0" def _permission(scope: str, label: str, description: str) -> PermissionDefinition: @@ -72,6 +85,16 @@ PERMISSIONS = ( "Administer connector sources", "Manage every tenant connector source and future source policies.", ), + _permission( + SANCTIONS_READ_SCOPE, + "View sanctions source evidence", + "Inspect immutable sanctions snapshots and acquisition health.", + ), + _permission( + SANCTIONS_REFRESH_SCOPE, + "Refresh sanctions sources", + "Acquire a new immutable sanctions source snapshot.", + ), ) ROLE_TEMPLATES = ( @@ -79,13 +102,18 @@ ROLE_TEMPLATES = ( slug="connector_source_manager", name="Connector source manager", description="Discover, import, preview, and retire tabular sources.", - permissions=(READ_SCOPE, WRITE_SCOPE), + permissions=( + READ_SCOPE, + WRITE_SCOPE, + SANCTIONS_READ_SCOPE, + SANCTIONS_REFRESH_SCOPE, + ), ), RoleTemplate( slug="connector_source_reader", name="Connector source reader", description="Discover and preview tabular connector sources.", - permissions=(READ_SCOPE,), + permissions=(READ_SCOPE, SANCTIONS_READ_SCOPE), ), ) @@ -104,6 +132,12 @@ def _datasource_origin_provider(_context) -> ConnectorDatasourceOriginProvider: return ConnectorDatasourceOriginProvider() +def _sanctions_snapshot_provider( + _context, +) -> SqlSanctionsSnapshotProvider: + return SqlSanctionsSnapshotProvider() + + def _tenant_summary(session, tenant_id: str) -> dict[str, int]: return { "connector_tabular_sources": ( @@ -113,7 +147,22 @@ def _tenant_summary(session, tenant_id: str) -> dict[str, int]: ConnectorTabularSource.deleted_at.is_(None), ) .count() - ) + ), + "connector_sanctions_snapshots": ( + session.query(ConnectorSanctionsSnapshot) + .filter( + ConnectorSanctionsSnapshot.tenant_id == tenant_id + ) + .count() + ), + "connector_sanctions_runs": ( + session.query(ConnectorSanctionsAcquisitionRun) + .filter( + ConnectorSanctionsAcquisitionRun.tenant_id + == tenant_id + ) + .count() + ), } @@ -121,7 +170,13 @@ manifest = ModuleManifest( id=MODULE_ID, name="Connectors", version=MODULE_VERSION, - optional_dependencies=("access", "audit", "files", "policy"), + optional_dependencies=( + "access", + "audit", + "files", + "policy", + "risk_compliance", + ), required_capabilities=( CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR, @@ -139,6 +194,10 @@ manifest = ModuleManifest( name="connectors.datasource_origins", version=DATASOURCE_ORIGIN_INTERFACE_VERSION, ), + ModuleInterfaceProvider( + name="connectors.sanctions_snapshots", + version=SANCTIONS_SNAPSHOT_INTERFACE_VERSION, + ), ), permissions=PERMISSIONS, role_templates=ROLE_TEMPLATES, @@ -147,6 +206,9 @@ manifest = ModuleManifest( CAPABILITY_CONNECTORS_TABULAR_SOURCES: _provider, CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER: _provider, CAPABILITY_DATASOURCE_ORIGINS: _datasource_origin_provider, + CAPABILITY_CONNECTORS_SANCTIONS_SNAPSHOTS: ( + _sanctions_snapshot_provider + ), }, tenant_summary_providers=(_tenant_summary,), migration_spec=MigrationSpec( @@ -155,6 +217,8 @@ manifest = ModuleManifest( script_location=str(Path(__file__).with_name("migrations") / "versions"), retirement_supported=True, retirement_provider=drop_table_retirement_provider( + ConnectorSanctionsSnapshot, + ConnectorSanctionsAcquisitionRun, ConnectorTabularSource, label="Connectors", ), @@ -165,6 +229,8 @@ manifest = ModuleManifest( ), uninstall_guard_providers=( persistent_table_uninstall_guard( + ConnectorSanctionsSnapshot, + ConnectorSanctionsAcquisitionRun, ConnectorTabularSource, label="Connectors", ), @@ -188,6 +254,27 @@ manifest = ModuleManifest( related_modules=("dataflow", "files", "reporting", "risk_compliance"), order=40, ), + DocumentationTopic( + id="connectors.sanctions-snapshots", + title="Sanctions source snapshots", + summary=( + "Acquire immutable, checksum-verifiable sanctions list " + "evidence without transmitting screening subjects." + ), + body=( + "Connectors provides a deterministic synthetic fixture and " + "the official United Nations Security Council consolidated " + "XML source. Each fetch records conditional transport " + "evidence, bounded retries, health state, source metadata, " + "raw evidence, and a SHA-256 checksum. Risk Compliance owns " + "normalization, matching, legal review, and dispositions." + ), + layer="available", + documentation_types=("admin", "user"), + audience=("operator", "module_admin", "compliance_reviewer"), + related_modules=("risk_compliance", "dataflow"), + order=41, + ), ), ) @@ -200,6 +287,7 @@ __all__ = [ "MODULE_ID", "MODULE_VERSION", "DATASOURCE_ORIGIN_INTERFACE_VERSION", + "SANCTIONS_SNAPSHOT_INTERFACE_VERSION", "TABULAR_SOURCE_INTERFACE_VERSION", "get_manifest", "manifest", diff --git a/src/govoplan_connectors/backend/migrations/versions/f7c8d9e0a1b2_v0114_sanctions_snapshots.py b/src/govoplan_connectors/backend/migrations/versions/f7c8d9e0a1b2_v0114_sanctions_snapshots.py new file mode 100644 index 0000000..a9a2b97 --- /dev/null +++ b/src/govoplan_connectors/backend/migrations/versions/f7c8d9e0a1b2_v0114_sanctions_snapshots.py @@ -0,0 +1,192 @@ +"""Add immutable sanctions source snapshots. + +Revision ID: f7c8d9e0a1b2 +Revises: e6b7c8d9f0a1 +Create Date: 2026-07-29 +""" + +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "f7c8d9e0a1b2" +down_revision = "e6b7c8d9f0a1" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "connector_sanctions_acquisition_runs", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("tenant_id", sa.String(length=36), nullable=False), + sa.Column("provider_id", sa.String(length=100), nullable=False), + sa.Column("source_id", sa.String(length=200), nullable=False), + sa.Column("status", sa.String(length=40), nullable=False), + sa.Column("attempt_count", sa.Integer(), nullable=False), + sa.Column("request_evidence", sa.JSON(), nullable=False), + sa.Column("response_evidence", sa.JSON(), nullable=False), + sa.Column( + "started_at", + sa.DateTime(timezone=True), + nullable=False, + ), + sa.Column( + "finished_at", + sa.DateTime(timezone=True), + nullable=True, + ), + sa.Column("snapshot_id", sa.String(length=36), nullable=True), + sa.Column("error", sa.Text(), nullable=True), + sa.Column("created_by", sa.String(length=255), nullable=True), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + nullable=False, + ), + sa.Column( + "updated_at", + sa.DateTime(timezone=True), + nullable=False, + ), + sa.PrimaryKeyConstraint( + "id", + name=op.f("pk_connector_sanctions_acquisition_runs"), + ), + ) + for column in ( + "tenant_id", + "provider_id", + "source_id", + "status", + "started_at", + "snapshot_id", + "created_by", + ): + op.create_index( + op.f( + "ix_connector_sanctions_acquisition_runs_" + f"{column}" + ), + "connector_sanctions_acquisition_runs", + [column], + ) + op.create_index( + "ix_connector_sanctions_run_health", + "connector_sanctions_acquisition_runs", + ["tenant_id", "provider_id", "status", "started_at"], + ) + + op.create_table( + "connector_sanctions_snapshots", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("tenant_id", sa.String(length=36), nullable=False), + sa.Column("provider_id", sa.String(length=100), nullable=False), + sa.Column("publisher", sa.String(length=300), nullable=False), + sa.Column("jurisdiction", sa.String(length=100), nullable=False), + sa.Column("list_type", sa.String(length=100), nullable=False), + sa.Column("source_id", sa.String(length=200), nullable=False), + sa.Column( + "source_version", + sa.String(length=255), + nullable=False, + ), + sa.Column( + "publication_at", + sa.DateTime(timezone=True), + nullable=True, + ), + sa.Column( + "effective_at", + sa.DateTime(timezone=True), + nullable=True, + ), + sa.Column( + "acquired_at", + sa.DateTime(timezone=True), + nullable=False, + ), + sa.Column("source_url", sa.String(length=1500), nullable=True), + sa.Column( + "content_type", + sa.String(length=200), + nullable=False, + ), + sa.Column("byte_count", sa.Integer(), nullable=False), + sa.Column("sha256", sa.String(length=64), nullable=False), + sa.Column("signature_evidence", sa.JSON(), nullable=False), + sa.Column( + "parser_version", + sa.String(length=100), + nullable=False, + ), + sa.Column("licence_notes", sa.Text(), nullable=True), + sa.Column("trust_notes", sa.Text(), nullable=True), + sa.Column( + "connector_run_id", + sa.String(length=36), + nullable=False, + ), + sa.Column("transport_evidence", sa.JSON(), nullable=False), + sa.Column("raw_content", sa.LargeBinary(), nullable=False), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + nullable=False, + ), + sa.Column( + "updated_at", + sa.DateTime(timezone=True), + nullable=False, + ), + sa.ForeignKeyConstraint( + ["connector_run_id"], + ["connector_sanctions_acquisition_runs.id"], + name=op.f( + "fk_connector_sanctions_snapshots_connector_run_id_" + "connector_sanctions_acquisition_runs" + ), + ondelete="RESTRICT", + ), + sa.PrimaryKeyConstraint( + "id", + name=op.f("pk_connector_sanctions_snapshots"), + ), + sa.UniqueConstraint( + "connector_run_id", + name="uq_connector_sanctions_snapshot_run", + ), + ) + for column in ( + "tenant_id", + "provider_id", + "jurisdiction", + "list_type", + "source_id", + "source_version", + "acquired_at", + "sha256", + "connector_run_id", + ): + op.create_index( + op.f(f"ix_connector_sanctions_snapshots_{column}"), + "connector_sanctions_snapshots", + [column], + ) + op.create_index( + "ix_connector_sanctions_snapshot_source", + "connector_sanctions_snapshots", + ["tenant_id", "provider_id", "acquired_at"], + ) + op.create_index( + "ix_connector_sanctions_snapshot_version", + "connector_sanctions_snapshots", + ["provider_id", "source_id", "source_version"], + ) + + +def downgrade() -> None: + op.drop_table("connector_sanctions_snapshots") + op.drop_table("connector_sanctions_acquisition_runs") diff --git a/src/govoplan_connectors/backend/router.py b/src/govoplan_connectors/backend/router.py index 604c6d9..a16259c 100644 --- a/src/govoplan_connectors/backend/router.py +++ b/src/govoplan_connectors/backend/router.py @@ -13,8 +13,16 @@ from govoplan_core.core.tabular_sources import ( TabularSourceError, TabularSourceNotFoundError, ) +from govoplan_core.core.sanctions import SanctionsSnapshotReference from govoplan_core.db.session import get_session from govoplan_connectors.backend.schemas import ( + SanctionsAcquisitionRunListResponse, + SanctionsAcquisitionRunResponse, + SanctionsRefreshResponse, + SanctionsSnapshotListResponse, + SanctionsSnapshotResponse, + SanctionsSourceListResponse, + SanctionsSourceResponse, SnapshotCreateRequest, TabularColumnResponse, TabularSourceDeleteResponse, @@ -22,6 +30,14 @@ from govoplan_connectors.backend.schemas import ( TabularSourcePreviewResponse, TabularSourceResponse, ) +from govoplan_connectors.backend.sanctions_sources import ( + SANCTIONS_READ_SCOPE, + SANCTIONS_REFRESH_SCOPE, + SanctionsSourceAccessError, + SanctionsSourceError, + SanctionsSourceNotFoundError, + SqlSanctionsSnapshotProvider, +) from govoplan_connectors.backend.tabular_sources import ( ADMIN_SCOPE, READ_SCOPE, @@ -33,6 +49,7 @@ from govoplan_connectors.backend.tabular_sources import ( router = APIRouter(prefix="/connectors", tags=["connectors"]) provider = SqlTabularSourceProvider() +sanctions_provider = SqlSanctionsSnapshotProvider() def _require_any_scope(principal: ApiPrincipal, *scopes: str) -> None: @@ -52,6 +69,25 @@ def _http_error(exc: TabularSourceError) -> HTTPException: return HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) +def _sanctions_http_error( + exc: SanctionsSourceError, +) -> HTTPException: + if isinstance(exc, SanctionsSourceNotFoundError): + return HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=str(exc), + ) + if isinstance(exc, SanctionsSourceAccessError): + return HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=str(exc), + ) + return HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, + detail=str(exc), + ) + + @router.get("/tabular-sources", response_model=TabularSourceListResponse) def api_list_tabular_sources( query: str = Query(default="", max_length=200), @@ -183,6 +219,147 @@ def api_delete_tabular_source( return TabularSourceDeleteResponse(deleted=True, source_ref=source_ref) +@router.get( + "/sanctions/sources", + response_model=SanctionsSourceListResponse, +) +def api_list_sanctions_sources( + principal: ApiPrincipal = Depends(get_api_principal), +) -> SanctionsSourceListResponse: + _require_any_scope( + principal, + SANCTIONS_READ_SCOPE, + SANCTIONS_REFRESH_SCOPE, + ADMIN_SCOPE, + ) + return SanctionsSourceListResponse( + sources=[ + SanctionsSourceResponse.model_validate( + source, + from_attributes=True, + ) + for source in sanctions_provider.available_sources() + ] + ) + + +@router.post( + "/sanctions/sources/{provider_id}/refresh", + response_model=SanctionsRefreshResponse, +) +def api_refresh_sanctions_source( + provider_id: str, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> SanctionsRefreshResponse: + _require_any_scope( + principal, + SANCTIONS_REFRESH_SCOPE, + ADMIN_SCOPE, + ) + try: + result = sanctions_provider.refresh_source( + session, + principal, + provider_id=provider_id, + ) + except SanctionsSourceError as exc: + raise _sanctions_http_error(exc) from exc + audit_event( + session, + tenant_id=principal.tenant_id, + user_id=getattr(principal.user, "id", None), + api_key_id=principal.api_key_id, + action="connectors.sanctions_source.refreshed", + object_type="connector_sanctions_acquisition_run", + object_id=result.run_id, + details={ + "provider_id": provider_id, + "status": result.status, + "snapshot_ref": ( + result.snapshot.ref + if result.snapshot is not None + else None + ), + }, + ) + session.commit() + return SanctionsRefreshResponse( + run_id=result.run_id, + provider_id=result.provider_id, + status=result.status, + snapshot=( + _sanctions_snapshot_response(result.snapshot) + if result.snapshot is not None + else None + ), + error=result.error, + ) + + +@router.get( + "/sanctions/snapshots", + response_model=SanctionsSnapshotListResponse, +) +def api_list_sanctions_snapshots( + limit: int = Query(default=100, ge=1, le=500), + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> SanctionsSnapshotListResponse: + _require_any_scope( + principal, + SANCTIONS_READ_SCOPE, + ADMIN_SCOPE, + ) + try: + snapshots = sanctions_provider.list_snapshots( + session, + principal, + limit=limit, + ) + except SanctionsSourceError as exc: + raise _sanctions_http_error(exc) from exc + return SanctionsSnapshotListResponse( + snapshots=[ + _sanctions_snapshot_response(item) + for item in snapshots + ] + ) + + +@router.get( + "/sanctions/runs", + response_model=SanctionsAcquisitionRunListResponse, +) +def api_list_sanctions_runs( + limit: int = Query(default=100, ge=1, le=500), + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> SanctionsAcquisitionRunListResponse: + _require_any_scope( + principal, + SANCTIONS_READ_SCOPE, + ADMIN_SCOPE, + ) + try: + runs = sanctions_provider.list_runs( + session, + principal, + limit=limit, + ) + except SanctionsSourceError as exc: + raise _sanctions_http_error(exc) from exc + return SanctionsAcquisitionRunListResponse( + runs=[ + SanctionsAcquisitionRunResponse.model_validate( + item, + from_attributes=True, + ) + for item in runs + ] + ) + + def _source_response(source: TabularSource) -> TabularSourceResponse: return TabularSourceResponse( ref=source.ref, @@ -208,4 +385,37 @@ def _source_response(source: TabularSource) -> TabularSourceResponse: ) +def _sanctions_snapshot_response( + snapshot: SanctionsSnapshotReference, +) -> SanctionsSnapshotResponse: + return SanctionsSnapshotResponse.model_validate( + { + "ref": snapshot.ref, + "provider_id": snapshot.provider_id, + "publisher": snapshot.publisher, + "jurisdiction": snapshot.jurisdiction, + "list_type": snapshot.list_type, + "source_id": snapshot.source_id, + "source_version": snapshot.source_version, + "publication_at": snapshot.publication_at, + "effective_at": snapshot.effective_at, + "acquired_at": snapshot.acquired_at, + "content_type": snapshot.content_type, + "byte_count": snapshot.byte_count, + "sha256": snapshot.sha256, + "parser_version": snapshot.parser_version, + "raw_evidence_ref": snapshot.raw_evidence_ref, + "connector_run_id": snapshot.connector_run_id, + "signature_evidence": dict( + snapshot.signature_evidence + ), + "licence_notes": snapshot.licence_notes, + "trust_notes": snapshot.trust_notes, + "transport_evidence": dict( + snapshot.transport_evidence + ), + } + ) + + __all__ = ["router"] diff --git a/src/govoplan_connectors/backend/sanctions_sources.py b/src/govoplan_connectors/backend/sanctions_sources.py new file mode 100644 index 0000000..ab81001 --- /dev/null +++ b/src/govoplan_connectors/backend/sanctions_sources.py @@ -0,0 +1,956 @@ +from __future__ import annotations + +import base64 +from collections.abc import Callable, Mapping, Sequence +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from email.utils import parsedate_to_datetime +import hashlib +import ssl +import time +from typing import Protocol +from urllib.error import HTTPError, URLError +from urllib.parse import urlsplit, urlunsplit +from urllib.request import ( + HTTPRedirectHandler, + HTTPSHandler, + Request, + build_opener, +) + +from defusedxml import ElementTree +from sqlalchemy import select +from sqlalchemy.orm import Session + +from govoplan_connectors.backend.db.models import ( + ConnectorSanctionsAcquisitionRun, + ConnectorSanctionsSnapshot, +) +from govoplan_core.auth import ApiPrincipal, has_scope +from govoplan_core.core.sanctions import ( + SanctionsSnapshotPayload, + SanctionsSnapshotReference, +) +from govoplan_core.db.base import utcnow + + +SANCTIONS_READ_SCOPE = "connectors:sanctions:read" +SANCTIONS_REFRESH_SCOPE = "connectors:sanctions:refresh" +CONNECTOR_ADMIN_SCOPE = "connectors:source:admin" +MAX_SANCTIONS_BYTES = 20_000_000 +MAX_FETCH_ATTEMPTS = 3 +STALE_AFTER = timedelta(hours=48) +UNSC_PROVIDER_ID = "un.security_council" +SYNTHETIC_PROVIDER_ID = "synthetic.un_fixture" +UNSC_SOURCE_URL = ( + "https://scsanctions.un.org/resources/xml/en/consolidated.xml" +) +ALLOWED_UNSC_HOSTS = frozenset( + { + "scsanctions.un.org", + "unsolprodfiles.blob.core.windows.net", + } +) +ALLOWED_XML_TYPES = frozenset( + { + "application/octet-stream", + "application/xml", + "text/xml", + } +) +SYNTHETIC_UN_XML = b""" + + + + fixture-person-1 + 1 + ALEX + EXAMPLE + Fixture regime + FXi.001 + 2026-01-15 + + Good + ALEXANDER EXAMPLE + + + 1980-05-12 + + + Passport + P-FIXTURE-001 + + + Example City + Exampleland + + + + + + fixture-entity-1 + 1 + EXAMPLE TRADING LTD + Fixture regime + FXe.001 + 2026-02-20 + + Good + EXAMPLE TRADING + + + Example Port + Exampleland + + + + +""" + + +class SanctionsSourceError(ValueError): + pass + + +class SanctionsSourceAccessError(SanctionsSourceError): + pass + + +class SanctionsSourceNotFoundError(SanctionsSourceError): + pass + + +class SanctionsSourceMalformedError(SanctionsSourceError): + pass + + +class SanctionsSourceUnexpectedChangeError(SanctionsSourceError): + pass + + +@dataclass(frozen=True, slots=True) +class SanctionsSourceDefinition: + provider_id: str + publisher: str + jurisdiction: str + list_type: str + source_id: str + source_url: str | None + parser_version: str + licence_notes: str + trust_notes: str + allowed_hosts: frozenset[str] = frozenset() + + +@dataclass(frozen=True, slots=True) +class TransportResponse: + status: int + final_url: str + headers: Mapping[str, str] + content: bytes + attempts: int + + +class SanctionsTransport(Protocol): + def fetch( + self, + definition: SanctionsSourceDefinition, + *, + headers: Mapping[str, str], + ) -> TransportResponse: + ... + + +@dataclass(frozen=True, slots=True) +class SanctionsAcquisitionResult: + run_id: str + provider_id: str + status: str + snapshot: SanctionsSnapshotReference | None + error: str | None = None + + +SOURCE_DEFINITIONS = { + SYNTHETIC_PROVIDER_ID: SanctionsSourceDefinition( + provider_id=SYNTHETIC_PROVIDER_ID, + publisher="GovOPlaN deterministic fixture", + jurisdiction="TEST", + list_type="synthetic_consolidated", + source_id="govoplan-un-shaped-fixture", + source_url=None, + parser_version="unsc-xml-v1", + licence_notes=( + "Synthetic fixture authored for GovOPlaN tests; not a legal list." + ), + trust_notes=( + "Deterministic non-production evidence. Never use for a legal " + "screening decision." + ), + ), + UNSC_PROVIDER_ID: SanctionsSourceDefinition( + provider_id=UNSC_PROVIDER_ID, + publisher="United Nations Security Council", + jurisdiction="UN", + list_type="consolidated_sanctions", + source_id="unsc-consolidated-list-by-reference-number", + source_url=UNSC_SOURCE_URL, + parser_version="unsc-xml-v1", + licence_notes=( + "Publicly provided by the United Nations Security Council. " + "Operators must confirm applicable reuse and retention policy." + ), + trust_notes=( + "Authoritative publisher transport. The consolidated list " + "facilitates implementation; measures remain regime-specific." + ), + allowed_hosts=ALLOWED_UNSC_HOSTS, + ), +} + + +class UrllibSanctionsTransport: + def __init__( + self, + *, + sleeper: Callable[[float], None] = time.sleep, + ) -> None: + self._sleeper = sleeper + + def fetch( + self, + definition: SanctionsSourceDefinition, + *, + headers: Mapping[str, str], + ) -> TransportResponse: + if definition.source_url is None: + return TransportResponse( + status=200, + final_url="fixture://govoplan/un-shaped", + headers={ + "content-type": "application/xml", + "etag": ( + '"' + + hashlib.sha256( + SYNTHETIC_UN_XML + ).hexdigest() + + '"' + ), + }, + content=SYNTHETIC_UN_XML, + attempts=1, + ) + opener = build_opener( + _AllowlistedRedirectHandler( + definition.allowed_hosts + ), + HTTPSHandler(context=ssl.create_default_context()), + ) + last_error: Exception | None = None + for attempt in range(1, MAX_FETCH_ATTEMPTS + 1): + request = Request( + definition.source_url, + headers={ + "Accept": "application/xml,text/xml;q=0.9", + "User-Agent": "GovOPlaN-sanctions-acquisition/1", + **dict(headers), + }, + method="GET", + ) + try: + with opener.open( + request, + timeout=30, + ) as response: + status = int(response.status) + return TransportResponse( + status=status, + final_url=_safe_url(response.geturl()), + headers=_safe_headers(response.headers), + content=_bounded_read(response), + attempts=attempt, + ) + except HTTPError as exc: + if exc.code == 304: + return TransportResponse( + status=304, + final_url=_safe_url(exc.geturl()), + headers=_safe_headers(exc.headers), + content=b"", + attempts=attempt, + ) + last_error = exc + if exc.code not in { + 408, + 425, + 429, + 500, + 502, + 503, + 504, + }: + break + except (TimeoutError, URLError) as exc: + last_error = exc + if attempt < MAX_FETCH_ATTEMPTS: + self._sleeper(float(2 ** (attempt - 1))) + raise SanctionsSourceError( + "Authoritative sanctions source is unavailable after bounded " + f"retries: {_bounded_error(last_error)}" + ) + + +class SqlSanctionsSnapshotProvider: + def __init__( + self, + transport: SanctionsTransport | None = None, + ) -> None: + self._transport = transport or UrllibSanctionsTransport() + + def available_sources( + self, + ) -> tuple[SanctionsSourceDefinition, ...]: + return tuple( + SOURCE_DEFINITIONS[key] + for key in sorted(SOURCE_DEFINITIONS) + ) + + def refresh_source( + self, + session: object, + principal: object, + *, + provider_id: str, + ) -> SanctionsAcquisitionResult: + db, api_principal = _context( + session, + principal, + SANCTIONS_REFRESH_SCOPE, + ) + definition = SOURCE_DEFINITIONS.get(provider_id) + if definition is None: + raise SanctionsSourceNotFoundError( + "Sanctions source provider is not available." + ) + now = utcnow() + latest = db.scalar( + select(ConnectorSanctionsSnapshot) + .where( + ConnectorSanctionsSnapshot.tenant_id + == api_principal.tenant_id, + ConnectorSanctionsSnapshot.provider_id + == provider_id, + ) + .order_by( + ConnectorSanctionsSnapshot.acquired_at.desc(), + ConnectorSanctionsSnapshot.id.desc(), + ) + .limit(1) + ) + conditional_headers = _conditional_headers(latest) + run = ConnectorSanctionsAcquisitionRun( + tenant_id=api_principal.tenant_id, + provider_id=provider_id, + source_id=definition.source_id, + status="running", + attempt_count=0, + request_evidence={ + "source_url": ( + _safe_url(definition.source_url) + if definition.source_url + else "fixture://govoplan/un-shaped" + ), + "conditional_headers": sorted( + conditional_headers + ), + "subject_data_transmitted": False, + }, + response_evidence={}, + started_at=now, + created_by=_actor_id(api_principal), + ) + db.add(run) + db.flush() + try: + response = self._transport.fetch( + definition, + headers=conditional_headers, + ) + run.attempt_count = response.attempts + run.response_evidence = { + "status": response.status, + "final_url": response.final_url, + "headers": dict(response.headers), + } + if response.status == 304: + if latest is None: + raise SanctionsSourceUnexpectedChangeError( + "Source returned not-modified without prior evidence." + ) + run.status = "not_modified" + run.snapshot_id = latest.id + run.finished_at = utcnow() + db.flush() + return SanctionsAcquisitionResult( + run_id=run.id, + provider_id=provider_id, + status=run.status, + snapshot=_snapshot_dto(latest), + ) + if response.status != 200: + raise SanctionsSourceError( + f"Unexpected source status: {response.status}" + ) + metadata = _validate_response( + definition, + response, + ) + acquired_at = utcnow() + checksum = hashlib.sha256(response.content).hexdigest() + signature_evidence = _checksum_evidence( + response.headers, + response.content, + ) + snapshot = ConnectorSanctionsSnapshot( + tenant_id=api_principal.tenant_id, + provider_id=provider_id, + publisher=definition.publisher, + jurisdiction=definition.jurisdiction, + list_type=definition.list_type, + source_id=definition.source_id, + source_version=_source_version( + metadata, + response.headers, + checksum, + ), + publication_at=metadata.get("publication_at"), + effective_at=metadata.get("effective_at"), + acquired_at=acquired_at, + source_url=response.final_url, + content_type=_content_type(response.headers), + byte_count=len(response.content), + sha256=checksum, + signature_evidence=signature_evidence, + parser_version=definition.parser_version, + licence_notes=definition.licence_notes, + trust_notes=definition.trust_notes, + connector_run_id=run.id, + transport_evidence={ + "etag": response.headers.get("etag"), + "last_modified": response.headers.get( + "last-modified" + ), + "digest": response.headers.get("digest"), + "attempts": response.attempts, + "tls_required": definition.source_url is not None, + }, + raw_content=response.content, + ) + db.add(snapshot) + db.flush() + run.status = "succeeded" + run.snapshot_id = snapshot.id + run.finished_at = acquired_at + db.flush() + return SanctionsAcquisitionResult( + run_id=run.id, + provider_id=provider_id, + status=run.status, + snapshot=_snapshot_dto(snapshot), + ) + except Exception as exc: + run.status = _failure_status( + exc, + latest=latest, + now=now, + ) + run.error = _bounded_error(exc) + run.finished_at = utcnow() + db.flush() + return SanctionsAcquisitionResult( + run_id=run.id, + provider_id=provider_id, + status=run.status, + snapshot=( + _snapshot_dto(latest) + if latest is not None + else None + ), + error=run.error, + ) + + def list_snapshots( + self, + session: object, + principal: object, + *, + limit: int = 100, + ) -> Sequence[SanctionsSnapshotReference]: + db, api_principal = _context( + session, + principal, + SANCTIONS_READ_SCOPE, + ) + return tuple( + _snapshot_dto(item) + for item in db.scalars( + select(ConnectorSanctionsSnapshot) + .where( + ConnectorSanctionsSnapshot.tenant_id + == api_principal.tenant_id + ) + .order_by( + ConnectorSanctionsSnapshot.acquired_at.desc(), + ConnectorSanctionsSnapshot.id.desc(), + ) + .limit(max(1, min(int(limit), 500))) + ) + ) + + def get_snapshot( + self, + session: object, + principal: object, + *, + snapshot_ref: str, + ) -> SanctionsSnapshotReference | None: + db, api_principal = _context( + session, + principal, + SANCTIONS_READ_SCOPE, + ) + item = _snapshot_record( + db, + tenant_id=api_principal.tenant_id, + snapshot_ref=snapshot_ref, + ) + return _snapshot_dto(item) if item is not None else None + + def read_snapshot( + self, + session: object, + principal: object, + *, + snapshot_ref: str, + ) -> SanctionsSnapshotPayload: + db, api_principal = _context( + session, + principal, + SANCTIONS_READ_SCOPE, + ) + item = _snapshot_record( + db, + tenant_id=api_principal.tenant_id, + snapshot_ref=snapshot_ref, + ) + if item is None: + raise SanctionsSourceNotFoundError( + "Sanctions snapshot was not found." + ) + checksum = hashlib.sha256(item.raw_content).hexdigest() + if checksum != item.sha256: + raise SanctionsSourceUnexpectedChangeError( + "Stored sanctions evidence failed checksum verification." + ) + return SanctionsSnapshotPayload( + snapshot=_snapshot_dto(item), + content=bytes(item.raw_content), + ) + + def list_runs( + self, + session: object, + principal: object, + *, + limit: int = 100, + ) -> tuple[ConnectorSanctionsAcquisitionRun, ...]: + db, api_principal = _context( + session, + principal, + SANCTIONS_READ_SCOPE, + ) + return tuple( + db.scalars( + select(ConnectorSanctionsAcquisitionRun) + .where( + ConnectorSanctionsAcquisitionRun.tenant_id + == api_principal.tenant_id + ) + .order_by( + ConnectorSanctionsAcquisitionRun.started_at.desc(), + ConnectorSanctionsAcquisitionRun.id.desc(), + ) + .limit(max(1, min(int(limit), 500))) + ) + ) + + +class _AllowlistedRedirectHandler(HTTPRedirectHandler): + def __init__(self, allowed_hosts: frozenset[str]) -> None: + super().__init__() + self._allowed_hosts = allowed_hosts + + def redirect_request( + self, + req, + fp, + code, + msg, + headers, + newurl, + ): + parsed = urlsplit(newurl) + if ( + parsed.scheme != "https" + or (parsed.hostname or "").casefold() + not in self._allowed_hosts + ): + raise SanctionsSourceUnexpectedChangeError( + "Sanctions source redirected outside its allowlist." + ) + return super().redirect_request( + req, + fp, + code, + msg, + headers, + newurl, + ) + + +def _validate_response( + definition: SanctionsSourceDefinition, + response: TransportResponse, +) -> dict[str, datetime | None]: + if len(response.content) > MAX_SANCTIONS_BYTES: + raise SanctionsSourceUnexpectedChangeError( + "Sanctions evidence exceeds the configured size limit." + ) + content_type = _content_type(response.headers) + if content_type not in ALLOWED_XML_TYPES: + raise SanctionsSourceUnexpectedChangeError( + f"Unexpected sanctions content type: {content_type}" + ) + _checksum_evidence(response.headers, response.content) + try: + root = ElementTree.fromstring(response.content) + except Exception as exc: + raise SanctionsSourceMalformedError( + "Sanctions XML is malformed or unsafe." + ) from exc + if _local_name(root.tag) != "CONSOLIDATED_LIST": + raise SanctionsSourceUnexpectedChangeError( + "Sanctions XML root contract changed unexpectedly." + ) + child_names = {_local_name(child.tag) for child in root} + if not {"INDIVIDUALS", "ENTITIES"}.issubset(child_names): + raise SanctionsSourceUnexpectedChangeError( + "Sanctions XML no longer contains expected list sections." + ) + generated = _parse_datetime( + root.attrib.get("dateGenerated") + or root.attrib.get("generated") + ) + return { + "publication_at": generated, + "effective_at": generated, + } + + +def _checksum_evidence( + headers: Mapping[str, str], + content: bytes, +) -> dict[str, object]: + digest = headers.get("digest") + evidence: dict[str, object] = { + "available": False, + "verified": False, + "algorithm": None, + "publisher_value": None, + "computed_sha256": hashlib.sha256(content).hexdigest(), + } + if not digest: + return evidence + for part in digest.split(","): + algorithm, separator, encoded = part.strip().partition("=") + if separator and algorithm.casefold() in {"sha-256", "sha256"}: + try: + expected = base64.b64decode(encoded.strip()) + except ValueError as exc: + raise SanctionsSourceUnexpectedChangeError( + "Publisher checksum evidence is malformed." + ) from exc + actual = hashlib.sha256(content).digest() + if expected != actual: + raise SanctionsSourceUnexpectedChangeError( + "Publisher checksum verification failed." + ) + evidence.update( + { + "available": True, + "verified": True, + "algorithm": "sha-256", + "publisher_value": encoded.strip(), + } + ) + return evidence + return evidence + + +def _source_version( + metadata: Mapping[str, datetime | None], + headers: Mapping[str, str], + checksum: str, +) -> str: + published = metadata.get("publication_at") + if isinstance(published, datetime): + return published.isoformat() + for key in ("etag", "last-modified"): + value = str(headers.get(key) or "").strip() + if value: + return value[:255] + return f"sha256:{checksum}"[:255] + + +def _conditional_headers( + latest: ConnectorSanctionsSnapshot | None, +) -> dict[str, str]: + if latest is None: + return {} + transport = latest.transport_evidence or {} + result: dict[str, str] = {} + if transport.get("etag"): + result["If-None-Match"] = str(transport["etag"]) + if transport.get("last_modified"): + result["If-Modified-Since"] = str( + transport["last_modified"] + ) + return result + + +def _snapshot_record( + session: Session, + *, + tenant_id: str, + snapshot_ref: str, +) -> ConnectorSanctionsSnapshot | None: + snapshot_id = snapshot_ref.removeprefix( + "sanctions-snapshot:" + ) + if not snapshot_id or snapshot_id == snapshot_ref: + return None + return session.scalar( + select(ConnectorSanctionsSnapshot).where( + ConnectorSanctionsSnapshot.id == snapshot_id, + ConnectorSanctionsSnapshot.tenant_id == tenant_id, + ) + ) + + +def _snapshot_dto( + item: ConnectorSanctionsSnapshot, +) -> SanctionsSnapshotReference: + return SanctionsSnapshotReference( + ref=f"sanctions-snapshot:{item.id}", + tenant_id=item.tenant_id, + provider_id=item.provider_id, + publisher=item.publisher, + jurisdiction=item.jurisdiction, + list_type=item.list_type, + source_id=item.source_id, + source_version=item.source_version, + publication_at=item.publication_at, + effective_at=item.effective_at, + acquired_at=item.acquired_at, + content_type=item.content_type, + byte_count=item.byte_count, + sha256=item.sha256, + parser_version=item.parser_version, + raw_evidence_ref=f"connector-evidence:{item.id}", + connector_run_id=item.connector_run_id, + signature_evidence=dict(item.signature_evidence), + licence_notes=item.licence_notes, + trust_notes=item.trust_notes, + transport_evidence=dict(item.transport_evidence), + ) + + +def _context( + session: object, + principal: object, + required_scope: str, +) -> tuple[Session, ApiPrincipal]: + if not isinstance(session, Session): + raise TypeError( + "Sanctions source operations require a SQLAlchemy session." + ) + if not isinstance(principal, ApiPrincipal): + raise SanctionsSourceAccessError( + "A tenant API principal is required." + ) + if not ( + has_scope(principal, required_scope) + or has_scope(principal, CONNECTOR_ADMIN_SCOPE) + ): + raise SanctionsSourceAccessError( + f"Missing scope: {required_scope}" + ) + return session, principal + + +def _failure_status( + exc: Exception, + *, + latest: ConnectorSanctionsSnapshot | None, + now: datetime, +) -> str: + if isinstance(exc, SanctionsSourceMalformedError): + return "malformed" + if isinstance(exc, SanctionsSourceUnexpectedChangeError): + return "unexpected_change" + if ( + latest is not None + and _as_utc(now) - _as_utc(latest.acquired_at) + > STALE_AFTER + ): + return "stale" + return "unavailable" + + +def _safe_headers(headers: object) -> dict[str, str]: + allowed = { + "content-type", + "content-length", + "etag", + "last-modified", + "digest", + } + return { + key.casefold(): str(value)[:2000] + for key, value in getattr(headers, "items", lambda: ())() + if key.casefold() in allowed + } + + +def _bounded_read(response: object) -> bytes: + content_length = getattr(response, "headers", {}).get( + "Content-Length" + ) + if content_length: + try: + if int(content_length) > MAX_SANCTIONS_BYTES: + raise SanctionsSourceUnexpectedChangeError( + "Sanctions evidence exceeds the configured size limit." + ) + except ValueError as exc: + raise SanctionsSourceUnexpectedChangeError( + "Sanctions source sent an invalid content length." + ) from exc + chunks: list[bytes] = [] + size = 0 + while True: + chunk = response.read(min(65_536, MAX_SANCTIONS_BYTES + 1 - size)) + if not chunk: + break + size += len(chunk) + if size > MAX_SANCTIONS_BYTES: + raise SanctionsSourceUnexpectedChangeError( + "Sanctions evidence exceeds the configured size limit." + ) + chunks.append(chunk) + return b"".join(chunks) + + +def _safe_url(value: str | None) -> str: + if not value: + return "" + parsed = urlsplit(value) + if parsed.scheme == "fixture": + return value + if parsed.scheme != "https" or not parsed.hostname: + raise SanctionsSourceUnexpectedChangeError( + "Sanctions source URL is not HTTPS." + ) + if parsed.username or parsed.password: + raise SanctionsSourceUnexpectedChangeError( + "Sanctions source URL contains credentials." + ) + return urlunsplit( + ( + parsed.scheme, + parsed.netloc, + parsed.path, + "", + "", + ) + ) + + +def _content_type(headers: Mapping[str, str]) -> str: + return ( + str(headers.get("content-type") or "") + .partition(";")[0] + .strip() + .casefold() + ) + + +def _parse_datetime(value: str | None) -> datetime | None: + clean = str(value or "").strip() + if not clean: + return None + try: + if clean.endswith("Z"): + clean = clean[:-1] + "+00:00" + return _as_utc(datetime.fromisoformat(clean)) + except ValueError: + try: + return _as_utc(parsedate_to_datetime(clean)) + except (TypeError, ValueError): + return None + + +def _local_name(value: str) -> str: + return value.rsplit("}", 1)[-1] + + +def _actor_id(principal: ApiPrincipal) -> str | None: + return ( + principal.account_id + or principal.membership_id + or principal.identity_id + ) + + +def _as_utc(value: datetime) -> datetime: + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + + +def _bounded_error(exc: Exception | None) -> str: + if exc is None: + return "Unknown transport failure" + clean = " ".join(str(exc).split()) + return (clean or type(exc).__name__)[:2000] + + +__all__ = [ + "MAX_SANCTIONS_BYTES", + "SANCTIONS_READ_SCOPE", + "SANCTIONS_REFRESH_SCOPE", + "SOURCE_DEFINITIONS", + "SYNTHETIC_PROVIDER_ID", + "SYNTHETIC_UN_XML", + "SanctionsAcquisitionResult", + "SanctionsSourceAccessError", + "SanctionsSourceDefinition", + "SanctionsSourceError", + "SanctionsSourceMalformedError", + "SanctionsSourceNotFoundError", + "SanctionsSourceUnexpectedChangeError", + "SanctionsTransport", + "SqlSanctionsSnapshotProvider", + "TransportResponse", + "UNSC_PROVIDER_ID", + "UrllibSanctionsTransport", +] diff --git a/src/govoplan_connectors/backend/schemas.py b/src/govoplan_connectors/backend/schemas.py index 986d3c8..1ebd26d 100644 --- a/src/govoplan_connectors/backend/schemas.py +++ b/src/govoplan_connectors/backend/schemas.py @@ -1,5 +1,6 @@ from __future__ import annotations +from datetime import datetime from typing import Any, Literal from pydantic import BaseModel, Field, model_validator @@ -69,8 +70,84 @@ class TabularSourceDeleteResponse(BaseModel): source_ref: str +class SanctionsSourceResponse(BaseModel): + provider_id: str + publisher: str + jurisdiction: str + list_type: str + source_id: str + source_url: str | None + parser_version: str + licence_notes: str + trust_notes: str + + +class SanctionsSourceListResponse(BaseModel): + sources: list[SanctionsSourceResponse] + + +class SanctionsSnapshotResponse(BaseModel): + ref: str + provider_id: str + publisher: str + jurisdiction: str + list_type: str + source_id: str + source_version: str + publication_at: datetime | None + effective_at: datetime | None + acquired_at: datetime + content_type: str + byte_count: int + sha256: str + parser_version: str + raw_evidence_ref: str + connector_run_id: str + signature_evidence: dict[str, Any] + licence_notes: str | None + trust_notes: str | None + transport_evidence: dict[str, Any] + + +class SanctionsSnapshotListResponse(BaseModel): + snapshots: list[SanctionsSnapshotResponse] + + +class SanctionsAcquisitionRunResponse(BaseModel): + id: str + provider_id: str + source_id: str + status: str + attempt_count: int + request_evidence: dict[str, Any] + response_evidence: dict[str, Any] + started_at: datetime + finished_at: datetime | None + snapshot_id: str | None + error: str | None + + +class SanctionsAcquisitionRunListResponse(BaseModel): + runs: list[SanctionsAcquisitionRunResponse] + + +class SanctionsRefreshResponse(BaseModel): + run_id: str + provider_id: str + status: str + snapshot: SanctionsSnapshotResponse | None + error: str | None + + __all__ = [ "SnapshotCreateRequest", + "SanctionsAcquisitionRunListResponse", + "SanctionsAcquisitionRunResponse", + "SanctionsRefreshResponse", + "SanctionsSnapshotListResponse", + "SanctionsSnapshotResponse", + "SanctionsSourceListResponse", + "SanctionsSourceResponse", "TabularColumnResponse", "TabularSourceDeleteResponse", "TabularSourceListResponse", diff --git a/tests/test_manifest.py b/tests/test_manifest.py index 572421d..31871c8 100644 --- a/tests/test_manifest.py +++ b/tests/test_manifest.py @@ -6,6 +6,9 @@ from govoplan_core.core.tabular_sources import ( CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER, CAPABILITY_CONNECTORS_TABULAR_SOURCES, ) +from govoplan_core.core.sanctions import ( + CAPABILITY_CONNECTORS_SANCTIONS_SNAPSHOTS, +) from govoplan_connectors.backend.manifest import manifest @@ -25,6 +28,10 @@ class ConnectorsManifestTests(unittest.TestCase): CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER, manifest.capability_factories, ) + self.assertIn( + CAPABILITY_CONNECTORS_SANCTIONS_SNAPSHOTS, + manifest.capability_factories, + ) self.assertIsNotNone(manifest.migration_spec) diff --git a/tests/test_migrations.py b/tests/test_migrations.py index f223fbd..9686a13 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -24,13 +24,21 @@ class ConnectorsMigrationTests(unittest.TestCase): try: with engine.connect() as connection: self.assertIn( - "e6b7c8d9f0a1", + "f7c8d9e0a1b2", set(MigrationContext.configure(connection).get_current_heads()), ) self.assertIn( "connector_tabular_sources", inspect(connection).get_table_names(), ) + self.assertIn( + "connector_sanctions_snapshots", + inspect(connection).get_table_names(), + ) + self.assertIn( + "connector_sanctions_acquisition_runs", + inspect(connection).get_table_names(), + ) finally: engine.dispose() diff --git a/tests/test_sanctions_sources.py b/tests/test_sanctions_sources.py new file mode 100644 index 0000000..bb32009 --- /dev/null +++ b/tests/test_sanctions_sources.py @@ -0,0 +1,324 @@ +from __future__ import annotations + +from datetime import timedelta +import hashlib +import unittest +from unittest.mock import patch +from urllib.error import URLError + +from sqlalchemy import create_engine +from sqlalchemy.orm import Session + +from govoplan_connectors.backend.db.models import ( + ConnectorSanctionsAcquisitionRun, + ConnectorSanctionsSnapshot, +) +from govoplan_connectors.backend.sanctions_sources import ( + SANCTIONS_READ_SCOPE, + SANCTIONS_REFRESH_SCOPE, + SOURCE_DEFINITIONS, + SYNTHETIC_PROVIDER_ID, + SYNTHETIC_UN_XML, + SanctionsSourceError, + SqlSanctionsSnapshotProvider, + TransportResponse, + UNSC_PROVIDER_ID, + UrllibSanctionsTransport, +) +from govoplan_core.auth import ApiPrincipal +from govoplan_core.core.access import PrincipalRef +from govoplan_core.db.base import Base, utcnow + + +def principal( + tenant_id: str = "tenant-1", + *, + scopes: tuple[str, ...] = ( + SANCTIONS_READ_SCOPE, + SANCTIONS_REFRESH_SCOPE, + ), +) -> ApiPrincipal: + return ApiPrincipal( + principal=PrincipalRef( + account_id="account-1", + membership_id="membership-1", + tenant_id=tenant_id, + scopes=frozenset(scopes), + ), + account=object(), + user=object(), + ) + + +class _Transport: + def __init__(self, responses): + self.responses = list(responses) + self.headers = [] + + def fetch(self, definition, *, headers): + del definition + self.headers.append(dict(headers)) + response = self.responses.pop(0) + if isinstance(response, Exception): + raise response + return response + + +def response( + content: bytes = SYNTHETIC_UN_XML, + *, + status: int = 200, + content_type: str = "application/xml", + etag: str = '"fixture-v1"', +) -> TransportResponse: + return TransportResponse( + status=status, + final_url="https://scsanctions.un.org/consolidated.xml", + headers={ + "content-type": content_type, + "etag": etag, + }, + content=content, + attempts=1, + ) + + +class SanctionsSourcesTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all( + self.engine, + tables=( + ConnectorSanctionsAcquisitionRun.__table__, + ConnectorSanctionsSnapshot.__table__, + ), + ) + self.session = Session(self.engine) + + def tearDown(self) -> None: + self.session.close() + self.engine.dispose() + + def test_fixture_refreshes_are_immutable_and_evidence_is_readable( + self, + ) -> None: + provider = SqlSanctionsSnapshotProvider() + + first = provider.refresh_source( + self.session, + principal(), + provider_id=SYNTHETIC_PROVIDER_ID, + ) + second = provider.refresh_source( + self.session, + principal(), + provider_id=SYNTHETIC_PROVIDER_ID, + ) + self.session.commit() + + self.assertEqual("succeeded", first.status) + self.assertEqual("succeeded", second.status) + self.assertNotEqual(first.snapshot.ref, second.snapshot.ref) + self.assertEqual( + hashlib.sha256(SYNTHETIC_UN_XML).hexdigest(), + first.snapshot.sha256, + ) + self.assertEqual( + SYNTHETIC_UN_XML, + provider.read_snapshot( + self.session, + principal(), + snapshot_ref=first.snapshot.ref, + ).content, + ) + runs = provider.list_runs( + self.session, + principal(), + ) + self.assertTrue( + all( + run.request_evidence["subject_data_transmitted"] + is False + for run in runs + ) + ) + + def test_conditional_fetch_reuses_prior_immutable_snapshot(self) -> None: + transport = _Transport( + ( + response(), + response(b"", status=304), + ) + ) + provider = SqlSanctionsSnapshotProvider(transport) + first = provider.refresh_source( + self.session, + principal(), + provider_id=UNSC_PROVIDER_ID, + ) + second = provider.refresh_source( + self.session, + principal(), + provider_id=UNSC_PROVIDER_ID, + ) + + self.assertEqual("not_modified", second.status) + self.assertEqual(first.snapshot.ref, second.snapshot.ref) + self.assertEqual( + {'If-None-Match': '"fixture-v1"'}, + transport.headers[1], + ) + self.assertEqual( + 1, + self.session.query(ConnectorSanctionsSnapshot).count(), + ) + + def test_malformed_and_changed_sources_have_explicit_health( + self, + ) -> None: + cases = ( + ( + b"", + "application/xml", + "malformed", + ), + ( + b"", + "application/xml", + "unexpected_change", + ), + ( + SYNTHETIC_UN_XML, + "text/html", + "unexpected_change", + ), + ) + for payload, content_type, expected in cases: + with self.subTest(expected=expected, content_type=content_type): + provider = SqlSanctionsSnapshotProvider( + _Transport( + ( + response( + payload, + content_type=content_type, + ), + ) + ) + ) + result = provider.refresh_source( + self.session, + principal(), + provider_id=UNSC_PROVIDER_ID, + ) + self.assertEqual(expected, result.status) + self.assertIsNotNone(result.error) + + def test_unavailable_source_becomes_stale_when_evidence_is_old( + self, + ) -> None: + provider = SqlSanctionsSnapshotProvider( + _Transport( + ( + response(), + SanctionsSourceError("offline"), + ) + ) + ) + first = provider.refresh_source( + self.session, + principal(), + provider_id=UNSC_PROVIDER_ID, + ) + record = self.session.get( + ConnectorSanctionsSnapshot, + first.snapshot.ref.removeprefix("sanctions-snapshot:"), + ) + record.acquired_at = utcnow() - timedelta(days=3) + + result = provider.refresh_source( + self.session, + principal(), + provider_id=UNSC_PROVIDER_ID, + ) + + self.assertEqual("stale", result.status) + self.assertEqual(first.snapshot.ref, result.snapshot.ref) + + def test_snapshot_access_is_tenant_and_scope_isolated(self) -> None: + provider = SqlSanctionsSnapshotProvider() + created = provider.refresh_source( + self.session, + principal(), + provider_id=SYNTHETIC_PROVIDER_ID, + ) + + self.assertIsNone( + provider.get_snapshot( + self.session, + principal("tenant-2"), + snapshot_ref=created.snapshot.ref, + ) + ) + with self.assertRaisesRegex(Exception, "Missing scope"): + provider.list_snapshots( + self.session, + principal(scopes=()), + ) + + def test_transport_retries_transient_network_failures(self) -> None: + class _Headers(dict): + pass + + class _Response: + status = 200 + headers = _Headers( + { + "Content-Type": "application/xml", + "Content-Length": str(len(SYNTHETIC_UN_XML)), + } + ) + + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + def geturl(self): + return ( + "https://scsanctions.un.org/" + "resources/xml/en/consolidated.xml" + ) + + def read(self, size): + del size + if hasattr(self, "_read"): + return b"" + self._read = True + return SYNTHETIC_UN_XML + + opener = unittest.mock.Mock() + opener.open.side_effect = ( + URLError("temporary"), + URLError("temporary"), + _Response(), + ) + sleeps = [] + transport = UrllibSanctionsTransport( + sleeper=sleeps.append + ) + with patch( + "govoplan_connectors.backend.sanctions_sources.build_opener", + return_value=opener, + ): + fetched = transport.fetch( + SOURCE_DEFINITIONS[UNSC_PROVIDER_ID], + headers={}, + ) + + self.assertEqual(3, fetched.attempts) + self.assertEqual([1.0, 2.0], sleeps) + + +if __name__ == "__main__": + unittest.main()