feat: acquire immutable sanctions snapshots

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"]

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",
+]

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",

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")

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"]

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"""<?xml version="1.0" encoding="UTF-8"?>
+<CONSOLIDATED_LIST dateGenerated="2026-07-01T00:00:00Z">
+  <INDIVIDUALS>
+    <INDIVIDUAL>
+      <DATAID>fixture-person-1</DATAID>
+      <VERSIONNUM>1</VERSIONNUM>
+      <FIRST_NAME>ALEX</FIRST_NAME>
+      <SECOND_NAME>EXAMPLE</SECOND_NAME>
+      <UN_LIST_TYPE>Fixture regime</UN_LIST_TYPE>
+      <REFERENCE_NUMBER>FXi.001</REFERENCE_NUMBER>
+      <LISTED_ON>2026-01-15</LISTED_ON>
+      <INDIVIDUAL_ALIAS>
+        <QUALITY>Good</QUALITY>
+        <ALIAS_NAME>ALEXANDER EXAMPLE</ALIAS_NAME>
+      </INDIVIDUAL_ALIAS>
+      <INDIVIDUAL_DATE_OF_BIRTH>
+        <DATE>1980-05-12</DATE>
+      </INDIVIDUAL_DATE_OF_BIRTH>
+      <INDIVIDUAL_DOCUMENT>
+        <TYPE_OF_DOCUMENT>Passport</TYPE_OF_DOCUMENT>
+        <NUMBER>P-FIXTURE-001</NUMBER>
+      </INDIVIDUAL_DOCUMENT>
+      <INDIVIDUAL_ADDRESS>
+        <CITY>Example City</CITY>
+        <COUNTRY>Exampleland</COUNTRY>
+      </INDIVIDUAL_ADDRESS>
+    </INDIVIDUAL>
+  </INDIVIDUALS>
+  <ENTITIES>
+    <ENTITY>
+      <DATAID>fixture-entity-1</DATAID>
+      <VERSIONNUM>1</VERSIONNUM>
+      <FIRST_NAME>EXAMPLE TRADING LTD</FIRST_NAME>
+      <UN_LIST_TYPE>Fixture regime</UN_LIST_TYPE>
+      <REFERENCE_NUMBER>FXe.001</REFERENCE_NUMBER>
+      <LISTED_ON>2026-02-20</LISTED_ON>
+      <ENTITY_ALIAS>
+        <QUALITY>Good</QUALITY>
+        <ALIAS_NAME>EXAMPLE TRADING</ALIAS_NAME>
+      </ENTITY_ALIAS>
+      <ENTITY_ADDRESS>
+        <CITY>Example Port</CITY>
+        <COUNTRY>Exampleland</COUNTRY>
+      </ENTITY_ADDRESS>
+    </ENTITY>
+  </ENTITIES>
+</CONSOLIDATED_LIST>
+"""
+
+
+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",
+]

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",

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)
 
 

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()
 

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"<CONSOLIDATED_LIST>",
+                "application/xml",
+                "malformed",
+            ),
+            (
+                b"<DIFFERENT><INDIVIDUALS/><ENTITIES/></DIFFERENT>",
+                "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()