GovOPlaN/govoplan-risk-compliance
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: GovOPlaN:v0.1.8
Branches Tags
GovOPlaN:main
GovOPlaN:v0.1.8 GovOPlaN:v0.1.7
..
compare: GovOPlaN:main
Branches Tags
GovOPlaN:main
GovOPlaN:v0.1.8 GovOPlaN:v0.1.7

2 Commits
v0.1.8 ... main

Author SHA1 Message Date
Albrecht Degering
11404ac42f feat: implement sanctions screening vertical 2026-07-29 18:46:53 +02:00
Albrecht Degering
05c3a2257a fix(manifest): normalize runtime module id 2026-07-20 20:01:50 +02:00
25 changed files with 6536 additions and 41 deletions
Expand all files Collapse all files

11
README.md
View File

@@ -4,9 +4,13 @@
**Repository type:** module (domain).
<!-- govoplan-repository-type:end -->
`govoplan-risk-compliance` is the GovOPlaN platform module seed for risk and compliance workflows for data protection incidents, DPIAs, compliance controls, audit measures, risk registers, and internal-control evidence.
`govoplan-risk-compliance` owns risk and compliance workflows for data protection incidents, DPIAs, compliance controls, audit measures, risk registers, internal-control evidence, and legal screening decisions.
This repository is initialized as a discoverable module seed. It exposes a module manifest, initial permissions, role templates, documentation metadata, Gitea workflow templates, and a focused manifest test. It intentionally does not yet add HTTP routes, database models, migrations, or WebUI navigation.
Its runtime module ID is `risk_compliance`; the repository and Python distribution retain the hyphenated `govoplan-risk-compliance` name.
The first complete vertical is sanctions screening. Connectors acquires immutable source evidence; Risk Compliance imports and normalizes exact list versions, runs deterministic version-pinned screening, and presents potential matches to an independent reviewer. Fuzzy matching only creates review candidates and never confirms a legal match.
The module includes database migrations, tenant-isolated APIs, an operational WebUI, append-only dispositions, time-bounded false-positive exceptions, and audit events. Queue and audit summaries deliberately retain only the minimum subject data needed for the workflow.
## Initial Ownership
@@ -16,6 +20,8 @@ This repository is initialized as a discoverable module seed. It exposes a modul
- data protection incident records
- audit measures
- internal-control evidence
- immutable sanctions list catalogues
- sanctions screening and reviewer dispositions
## Boundaries
@@ -38,6 +44,7 @@ Expected optional integrations:
- files
- tasks
- notifications
- connectors
## Development Install

31
package.json
View File

@@ -1,8 +1,33 @@
{
"name": "@govoplan/risk-compliance",
"name": "@govoplan/risk-compliance-webui",
"version": "0.1.8",
"private": true,
"description": "GovOPlaN Risk Compliance platform module seed.",
"type": "module",
"peerDependencies": {}
"main": "webui/src/index.ts",
"module": "webui/src/index.ts",
"types": "webui/src/index.ts",
"exports": {
".": {
"types": "./webui/src/index.ts",
"import": "./webui/src/index.ts"
},
"./styles/risk-compliance.css": "./webui/src/styles/risk-compliance.css"
},
"files": [
"webui/src",
"README.md",
"LICENSE"
],
"peerDependencies": {
"@govoplan/core-webui": "^0.1.14",
"lucide-react": "^1.23.0",
"react": "^19.0.0",
"react-dom": "^19.0.0",
"react-router-dom": ">=7.18.2 <8"
},
"peerDependenciesMeta": {
"@govoplan/core-webui": {
"optional": true
}
}
}

3
pyproject.toml
View File

@@ -11,6 +11,7 @@ requires-python = ">=3.12"
license = { file = "LICENSE" }
authors = [{ name = "GovOPlaN" }]
dependencies = [
"defusedxml>=0.7.1",
"govoplan-core>=0.1.8",
"govoplan-access>=0.1.8",
]
@@ -22,4 +23,4 @@ where = ["src"]
govoplan_risk_compliance = ["py.typed"]
[project.entry-points."govoplan.modules"]
"risk-compliance" = "govoplan_risk_compliance.backend.manifest:get_manifest"
risk_compliance = "govoplan_risk_compliance.backend.manifest:get_manifest"

2
src/govoplan_risk_compliance/backend/db/__init__.py Normal file
View File

@@ -0,0 +1,2 @@
"""Risk Compliance persistence models."""

857
src/govoplan_risk_compliance/backend/db/models.py Normal file
View File

@@ -0,0 +1,857 @@
from __future__ import annotations
from datetime import date, datetime
from typing import Any
import uuid
from sqlalchemy import (
Date,
DateTime,
ForeignKey,
Index,
Integer,
JSON,
String,
Text,
UniqueConstraint,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship
from govoplan_core.db.base import Base, TimestampMixin
def new_uuid() -> str:
return str(uuid.uuid4())
class RiskSanctionsListSnapshot(Base, TimestampMixin):
__tablename__ = "risk_sanctions_list_snapshots"
__table_args__ = (
UniqueConstraint(
"tenant_id",
"connector_snapshot_ref",
name="uq_risk_sanctions_connector_snapshot",
),
Index(
"ix_risk_sanctions_snapshot_current",
"tenant_id",
"provider_id",
"status",
"acquired_at",
),
)
id: Mapped[str] = mapped_column(
String(36),
primary_key=True,
default=new_uuid,
)
tenant_id: Mapped[str | None] = mapped_column(
String(36),
nullable=True,
index=True,
)
visibility: Mapped[str] = mapped_column(
String(20),
default="tenant",
nullable=False,
index=True,
)
connector_snapshot_ref: Mapped[str] = mapped_column(
String(300),
nullable=False,
)
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,
)
sha256: Mapped[str] = mapped_column(
String(64),
nullable=False,
index=True,
)
connector_run_id: Mapped[str] = mapped_column(
String(36),
nullable=False,
)
raw_evidence_ref: Mapped[str] = mapped_column(
String(300),
nullable=False,
)
source_parser_version: Mapped[str] = mapped_column(
String(100),
nullable=False,
)
normalization_version: Mapped[str] = mapped_column(
String(100),
nullable=False,
)
signature_evidence: Mapped[dict[str, Any]] = mapped_column(
JSON,
default=dict,
nullable=False,
)
provenance: Mapped[dict[str, Any]] = mapped_column(
JSON,
default=dict,
nullable=False,
)
entry_count: Mapped[int] = mapped_column(
Integer,
default=0,
nullable=False,
)
status: Mapped[str] = mapped_column(
String(30),
default="active",
nullable=False,
index=True,
)
imported_by: Mapped[str | None] = mapped_column(
String(255),
nullable=True,
index=True,
)
imported_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
nullable=False,
)
entries: Mapped[list["RiskSanctionsEntry"]] = relationship(
back_populates="snapshot",
cascade="all, delete-orphan",
)
class RiskSanctionsEntry(Base, TimestampMixin):
__tablename__ = "risk_sanctions_entries"
__table_args__ = (
UniqueConstraint(
"snapshot_id",
"source_entry_id",
name="uq_risk_sanctions_entry_source",
),
Index(
"ix_risk_sanctions_entry_name",
"snapshot_id",
"normalized_name",
),
)
id: Mapped[str] = mapped_column(
String(36),
primary_key=True,
default=new_uuid,
)
snapshot_id: Mapped[str] = mapped_column(
ForeignKey(
"risk_sanctions_list_snapshots.id",
ondelete="CASCADE",
),
nullable=False,
index=True,
)
source_entry_id: Mapped[str] = mapped_column(
String(255),
nullable=False,
index=True,
)
subject_type: Mapped[str] = mapped_column(
String(30),
nullable=False,
index=True,
)
primary_name: Mapped[str] = mapped_column(
String(1000),
nullable=False,
)
normalized_name: Mapped[str] = mapped_column(
String(1000),
nullable=False,
index=True,
)
original_script_name: Mapped[str | None] = mapped_column(
String(1000),
nullable=True,
)
reference_number: Mapped[str | None] = mapped_column(
String(255),
nullable=True,
index=True,
)
listed_on: Mapped[date | None] = mapped_column(
Date,
nullable=True,
)
programmes: Mapped[list[str]] = mapped_column(
JSON,
default=list,
nullable=False,
)
measures: Mapped[list[str]] = mapped_column(
JSON,
default=list,
nullable=False,
)
raw_evidence_locator: Mapped[str] = mapped_column(
String(500),
nullable=False,
)
details: Mapped[dict[str, Any]] = mapped_column(
JSON,
default=dict,
nullable=False,
)
snapshot: Mapped[RiskSanctionsListSnapshot] = relationship(
back_populates="entries"
)
aliases: Mapped[list["RiskSanctionsAlias"]] = relationship(
back_populates="entry",
cascade="all, delete-orphan",
)
identifiers: Mapped[list["RiskSanctionsIdentifier"]] = relationship(
back_populates="entry",
cascade="all, delete-orphan",
)
dates: Mapped[list["RiskSanctionsDate"]] = relationship(
back_populates="entry",
cascade="all, delete-orphan",
)
addresses: Mapped[list["RiskSanctionsAddress"]] = relationship(
back_populates="entry",
cascade="all, delete-orphan",
)
class RiskSanctionsAlias(Base):
__tablename__ = "risk_sanctions_aliases"
__table_args__ = (
Index(
"ix_risk_sanctions_alias_name",
"entry_id",
"normalized_name",
),
)
id: Mapped[str] = mapped_column(
String(36),
primary_key=True,
default=new_uuid,
)
entry_id: Mapped[str] = mapped_column(
ForeignKey(
"risk_sanctions_entries.id",
ondelete="CASCADE",
),
nullable=False,
index=True,
)
name: Mapped[str] = mapped_column(
String(1000),
nullable=False,
)
normalized_name: Mapped[str] = mapped_column(
String(1000),
nullable=False,
index=True,
)
quality: Mapped[str | None] = mapped_column(
String(100),
nullable=True,
)
alias_type: Mapped[str] = mapped_column(
String(50),
default="aka",
nullable=False,
)
entry: Mapped[RiskSanctionsEntry] = relationship(
back_populates="aliases"
)
class RiskSanctionsIdentifier(Base):
__tablename__ = "risk_sanctions_identifiers"
__table_args__ = (
Index(
"ix_risk_sanctions_identifier_value",
"entry_id",
"normalized_value",
),
)
id: Mapped[str] = mapped_column(
String(36),
primary_key=True,
default=new_uuid,
)
entry_id: Mapped[str] = mapped_column(
ForeignKey(
"risk_sanctions_entries.id",
ondelete="CASCADE",
),
nullable=False,
index=True,
)
identifier_type: Mapped[str] = mapped_column(
String(100),
nullable=False,
)
value: Mapped[str] = mapped_column(
String(1000),
nullable=False,
)
normalized_value: Mapped[str] = mapped_column(
String(1000),
nullable=False,
index=True,
)
issuing_country: Mapped[str | None] = mapped_column(
String(255),
nullable=True,
)
note: Mapped[str | None] = mapped_column(
Text,
nullable=True,
)
entry: Mapped[RiskSanctionsEntry] = relationship(
back_populates="identifiers"
)
class RiskSanctionsDate(Base):
__tablename__ = "risk_sanctions_dates"
id: Mapped[str] = mapped_column(
String(36),
primary_key=True,
default=new_uuid,
)
entry_id: Mapped[str] = mapped_column(
ForeignKey(
"risk_sanctions_entries.id",
ondelete="CASCADE",
),
nullable=False,
index=True,
)
date_type: Mapped[str] = mapped_column(
String(50),
nullable=False,
)
value: Mapped[str] = mapped_column(
String(100),
nullable=False,
)
precision: Mapped[str] = mapped_column(
String(30),
default="unknown",
nullable=False,
)
entry: Mapped[RiskSanctionsEntry] = relationship(
back_populates="dates"
)
class RiskSanctionsAddress(Base):
__tablename__ = "risk_sanctions_addresses"
id: Mapped[str] = mapped_column(
String(36),
primary_key=True,
default=new_uuid,
)
entry_id: Mapped[str] = mapped_column(
ForeignKey(
"risk_sanctions_entries.id",
ondelete="CASCADE",
),
nullable=False,
index=True,
)
street: Mapped[str | None] = mapped_column(
String(1000),
nullable=True,
)
city: Mapped[str | None] = mapped_column(
String(500),
nullable=True,
)
region: Mapped[str | None] = mapped_column(
String(500),
nullable=True,
)
postal_code: Mapped[str | None] = mapped_column(
String(100),
nullable=True,
)
country: Mapped[str | None] = mapped_column(
String(255),
nullable=True,
)
note: Mapped[str | None] = mapped_column(
Text,
nullable=True,
)
entry: Mapped[RiskSanctionsEntry] = relationship(
back_populates="addresses"
)
class RiskScreeningSubjectSnapshot(Base, TimestampMixin):
__tablename__ = "risk_screening_subject_snapshots"
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,
)
subject_ref: Mapped[str | None] = mapped_column(
String(500),
nullable=True,
index=True,
)
subject_type: Mapped[str] = mapped_column(
String(30),
nullable=False,
)
primary_name: Mapped[str | None] = mapped_column(
String(1000),
nullable=True,
)
normalized_name: Mapped[str | None] = mapped_column(
String(1000),
nullable=True,
index=True,
)
aliases: Mapped[list[str]] = mapped_column(
JSON,
default=list,
nullable=False,
)
identifiers: Mapped[list[dict[str, str]]] = mapped_column(
JSON,
default=list,
nullable=False,
)
dates: Mapped[list[str]] = mapped_column(
JSON,
default=list,
nullable=False,
)
addresses: Mapped[list[dict[str, str]]] = mapped_column(
JSON,
default=list,
nullable=False,
)
fingerprint: Mapped[str] = mapped_column(
String(64),
nullable=False,
index=True,
)
submitted_by: Mapped[str | None] = mapped_column(
String(255),
nullable=True,
)
class RiskScreeningRun(Base, TimestampMixin):
__tablename__ = "risk_screening_runs"
__table_args__ = (
UniqueConstraint(
"tenant_id",
"idempotency_key",
name="uq_risk_screening_run_idempotency",
),
Index(
"ix_risk_screening_run_status",
"tenant_id",
"outcome",
"created_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,
)
subject_snapshot_id: Mapped[str] = mapped_column(
ForeignKey(
"risk_screening_subject_snapshots.id",
ondelete="RESTRICT",
),
nullable=False,
index=True,
)
list_snapshot_id: Mapped[str] = mapped_column(
ForeignKey(
"risk_sanctions_list_snapshots.id",
ondelete="RESTRICT",
),
nullable=False,
index=True,
)
idempotency_key: Mapped[str] = mapped_column(
String(255),
nullable=False,
)
request_hash: Mapped[str] = mapped_column(
String(64),
nullable=False,
)
matcher_version: Mapped[str] = mapped_column(
String(100),
nullable=False,
)
normalization_version: Mapped[str] = mapped_column(
String(100),
nullable=False,
)
policy_version: Mapped[str] = mapped_column(
String(100),
nullable=False,
)
policy_snapshot: Mapped[dict[str, Any]] = mapped_column(
JSON,
default=dict,
nullable=False,
)
status: Mapped[str] = mapped_column(
String(30),
nullable=False,
index=True,
)
outcome: Mapped[str] = mapped_column(
String(30),
nullable=False,
index=True,
)
candidate_count: Mapped[int] = mapped_column(
Integer,
default=0,
nullable=False,
)
started_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
nullable=False,
)
completed_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True),
nullable=True,
)
created_by: Mapped[str | None] = mapped_column(
String(255),
nullable=True,
index=True,
)
subject_snapshot: Mapped[RiskScreeningSubjectSnapshot] = relationship()
list_snapshot: Mapped[RiskSanctionsListSnapshot] = relationship()
candidates: Mapped[list["RiskScreeningCandidate"]] = relationship(
back_populates="run",
cascade="all, delete-orphan",
)
class RiskScreeningCandidate(Base, TimestampMixin):
__tablename__ = "risk_screening_candidates"
__table_args__ = (
UniqueConstraint(
"run_id",
"entry_id",
name="uq_risk_screening_candidate_entry",
),
Index(
"ix_risk_screening_candidate_queue",
"tenant_id",
"review_status",
"score",
),
)
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,
)
run_id: Mapped[str] = mapped_column(
ForeignKey(
"risk_screening_runs.id",
ondelete="CASCADE",
),
nullable=False,
index=True,
)
entry_id: Mapped[str] = mapped_column(
ForeignKey(
"risk_sanctions_entries.id",
ondelete="RESTRICT",
),
nullable=False,
index=True,
)
score: Mapped[int] = mapped_column(
Integer,
nullable=False,
index=True,
)
match_kind: Mapped[str] = mapped_column(
String(50),
nullable=False,
index=True,
)
evidence: Mapped[list[dict[str, Any]]] = mapped_column(
JSON,
default=list,
nullable=False,
)
review_status: Mapped[str] = mapped_column(
String(30),
default="pending",
nullable=False,
index=True,
)
current_disposition_id: Mapped[str | None] = mapped_column(
String(36),
nullable=True,
)
run: Mapped[RiskScreeningRun] = relationship(
back_populates="candidates"
)
entry: Mapped[RiskSanctionsEntry] = relationship()
class RiskScreeningDisposition(Base):
__tablename__ = "risk_screening_dispositions"
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,
)
candidate_id: Mapped[str] = mapped_column(
ForeignKey(
"risk_screening_candidates.id",
ondelete="RESTRICT",
),
nullable=False,
index=True,
)
decision: Mapped[str] = mapped_column(
String(30),
nullable=False,
index=True,
)
reason: Mapped[str] = mapped_column(
Text,
nullable=False,
)
evidence_refs: Mapped[list[str]] = mapped_column(
JSON,
default=list,
nullable=False,
)
scope: Mapped[str] = mapped_column(
String(30),
default="candidate",
nullable=False,
)
expires_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True),
nullable=True,
)
review_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True),
nullable=True,
)
actor_account_id: Mapped[str | None] = mapped_column(
String(36),
nullable=True,
index=True,
)
actor_membership_id: Mapped[str | None] = mapped_column(
String(36),
nullable=True,
)
actor_authority: Mapped[dict[str, Any]] = mapped_column(
JSON,
default=dict,
nullable=False,
)
separation_status: Mapped[str] = mapped_column(
String(30),
nullable=False,
)
override_reason: Mapped[str | None] = mapped_column(
Text,
nullable=True,
)
supersedes_id: Mapped[str | None] = mapped_column(
ForeignKey(
"risk_screening_dispositions.id",
ondelete="RESTRICT",
),
nullable=True,
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
nullable=False,
index=True,
)
class RiskScreeningException(Base, TimestampMixin):
__tablename__ = "risk_screening_exceptions"
__table_args__ = (
Index(
"ix_risk_screening_exception_active",
"tenant_id",
"subject_fingerprint",
"source_entry_ref",
"status",
),
)
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,
)
subject_fingerprint: Mapped[str] = mapped_column(
String(64),
nullable=False,
index=True,
)
source_entry_ref: Mapped[str] = mapped_column(
String(500),
nullable=False,
index=True,
)
scope: Mapped[str] = mapped_column(
String(30),
nullable=False,
)
status: Mapped[str] = mapped_column(
String(30),
default="active",
nullable=False,
index=True,
)
reason: Mapped[str] = mapped_column(
Text,
nullable=False,
)
evidence_refs: Mapped[list[str]] = mapped_column(
JSON,
default=list,
nullable=False,
)
starts_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
nullable=False,
)
expires_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
nullable=False,
index=True,
)
review_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True),
nullable=True,
)
originating_disposition_id: Mapped[str] = mapped_column(
ForeignKey(
"risk_screening_dispositions.id",
ondelete="RESTRICT",
),
nullable=False,
)
created_by: Mapped[str | None] = mapped_column(
String(36),
nullable=True,
)
__all__ = [
"RiskSanctionsAddress",
"RiskSanctionsAlias",
"RiskSanctionsDate",
"RiskSanctionsEntry",
"RiskSanctionsIdentifier",
"RiskSanctionsListSnapshot",
"RiskScreeningCandidate",
"RiskScreeningDisposition",
"RiskScreeningException",
"RiskScreeningRun",
"RiskScreeningSubjectSnapshot",
"new_uuid",
]

316
src/govoplan_risk_compliance/backend/manifest.py
View File

@@ -1,14 +1,56 @@
from __future__ import annotations
from govoplan_core.core.access import CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER
from govoplan_core.core.modules import DocumentationLink, DocumentationTopic, ModuleManifest, PermissionDefinition, RoleTemplate
from pathlib import Path
MODULE_ID = "risk-compliance"
from govoplan_core.core.access import (
CAPABILITY_AUTH_PERMISSION_EVALUATOR,
CAPABILITY_AUTH_PRINCIPAL_RESOLVER,
)
from govoplan_core.core.module_guards import (
drop_table_retirement_provider,
persistent_table_uninstall_guard,
)
from govoplan_core.core.modules import (
DocumentationLink,
DocumentationTopic,
FrontendModule,
FrontendRoute,
MigrationSpec,
ModuleInterfaceRequirement,
ModuleManifest,
NavItem,
PermissionDefinition,
RoleTemplate,
)
from govoplan_core.core.views import ViewSurface
from govoplan_core.db.base import Base
from govoplan_risk_compliance.backend.db.models import (
RiskSanctionsAddress,
RiskSanctionsAlias,
RiskSanctionsDate,
RiskSanctionsEntry,
RiskSanctionsIdentifier,
RiskSanctionsListSnapshot,
RiskScreeningCandidate,
RiskScreeningDisposition,
RiskScreeningException,
RiskScreeningRun,
RiskScreeningSubjectSnapshot,
)
from govoplan_risk_compliance.backend.permissions import (
ADMIN_SCOPE,
READ_SCOPE,
SANCTIONS_ADMIN_SCOPE,
SANCTIONS_READ_SCOPE,
SANCTIONS_REVIEW_SCOPE,
SANCTIONS_SCREEN_SCOPE,
WRITE_SCOPE,
)
MODULE_ID = "risk_compliance"
MODULE_NAME = "Risk Compliance"
MODULE_VERSION = "0.1.8"
READ_SCOPE = "risk-compliance:workspace:read"
WRITE_SCOPE = "risk-compliance:workspace:write"
ADMIN_SCOPE = "risk-compliance:workspace:admin"
OPTIONAL_DEPENDENCIES = (
"audit",
"policy",
@@ -17,10 +59,28 @@ OPTIONAL_DEPENDENCIES = (
"files",
"tasks",
"notifications",
"connectors",
)
_PERSISTENT_MODELS = (
RiskScreeningException,
RiskScreeningDisposition,
RiskScreeningCandidate,
RiskScreeningRun,
RiskScreeningSubjectSnapshot,
RiskSanctionsAddress,
RiskSanctionsDate,
RiskSanctionsIdentifier,
RiskSanctionsAlias,
RiskSanctionsEntry,
RiskSanctionsListSnapshot,
)
def _permission(scope: str, label: str, description: str) -> PermissionDefinition:
def _permission(
scope: str,
label: str,
description: str,
) -> PermissionDefinition:
module_id, resource, action = scope.split(":", 2)
return PermissionDefinition(
scope=scope,
@@ -35,52 +95,159 @@ def _permission(scope: str, label: str, description: str) -> PermissionDefinitio
PERMISSIONS = (
_permission(READ_SCOPE, "View risk compliance workspace", "Read risk compliance records, configuration, and workflow context."),
_permission(WRITE_SCOPE, "Manage risk compliance workspace", "Create and update risk compliance records and workflow state."),
_permission(ADMIN_SCOPE, "Administer risk compliance workspace", "Configure risk compliance policies, templates, and tenant-level administration."),
_permission(
READ_SCOPE,
"View risk compliance workspace",
"Read risk compliance records, configuration, and workflow context.",
),
_permission(
WRITE_SCOPE,
"Manage risk compliance workspace",
"Create and update risk compliance records and workflow state.",
),
_permission(
ADMIN_SCOPE,
"Administer risk compliance workspace",
"Configure risk compliance policies, templates, and tenant administration.",
),
_permission(
SANCTIONS_READ_SCOPE,
"View sanctions screening evidence",
"Read list snapshots, screening runs, and candidate evidence.",
),
_permission(
SANCTIONS_SCREEN_SCOPE,
"Run sanctions screening",
"Submit subjects for deterministic screening against an immutable list.",
),
_permission(
SANCTIONS_REVIEW_SCOPE,
"Review sanctions candidates",
"Record evidence-backed candidate dispositions.",
),
_permission(
SANCTIONS_ADMIN_SCOPE,
"Administer sanctions screening",
"Import source snapshots, configure policy, and authorize review overrides.",
),
)
ROLE_TEMPLATES = (
RoleTemplate(
slug="risk_compliance_manager",
name="Risk Compliance manager",
description="Manage risk compliance records and workflow state.",
permissions=(READ_SCOPE, WRITE_SCOPE),
description=(
"Manage compliance workflows and administer sanctions screening."
),
permissions=(
READ_SCOPE,
WRITE_SCOPE,
SANCTIONS_READ_SCOPE,
SANCTIONS_SCREEN_SCOPE,
SANCTIONS_REVIEW_SCOPE,
SANCTIONS_ADMIN_SCOPE,
),
),
RoleTemplate(
slug="risk_compliance_reviewer",
name="Risk Compliance reviewer",
description=(
"Run screenings and independently review potential matches."
),
permissions=(
READ_SCOPE,
SANCTIONS_READ_SCOPE,
SANCTIONS_SCREEN_SCOPE,
SANCTIONS_REVIEW_SCOPE,
),
),
RoleTemplate(
slug="risk_compliance_viewer",
name="Risk Compliance viewer",
description="Read risk compliance records and workflow context.",
permissions=(READ_SCOPE,),
description="Read risk compliance records and screening evidence.",
permissions=(READ_SCOPE, SANCTIONS_READ_SCOPE),
),
)
def _route_factory(_context):
from govoplan_risk_compliance.backend.router import router
return router
def _tenant_summary(session, tenant_id: str) -> dict[str, int]:
return {
"risk_sanctions_list_snapshots": (
session.query(RiskSanctionsListSnapshot)
.filter(
RiskSanctionsListSnapshot.tenant_id == tenant_id
)
.count()
),
"risk_screening_runs": (
session.query(RiskScreeningRun)
.filter(RiskScreeningRun.tenant_id == tenant_id)
.count()
),
"risk_pending_screening_candidates": (
session.query(RiskScreeningCandidate)
.filter(
RiskScreeningCandidate.tenant_id == tenant_id,
RiskScreeningCandidate.review_status.in_(
("pending", "exception_review")
),
)
.count()
),
}
DOCUMENTATION = (
DocumentationTopic(
id=f"{MODULE_ID}.module-boundary",
title=f"{MODULE_NAME} module boundary",
summary="Risk and compliance workflows for data protection incidents, DPIAs, compliance controls, audit measures, risk registers, and internal-control evidence.",
summary=(
"Risk and compliance workflows own legal evaluation, immutable "
"screening evidence, review, and dispositions."
),
body=(
"This repository is currently a platform module seed. It registers the domain boundary, "
"permission surface, role templates, and documentation metadata before runtime APIs, "
"database models, migrations, and WebUI routes are introduced."
"Connectors may acquire source evidence, but Risk Compliance "
"owns immutable normalized sanctions lists, version-pinned "
"screening, candidate review, and legal dispositions. Fuzzy "
"matching only creates candidates and never confirms a match."
),
layer="available",
documentation_types=("admin",),
audience=("operator", "module_admin", "product_owner"),
documentation_types=("admin", "user"),
audience=(
"operator",
"module_admin",
"compliance_reviewer",
),
order=100,
related_modules=OPTIONAL_DEPENDENCIES,
links=(
DocumentationLink(
label="Repository domain boundary",
href="govoplan-risk-compliance/docs/RISK_COMPLIANCE_DOMAIN_BOUNDARY.md",
href=(
"govoplan-risk-compliance/"
"docs/RISK_COMPLIANCE_DOMAIN_BOUNDARY.md"
),
kind="repository",
),
),
metadata={
"seed": True,
"domain_objects": ['risk registers', 'compliance controls', 'DPIA records', 'data protection incident records', 'audit measures', 'internal-control evidence'],
"first_slice": "Define risk register, control, evidence, DPIA, incident, measure, and review-cycle concepts.",
"domain_objects": [
"sanctions list snapshots",
"screening runs",
"candidate evidence",
"review dispositions",
"time-bounded exceptions",
],
"privacy": (
"Queue and audit summaries contain stable references and "
"minimal subject data."
),
},
),
)
@@ -91,12 +258,111 @@ manifest = ModuleManifest(
version=MODULE_VERSION,
dependencies=("access",),
optional_dependencies=OPTIONAL_DEPENDENCIES,
required_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR),
required_capabilities=(
CAPABILITY_AUTH_PRINCIPAL_RESOLVER,
CAPABILITY_AUTH_PERMISSION_EVALUATOR,
),
requires_interfaces=(
ModuleInterfaceRequirement(
name="connectors.sanctions_snapshots",
version_min="1.0.0",
version_max_exclusive="2.0.0",
optional=True,
),
),
permissions=PERMISSIONS,
role_templates=ROLE_TEMPLATES,
route_factory=_route_factory,
frontend=FrontendModule(
module_id=MODULE_ID,
package_name="@govoplan/risk-compliance-webui",
routes=(
FrontendRoute(
path="/risk-compliance",
component="RiskCompliancePage",
required_any=(SANCTIONS_READ_SCOPE,),
order=115,
surface_id="risk_compliance.workspace",
),
),
nav_items=(
NavItem(
path="/risk-compliance",
label="Risk Compliance",
icon="shield-check",
required_any=(SANCTIONS_READ_SCOPE,),
order=115,
surface_id="risk_compliance.navigation",
),
),
view_surfaces=(
ViewSurface(
id="risk_compliance.sanctions.sources",
module_id=MODULE_ID,
kind="section",
label="Sanctions source snapshots",
order=20,
),
ViewSurface(
id="risk_compliance.sanctions.screening",
module_id=MODULE_ID,
kind="section",
label="Sanctions screening",
order=30,
),
ViewSurface(
id="risk_compliance.sanctions.review",
module_id=MODULE_ID,
kind="section",
label="Sanctions review queue",
order=40,
),
),
),
tenant_summary_providers=(_tenant_summary,),
migration_spec=MigrationSpec(
module_id=MODULE_ID,
metadata=Base.metadata,
script_location=str(
Path(__file__).with_name("migrations") / "versions"
),
migration_after=("connectors",),
retirement_supported=True,
retirement_provider=drop_table_retirement_provider(
*_PERSISTENT_MODELS,
label="Risk Compliance",
),
retirement_notes=(
"Destructive retirement removes immutable sanctions list, "
"screening, and review evidence after a database snapshot."
),
),
uninstall_guard_providers=(
persistent_table_uninstall_guard(
*_PERSISTENT_MODELS,
label="Risk Compliance",
),
),
documentation=DOCUMENTATION,
)
def get_manifest() -> ModuleManifest:
return manifest
__all__ = [
"ADMIN_SCOPE",
"MODULE_ID",
"MODULE_VERSION",
"PERMISSIONS",
"READ_SCOPE",
"ROLE_TEMPLATES",
"SANCTIONS_ADMIN_SCOPE",
"SANCTIONS_READ_SCOPE",
"SANCTIONS_REVIEW_SCOPE",
"SANCTIONS_SCREEN_SCOPE",
"WRITE_SCOPE",
"get_manifest",
"manifest",
]

1
src/govoplan_risk_compliance/backend/migrations/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Risk Compliance database migrations."""

1
src/govoplan_risk_compliance/backend/migrations/versions/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Risk Compliance Alembic revisions."""

834
src/govoplan_risk_compliance/backend/migrations/versions/a8b9c0d1e2f3_v0114_sanctions_screening.py Normal file
View File

@@ -0,0 +1,834 @@
"""Add sanctions catalogue, screening, and review evidence.
Revision ID: a8b9c0d1e2f3
Revises:
Create Date: 2026-07-29
"""
from __future__ import annotations
from alembic import op
import sqlalchemy as sa
revision = "a8b9c0d1e2f3"
down_revision = None
branch_labels = None
depends_on = None
def _timestamps() -> tuple[sa.Column, sa.Column]:
return (
sa.Column(
"created_at",
sa.DateTime(timezone=True),
nullable=False,
),
sa.Column(
"updated_at",
sa.DateTime(timezone=True),
nullable=False,
),
)
def upgrade() -> None:
op.create_table(
"risk_sanctions_list_snapshots",
sa.Column("id", sa.String(length=36), nullable=False),
sa.Column("tenant_id", sa.String(length=36), nullable=True),
sa.Column(
"visibility",
sa.String(length=20),
nullable=False,
),
sa.Column(
"connector_snapshot_ref",
sa.String(length=300),
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("sha256", sa.String(length=64), nullable=False),
sa.Column(
"connector_run_id",
sa.String(length=36),
nullable=False,
),
sa.Column(
"raw_evidence_ref",
sa.String(length=300),
nullable=False,
),
sa.Column(
"source_parser_version",
sa.String(length=100),
nullable=False,
),
sa.Column(
"normalization_version",
sa.String(length=100),
nullable=False,
),
sa.Column(
"signature_evidence",
sa.JSON(),
nullable=False,
),
sa.Column("provenance", sa.JSON(), nullable=False),
sa.Column("entry_count", sa.Integer(), nullable=False),
sa.Column("status", sa.String(length=30), nullable=False),
sa.Column(
"imported_by",
sa.String(length=255),
nullable=True,
),
sa.Column(
"imported_at",
sa.DateTime(timezone=True),
nullable=False,
),
*_timestamps(),
sa.PrimaryKeyConstraint(
"id",
name=op.f("pk_risk_sanctions_list_snapshots"),
),
sa.UniqueConstraint(
"tenant_id",
"connector_snapshot_ref",
name="uq_risk_sanctions_connector_snapshot",
),
)
_indexes(
"risk_sanctions_list_snapshots",
(
"tenant_id",
"visibility",
"provider_id",
"jurisdiction",
"list_type",
"source_id",
"source_version",
"acquired_at",
"sha256",
"status",
"imported_by",
),
)
op.create_index(
"ix_risk_sanctions_snapshot_current",
"risk_sanctions_list_snapshots",
["tenant_id", "provider_id", "status", "acquired_at"],
)
op.create_table(
"risk_sanctions_entries",
sa.Column("id", sa.String(length=36), nullable=False),
sa.Column("snapshot_id", sa.String(length=36), nullable=False),
sa.Column(
"source_entry_id",
sa.String(length=255),
nullable=False,
),
sa.Column(
"subject_type",
sa.String(length=30),
nullable=False,
),
sa.Column(
"primary_name",
sa.String(length=1000),
nullable=False,
),
sa.Column(
"normalized_name",
sa.String(length=1000),
nullable=False,
),
sa.Column(
"original_script_name",
sa.String(length=1000),
nullable=True,
),
sa.Column(
"reference_number",
sa.String(length=255),
nullable=True,
),
sa.Column("listed_on", sa.Date(), nullable=True),
sa.Column("programmes", sa.JSON(), nullable=False),
sa.Column("measures", sa.JSON(), nullable=False),
sa.Column(
"raw_evidence_locator",
sa.String(length=500),
nullable=False,
),
sa.Column("details", sa.JSON(), nullable=False),
*_timestamps(),
sa.ForeignKeyConstraint(
["snapshot_id"],
["risk_sanctions_list_snapshots.id"],
name=op.f(
"fk_risk_sanctions_entries_snapshot_id_"
"risk_sanctions_list_snapshots"
),
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint(
"id",
name=op.f("pk_risk_sanctions_entries"),
),
sa.UniqueConstraint(
"snapshot_id",
"source_entry_id",
name="uq_risk_sanctions_entry_source",
),
)
_indexes(
"risk_sanctions_entries",
(
"snapshot_id",
"source_entry_id",
"subject_type",
"normalized_name",
"reference_number",
),
)
op.create_index(
"ix_risk_sanctions_entry_name",
"risk_sanctions_entries",
["snapshot_id", "normalized_name"],
)
op.create_table(
"risk_sanctions_aliases",
sa.Column("id", sa.String(length=36), nullable=False),
sa.Column("entry_id", sa.String(length=36), nullable=False),
sa.Column(
"name",
sa.String(length=1000),
nullable=False,
),
sa.Column(
"normalized_name",
sa.String(length=1000),
nullable=False,
),
sa.Column(
"quality",
sa.String(length=100),
nullable=True,
),
sa.Column(
"alias_type",
sa.String(length=50),
nullable=False,
),
sa.ForeignKeyConstraint(
["entry_id"],
["risk_sanctions_entries.id"],
name=op.f(
"fk_risk_sanctions_aliases_entry_id_"
"risk_sanctions_entries"
),
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint(
"id",
name=op.f("pk_risk_sanctions_aliases"),
),
)
_indexes(
"risk_sanctions_aliases",
("entry_id", "normalized_name"),
)
op.create_index(
"ix_risk_sanctions_alias_name",
"risk_sanctions_aliases",
["entry_id", "normalized_name"],
)
op.create_table(
"risk_sanctions_identifiers",
sa.Column("id", sa.String(length=36), nullable=False),
sa.Column("entry_id", sa.String(length=36), nullable=False),
sa.Column(
"identifier_type",
sa.String(length=100),
nullable=False,
),
sa.Column(
"value",
sa.String(length=1000),
nullable=False,
),
sa.Column(
"normalized_value",
sa.String(length=1000),
nullable=False,
),
sa.Column(
"issuing_country",
sa.String(length=255),
nullable=True,
),
sa.Column("note", sa.Text(), nullable=True),
sa.ForeignKeyConstraint(
["entry_id"],
["risk_sanctions_entries.id"],
name=op.f(
"fk_risk_sanctions_identifiers_entry_id_"
"risk_sanctions_entries"
),
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint(
"id",
name=op.f("pk_risk_sanctions_identifiers"),
),
)
_indexes(
"risk_sanctions_identifiers",
("entry_id", "normalized_value"),
)
op.create_index(
"ix_risk_sanctions_identifier_value",
"risk_sanctions_identifiers",
["entry_id", "normalized_value"],
)
op.create_table(
"risk_sanctions_dates",
sa.Column("id", sa.String(length=36), nullable=False),
sa.Column("entry_id", sa.String(length=36), nullable=False),
sa.Column(
"date_type",
sa.String(length=50),
nullable=False,
),
sa.Column("value", sa.String(length=100), nullable=False),
sa.Column(
"precision",
sa.String(length=30),
nullable=False,
),
sa.ForeignKeyConstraint(
["entry_id"],
["risk_sanctions_entries.id"],
name=op.f(
"fk_risk_sanctions_dates_entry_id_"
"risk_sanctions_entries"
),
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint(
"id",
name=op.f("pk_risk_sanctions_dates"),
),
)
_indexes("risk_sanctions_dates", ("entry_id",))
op.create_table(
"risk_sanctions_addresses",
sa.Column("id", sa.String(length=36), nullable=False),
sa.Column("entry_id", sa.String(length=36), nullable=False),
sa.Column(
"street",
sa.String(length=1000),
nullable=True,
),
sa.Column("city", sa.String(length=500), nullable=True),
sa.Column("region", sa.String(length=500), nullable=True),
sa.Column(
"postal_code",
sa.String(length=100),
nullable=True,
),
sa.Column(
"country",
sa.String(length=255),
nullable=True,
),
sa.Column("note", sa.Text(), nullable=True),
sa.ForeignKeyConstraint(
["entry_id"],
["risk_sanctions_entries.id"],
name=op.f(
"fk_risk_sanctions_addresses_entry_id_"
"risk_sanctions_entries"
),
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint(
"id",
name=op.f("pk_risk_sanctions_addresses"),
),
)
_indexes("risk_sanctions_addresses", ("entry_id",))
op.create_table(
"risk_screening_subject_snapshots",
sa.Column("id", sa.String(length=36), nullable=False),
sa.Column("tenant_id", sa.String(length=36), nullable=False),
sa.Column(
"subject_ref",
sa.String(length=500),
nullable=True,
),
sa.Column(
"subject_type",
sa.String(length=30),
nullable=False,
),
sa.Column(
"primary_name",
sa.String(length=1000),
nullable=True,
),
sa.Column(
"normalized_name",
sa.String(length=1000),
nullable=True,
),
sa.Column("aliases", sa.JSON(), nullable=False),
sa.Column("identifiers", sa.JSON(), nullable=False),
sa.Column("dates", sa.JSON(), nullable=False),
sa.Column("addresses", sa.JSON(), nullable=False),
sa.Column(
"fingerprint",
sa.String(length=64),
nullable=False,
),
sa.Column(
"submitted_by",
sa.String(length=255),
nullable=True,
),
*_timestamps(),
sa.PrimaryKeyConstraint(
"id",
name=op.f("pk_risk_screening_subject_snapshots"),
),
)
_indexes(
"risk_screening_subject_snapshots",
(
"tenant_id",
"subject_ref",
"normalized_name",
"fingerprint",
),
)
op.create_table(
"risk_screening_runs",
sa.Column("id", sa.String(length=36), nullable=False),
sa.Column("tenant_id", sa.String(length=36), nullable=False),
sa.Column(
"subject_snapshot_id",
sa.String(length=36),
nullable=False,
),
sa.Column(
"list_snapshot_id",
sa.String(length=36),
nullable=False,
),
sa.Column(
"idempotency_key",
sa.String(length=255),
nullable=False,
),
sa.Column(
"request_hash",
sa.String(length=64),
nullable=False,
),
sa.Column(
"matcher_version",
sa.String(length=100),
nullable=False,
),
sa.Column(
"normalization_version",
sa.String(length=100),
nullable=False,
),
sa.Column(
"policy_version",
sa.String(length=100),
nullable=False,
),
sa.Column("policy_snapshot", sa.JSON(), nullable=False),
sa.Column("status", sa.String(length=30), nullable=False),
sa.Column("outcome", sa.String(length=30), nullable=False),
sa.Column("candidate_count", sa.Integer(), nullable=False),
sa.Column(
"started_at",
sa.DateTime(timezone=True),
nullable=False,
),
sa.Column(
"completed_at",
sa.DateTime(timezone=True),
nullable=True,
),
sa.Column(
"created_by",
sa.String(length=255),
nullable=True,
),
*_timestamps(),
sa.ForeignKeyConstraint(
["list_snapshot_id"],
["risk_sanctions_list_snapshots.id"],
name=op.f(
"fk_risk_screening_runs_list_snapshot_id_"
"risk_sanctions_list_snapshots"
),
ondelete="RESTRICT",
),
sa.ForeignKeyConstraint(
["subject_snapshot_id"],
["risk_screening_subject_snapshots.id"],
name=op.f(
"fk_risk_screening_runs_subject_snapshot_id_"
"risk_screening_subject_snapshots"
),
ondelete="RESTRICT",
),
sa.PrimaryKeyConstraint(
"id",
name=op.f("pk_risk_screening_runs"),
),
sa.UniqueConstraint(
"tenant_id",
"idempotency_key",
name="uq_risk_screening_run_idempotency",
),
)
_indexes(
"risk_screening_runs",
(
"tenant_id",
"subject_snapshot_id",
"list_snapshot_id",
"status",
"outcome",
"created_by",
),
)
op.create_index(
"ix_risk_screening_run_status",
"risk_screening_runs",
["tenant_id", "outcome", "created_at"],
)
op.create_table(
"risk_screening_candidates",
sa.Column("id", sa.String(length=36), nullable=False),
sa.Column("tenant_id", sa.String(length=36), nullable=False),
sa.Column("run_id", sa.String(length=36), nullable=False),
sa.Column("entry_id", sa.String(length=36), nullable=False),
sa.Column("score", sa.Integer(), nullable=False),
sa.Column(
"match_kind",
sa.String(length=50),
nullable=False,
),
sa.Column("evidence", sa.JSON(), nullable=False),
sa.Column(
"review_status",
sa.String(length=30),
nullable=False,
),
sa.Column(
"current_disposition_id",
sa.String(length=36),
nullable=True,
),
*_timestamps(),
sa.ForeignKeyConstraint(
["entry_id"],
["risk_sanctions_entries.id"],
name=op.f(
"fk_risk_screening_candidates_entry_id_"
"risk_sanctions_entries"
),
ondelete="RESTRICT",
),
sa.ForeignKeyConstraint(
["run_id"],
["risk_screening_runs.id"],
name=op.f(
"fk_risk_screening_candidates_run_id_"
"risk_screening_runs"
),
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint(
"id",
name=op.f("pk_risk_screening_candidates"),
),
sa.UniqueConstraint(
"run_id",
"entry_id",
name="uq_risk_screening_candidate_entry",
),
)
_indexes(
"risk_screening_candidates",
(
"tenant_id",
"run_id",
"entry_id",
"score",
"match_kind",
"review_status",
),
)
op.create_index(
"ix_risk_screening_candidate_queue",
"risk_screening_candidates",
["tenant_id", "review_status", "score"],
)
op.create_table(
"risk_screening_dispositions",
sa.Column("id", sa.String(length=36), nullable=False),
sa.Column("tenant_id", sa.String(length=36), nullable=False),
sa.Column(
"candidate_id",
sa.String(length=36),
nullable=False,
),
sa.Column(
"decision",
sa.String(length=30),
nullable=False,
),
sa.Column("reason", sa.Text(), nullable=False),
sa.Column("evidence_refs", sa.JSON(), nullable=False),
sa.Column("scope", sa.String(length=30), nullable=False),
sa.Column(
"expires_at",
sa.DateTime(timezone=True),
nullable=True,
),
sa.Column(
"review_at",
sa.DateTime(timezone=True),
nullable=True,
),
sa.Column(
"actor_account_id",
sa.String(length=36),
nullable=True,
),
sa.Column(
"actor_membership_id",
sa.String(length=36),
nullable=True,
),
sa.Column("actor_authority", sa.JSON(), nullable=False),
sa.Column(
"separation_status",
sa.String(length=30),
nullable=False,
),
sa.Column("override_reason", sa.Text(), nullable=True),
sa.Column(
"supersedes_id",
sa.String(length=36),
nullable=True,
),
sa.Column(
"created_at",
sa.DateTime(timezone=True),
nullable=False,
),
sa.ForeignKeyConstraint(
["candidate_id"],
["risk_screening_candidates.id"],
name=op.f(
"fk_risk_screening_dispositions_candidate_id_"
"risk_screening_candidates"
),
ondelete="RESTRICT",
),
sa.ForeignKeyConstraint(
["supersedes_id"],
["risk_screening_dispositions.id"],
name=op.f(
"fk_risk_screening_dispositions_supersedes_id_"
"risk_screening_dispositions"
),
ondelete="RESTRICT",
),
sa.PrimaryKeyConstraint(
"id",
name=op.f("pk_risk_screening_dispositions"),
),
)
_indexes(
"risk_screening_dispositions",
(
"tenant_id",
"candidate_id",
"decision",
"actor_account_id",
"created_at",
),
)
op.create_table(
"risk_screening_exceptions",
sa.Column("id", sa.String(length=36), nullable=False),
sa.Column("tenant_id", sa.String(length=36), nullable=False),
sa.Column(
"subject_fingerprint",
sa.String(length=64),
nullable=False,
),
sa.Column(
"source_entry_ref",
sa.String(length=500),
nullable=False,
),
sa.Column("scope", sa.String(length=30), nullable=False),
sa.Column("status", sa.String(length=30), nullable=False),
sa.Column("reason", sa.Text(), nullable=False),
sa.Column("evidence_refs", sa.JSON(), nullable=False),
sa.Column(
"starts_at",
sa.DateTime(timezone=True),
nullable=False,
),
sa.Column(
"expires_at",
sa.DateTime(timezone=True),
nullable=False,
),
sa.Column(
"review_at",
sa.DateTime(timezone=True),
nullable=True,
),
sa.Column(
"originating_disposition_id",
sa.String(length=36),
nullable=False,
),
sa.Column(
"created_by",
sa.String(length=36),
nullable=True,
),
*_timestamps(),
sa.ForeignKeyConstraint(
["originating_disposition_id"],
["risk_screening_dispositions.id"],
name=op.f(
"fk_risk_screening_exceptions_"
"originating_disposition_id_"
"risk_screening_dispositions"
),
ondelete="RESTRICT",
),
sa.PrimaryKeyConstraint(
"id",
name=op.f("pk_risk_screening_exceptions"),
),
)
_indexes(
"risk_screening_exceptions",
(
"tenant_id",
"subject_fingerprint",
"source_entry_ref",
"status",
"expires_at",
),
)
op.create_index(
"ix_risk_screening_exception_active",
"risk_screening_exceptions",
[
"tenant_id",
"subject_fingerprint",
"source_entry_ref",
"status",
],
)
def downgrade() -> None:
for table_name in (
"risk_screening_exceptions",
"risk_screening_dispositions",
"risk_screening_candidates",
"risk_screening_runs",
"risk_screening_subject_snapshots",
"risk_sanctions_addresses",
"risk_sanctions_dates",
"risk_sanctions_identifiers",
"risk_sanctions_aliases",
"risk_sanctions_entries",
"risk_sanctions_list_snapshots",
):
op.drop_table(table_name)
def _indexes(
table_name: str,
columns: tuple[str, ...],
) -> None:
for column in columns:
op.create_index(
op.f(f"ix_{table_name}_{column}"),
table_name,
[column],
)

54
src/govoplan_risk_compliance/backend/normalization.py Normal file
View File

@@ -0,0 +1,54 @@
from __future__ import annotations
import hashlib
import json
import re
import unicodedata
NORMALIZATION_VERSION = "sanctions-normalization-v1"
_NON_ALNUM = re.compile(r"[^\w]+", re.UNICODE)
_WHITESPACE = re.compile(r"\s+")
def normalize_name(value: str | None) -> str:
clean = unicodedata.normalize("NFKD", str(value or ""))
without_marks = "".join(
character
for character in clean
if not unicodedata.combining(character)
)
folded = without_marks.casefold()
return _WHITESPACE.sub(
" ",
_NON_ALNUM.sub(" ", folded).replace("_", " "),
).strip()
def normalize_identifier(value: str | None) -> str:
return "".join(
character
for character in unicodedata.normalize(
"NFKC",
str(value or ""),
).casefold()
if character.isalnum()
)
def subject_fingerprint(payload: dict[str, object]) -> str:
encoded = json.dumps(
payload,
sort_keys=True,
separators=(",", ":"),
ensure_ascii=False,
).encode("utf-8")
return hashlib.sha256(encoded).hexdigest()
__all__ = [
"NORMALIZATION_VERSION",
"normalize_identifier",
"normalize_name",
"subject_fingerprint",
]

18
src/govoplan_risk_compliance/backend/permissions.py Normal file
View File

@@ -0,0 +1,18 @@
READ_SCOPE = "risk_compliance:workspace:read"
WRITE_SCOPE = "risk_compliance:workspace:write"
ADMIN_SCOPE = "risk_compliance:workspace:admin"
SANCTIONS_READ_SCOPE = "risk_compliance:sanctions:read"
SANCTIONS_SCREEN_SCOPE = "risk_compliance:sanctions:screen"
SANCTIONS_REVIEW_SCOPE = "risk_compliance:sanctions:review"
SANCTIONS_ADMIN_SCOPE = "risk_compliance:sanctions:admin"
__all__ = [
"ADMIN_SCOPE",
"READ_SCOPE",
"SANCTIONS_ADMIN_SCOPE",
"SANCTIONS_READ_SCOPE",
"SANCTIONS_REVIEW_SCOPE",
"SANCTIONS_SCREEN_SCOPE",
"WRITE_SCOPE",
]

304
src/govoplan_risk_compliance/backend/review.py Normal file
View File

@@ -0,0 +1,304 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Literal
from sqlalchemy import select
from sqlalchemy.orm import Session, selectinload
from govoplan_core.auth import ApiPrincipal, has_scope
from govoplan_core.db.base import utcnow
from govoplan_risk_compliance.backend.db.models import (
RiskSanctionsEntry,
RiskScreeningCandidate,
RiskScreeningDisposition,
RiskScreeningException,
RiskScreeningRun,
)
from govoplan_risk_compliance.backend.permissions import (
SANCTIONS_ADMIN_SCOPE,
SANCTIONS_REVIEW_SCOPE,
)
from govoplan_risk_compliance.backend.sanctions_catalog import (
RiskSanctionsAccessError,
RiskSanctionsConflictError,
RiskSanctionsNotFoundError,
)
DispositionDecision = Literal[
"true_match",
"false_positive",
"needs_information",
"escalated",
]
@dataclass(frozen=True, slots=True)
class DispositionInput:
decision: DispositionDecision
reason: str
evidence_refs: tuple[str, ...] = ()
exception_scope: Literal["candidate", "subject_entry"] = "candidate"
expires_at: datetime | None = None
review_at: datetime | None = None
override_reason: str | None = None
def __post_init__(self) -> None:
if self.decision not in {
"true_match",
"false_positive",
"needs_information",
"escalated",
}:
raise RiskSanctionsConflictError(
"Unsupported screening disposition."
)
if len(self.reason.strip()) < 3:
raise RiskSanctionsConflictError(
"A disposition reason is required."
)
if (
self.exception_scope == "subject_entry"
and self.decision != "false_positive"
):
raise RiskSanctionsConflictError(
"Only false-positive decisions can create a reusable exception."
)
if self.exception_scope == "subject_entry":
if self.expires_at is None:
raise RiskSanctionsConflictError(
"Reusable exceptions require an expiry time."
)
if _aware(self.expires_at) <= _aware(utcnow()):
raise RiskSanctionsConflictError(
"Reusable exception expiry must be in the future."
)
def list_review_queue(
session: Session,
principal: ApiPrincipal,
*,
status: str = "pending",
limit: int = 100,
) -> tuple[RiskScreeningCandidate, ...]:
_require_review(principal)
query = (
select(RiskScreeningCandidate)
.where(RiskScreeningCandidate.tenant_id == principal.tenant_id)
.options(
selectinload(RiskScreeningCandidate.run).options(
selectinload(RiskScreeningRun.subject_snapshot),
selectinload(RiskScreeningRun.list_snapshot),
),
selectinload(RiskScreeningCandidate.entry).options(
selectinload(RiskSanctionsEntry.aliases),
selectinload(RiskSanctionsEntry.identifiers),
selectinload(RiskSanctionsEntry.dates),
selectinload(RiskSanctionsEntry.addresses),
),
)
.order_by(
RiskScreeningCandidate.score.desc(),
RiskScreeningCandidate.created_at.asc(),
)
.limit(max(1, min(limit, 500)))
)
if status != "all":
requested = (
("pending", "exception_review")
if status == "pending"
else (status,)
)
query = query.where(
RiskScreeningCandidate.review_status.in_(requested)
)
return tuple(session.scalars(query))
def get_candidate(
session: Session,
principal: ApiPrincipal,
*,
candidate_id: str,
) -> RiskScreeningCandidate:
_require_review(principal)
item = session.scalar(
select(RiskScreeningCandidate)
.where(
RiskScreeningCandidate.id == candidate_id,
RiskScreeningCandidate.tenant_id == principal.tenant_id,
)
.options(
selectinload(RiskScreeningCandidate.run).options(
selectinload(RiskScreeningRun.subject_snapshot),
selectinload(RiskScreeningRun.list_snapshot),
),
selectinload(RiskScreeningCandidate.entry).options(
selectinload(RiskSanctionsEntry.aliases),
selectinload(RiskSanctionsEntry.identifiers),
selectinload(RiskSanctionsEntry.dates),
selectinload(RiskSanctionsEntry.addresses),
),
)
)
if item is None:
raise RiskSanctionsNotFoundError(
"Screening candidate was not found."
)
return item
def record_disposition(
session: Session,
principal: ApiPrincipal,
*,
candidate_id: str,
disposition: DispositionInput,
) -> tuple[RiskScreeningCandidate, RiskScreeningDisposition]:
candidate = get_candidate(
session,
principal,
candidate_id=candidate_id,
)
actor_id = principal.account_id
same_actor = bool(
candidate.run.created_by
and candidate.run.created_by
in {
principal.account_id,
principal.membership_id,
principal.identity_id,
}
)
separation_status = "independent"
clean_override = (disposition.override_reason or "").strip()
if same_actor:
if not has_scope(principal, SANCTIONS_ADMIN_SCOPE):
raise RiskSanctionsAccessError(
"The screening submitter cannot review the same candidate."
)
if len(clean_override) < 3:
raise RiskSanctionsConflictError(
"An administrator override reason is required when "
"reviewing your own screening."
)
separation_status = "administrator_override"
now = utcnow()
item = RiskScreeningDisposition(
tenant_id=principal.tenant_id,
candidate_id=candidate.id,
decision=disposition.decision,
reason=disposition.reason.strip(),
evidence_refs=[
value.strip()[:500]
for value in disposition.evidence_refs
if value.strip()
][:100],
scope=disposition.exception_scope,
expires_at=disposition.expires_at,
review_at=disposition.review_at,
actor_account_id=actor_id,
actor_membership_id=principal.membership_id,
actor_authority={
"required_scope": SANCTIONS_REVIEW_SCOPE,
"admin": has_scope(principal, SANCTIONS_ADMIN_SCOPE),
"auth_method": principal.auth_method,
"acting_for_account_id": principal.acting_for_account_id,
},
separation_status=separation_status,
override_reason=clean_override or None,
supersedes_id=candidate.current_disposition_id,
created_at=now,
)
session.add(item)
session.flush()
candidate.current_disposition_id = item.id
candidate.review_status = {
"true_match": "confirmed",
"false_positive": "false_positive",
"needs_information": "needs_information",
"escalated": "escalated",
}[disposition.decision]
if disposition.exception_scope == "subject_entry":
session.add(
RiskScreeningException(
tenant_id=principal.tenant_id,
subject_fingerprint=(
candidate.run.subject_snapshot.fingerprint
),
source_entry_ref=candidate.entry.source_entry_id,
scope="subject_entry",
status="active",
reason=disposition.reason.strip(),
evidence_refs=list(item.evidence_refs),
starts_at=now,
expires_at=disposition.expires_at,
review_at=disposition.review_at,
originating_disposition_id=item.id,
created_by=actor_id,
)
)
session.flush()
return candidate, item
def list_candidate_dispositions(
session: Session,
principal: ApiPrincipal,
*,
candidate_id: str,
) -> tuple[RiskScreeningDisposition, ...]:
candidate = get_candidate(
session,
principal,
candidate_id=candidate_id,
)
return tuple(
session.scalars(
select(RiskScreeningDisposition)
.where(
RiskScreeningDisposition.candidate_id == candidate.id,
RiskScreeningDisposition.tenant_id
== principal.tenant_id,
)
.order_by(
RiskScreeningDisposition.created_at.asc(),
RiskScreeningDisposition.id.asc(),
)
)
)
def _require_review(principal: ApiPrincipal) -> None:
if not isinstance(principal, ApiPrincipal):
raise RiskSanctionsAccessError(
"A tenant API principal is required."
)
if not (
has_scope(principal, SANCTIONS_REVIEW_SCOPE)
or has_scope(principal, SANCTIONS_ADMIN_SCOPE)
):
raise RiskSanctionsAccessError(
f"Missing scope: {SANCTIONS_REVIEW_SCOPE}"
)
def _aware(value: datetime) -> datetime:
if value.tzinfo is None:
return value.replace(tzinfo=timezone.utc)
return value
__all__ = [
"DispositionDecision",
"DispositionInput",
"get_candidate",
"list_candidate_dispositions",
"list_review_queue",
"record_disposition",
]

602
src/govoplan_risk_compliance/backend/router.py Normal file
View File

@@ -0,0 +1,602 @@
from __future__ import annotations
from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
from sqlalchemy.orm import Session
from govoplan_core.audit.logging import audit_from_principal
from govoplan_core.auth import ApiPrincipal, get_api_principal
from govoplan_core.core.sanctions import sanctions_snapshot_provider
from govoplan_core.db.session import get_session
from govoplan_risk_compliance.backend.review import (
DispositionInput,
get_candidate,
list_candidate_dispositions,
list_review_queue,
record_disposition,
)
from govoplan_risk_compliance.backend.sanctions_catalog import (
RiskSanctionsAccessError,
RiskSanctionsConflictError,
RiskSanctionsError,
RiskSanctionsNotFoundError,
import_connector_snapshot,
list_list_snapshots,
)
from govoplan_risk_compliance.backend.schemas import (
CandidateDetailResponse,
CandidateResponse,
ConnectorSnapshotListResponse,
ConnectorSnapshotResponse,
DispositionCreateRequest,
DispositionCreateResponse,
DispositionListResponse,
DispositionResponse,
ListSnapshotImportRequest,
ListSnapshotImportResponse,
ListSnapshotListResponse,
ListSnapshotResponse,
ReviewQueueItemResponse,
ReviewQueueResponse,
ScreeningRunCreateRequest,
ScreeningRunCreateResponse,
ScreeningRunListResponse,
ScreeningRunResponse,
ScreeningRunSummaryResponse,
SubjectSnapshotResponse,
)
from govoplan_risk_compliance.backend.screening import (
ScreeningPolicy,
ScreeningSubject,
get_screening_run,
list_screening_runs,
run_screening,
)
router = APIRouter(
prefix="/risk-compliance",
tags=["risk-compliance"],
)
def _registry(request: Request) -> object | None:
return getattr(request.app.state, "govoplan_registry", None)
def _http_error(exc: RiskSanctionsError) -> HTTPException:
if isinstance(exc, RiskSanctionsNotFoundError):
return HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=str(exc),
)
if isinstance(exc, RiskSanctionsAccessError):
return HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=str(exc),
)
if isinstance(exc, RiskSanctionsConflictError):
return HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=str(exc),
)
return HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
detail=str(exc),
)
@router.get(
"/sanctions/source-snapshots",
response_model=ConnectorSnapshotListResponse,
)
def api_list_connector_snapshots(
request: Request,
limit: int = Query(default=100, ge=1, le=500),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> ConnectorSnapshotListResponse:
provider = sanctions_snapshot_provider(_registry(request))
if provider is None:
return ConnectorSnapshotListResponse(
available=False,
snapshots=[],
)
try:
snapshots = provider.list_snapshots(
session,
principal,
limit=limit,
)
except ValueError as exc:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
detail=str(exc),
) from exc
return ConnectorSnapshotListResponse(
available=True,
snapshots=[
ConnectorSnapshotResponse.model_validate(
{
"ref": item.ref,
"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,
"byte_count": item.byte_count,
"sha256": item.sha256,
"parser_version": item.parser_version,
"raw_evidence_ref": item.raw_evidence_ref,
"connector_run_id": item.connector_run_id,
"licence_notes": item.licence_notes,
"trust_notes": item.trust_notes,
}
)
for item in snapshots
],
)
@router.post(
"/sanctions/list-snapshots/import",
response_model=ListSnapshotImportResponse,
)
def api_import_list_snapshot(
payload: ListSnapshotImportRequest,
request: Request,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> ListSnapshotImportResponse:
try:
item, created = import_connector_snapshot(
session,
principal,
registry=_registry(request),
connector_snapshot_ref=payload.connector_snapshot_ref,
)
except RiskSanctionsError as exc:
raise _http_error(exc) from exc
audit_from_principal(
session,
principal,
action="risk_compliance.sanctions_snapshot.imported",
scope="tenant",
object_type="risk_sanctions_list_snapshot",
object_id=item.id,
details={
"created": created,
"provider_id": item.provider_id,
"source_version": item.source_version,
"sha256": item.sha256,
"entry_count": item.entry_count,
},
)
session.commit()
return ListSnapshotImportResponse(
snapshot=_list_snapshot_response(item),
created=created,
)
@router.get(
"/sanctions/list-snapshots",
response_model=ListSnapshotListResponse,
)
def api_list_list_snapshots(
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> ListSnapshotListResponse:
try:
items = list_list_snapshots(session, principal)
except RiskSanctionsError as exc:
raise _http_error(exc) from exc
return ListSnapshotListResponse(
snapshots=[
_list_snapshot_response(item)
for item in items
]
)
@router.post(
"/sanctions/screenings",
response_model=ScreeningRunCreateResponse,
)
def api_run_screening(
payload: ScreeningRunCreateRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> ScreeningRunCreateResponse:
try:
item, created = run_screening(
session,
principal,
list_snapshot_id=payload.list_snapshot_id,
idempotency_key=payload.idempotency_key,
subject=ScreeningSubject(
subject_type=payload.subject.subject_type,
subject_ref=payload.subject.subject_ref,
primary_name=payload.subject.primary_name,
aliases=tuple(payload.subject.aliases),
identifiers=tuple(
value.model_dump()
for value in payload.subject.identifiers
),
dates=tuple(payload.subject.dates),
addresses=tuple(
value.model_dump(exclude_none=True)
for value in payload.subject.addresses
),
),
policy=ScreeningPolicy(
fuzzy_threshold=payload.policy.fuzzy_threshold,
max_snapshot_age_days=(
payload.policy.max_snapshot_age_days
),
max_candidates=payload.policy.max_candidates,
),
)
except RiskSanctionsError as exc:
raise _http_error(exc) from exc
audit_from_principal(
session,
principal,
action="risk_compliance.sanctions_screening.completed",
scope="tenant",
object_type="risk_screening_run",
object_id=item.id,
details={
"created": created,
"outcome": item.outcome,
"candidate_count": item.candidate_count,
"list_snapshot_id": item.list_snapshot_id,
"matcher_version": item.matcher_version,
"normalization_version": item.normalization_version,
"policy_version": item.policy_version,
},
)
session.commit()
return ScreeningRunCreateResponse(
run=_run_response(item),
created=created,
)
@router.get(
"/sanctions/screenings",
response_model=ScreeningRunListResponse,
)
def api_list_screenings(
limit: int = Query(default=100, ge=1, le=500),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> ScreeningRunListResponse:
try:
items = list_screening_runs(
session,
principal,
limit=limit,
)
except RiskSanctionsError as exc:
raise _http_error(exc) from exc
return ScreeningRunListResponse(
runs=[_run_summary(item) for item in items]
)
@router.get(
"/sanctions/screenings/{run_id}",
response_model=ScreeningRunResponse,
)
def api_get_screening(
run_id: str,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> ScreeningRunResponse:
try:
item = get_screening_run(
session,
principal,
run_id=run_id,
)
except RiskSanctionsError as exc:
raise _http_error(exc) from exc
return _run_response(item)
@router.get(
"/sanctions/review-queue",
response_model=ReviewQueueResponse,
)
def api_list_review_queue(
review_status: str = Query(
default="pending",
pattern=(
"^(pending|all|confirmed|false_positive|"
"needs_information|escalated|exception_review)$"
),
),
limit: int = Query(default=100, ge=1, le=500),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> ReviewQueueResponse:
try:
items = list_review_queue(
session,
principal,
status=review_status,
limit=limit,
)
except RiskSanctionsError as exc:
raise _http_error(exc) from exc
return ReviewQueueResponse(
candidates=[
ReviewQueueItemResponse(
id=item.id,
run_id=item.run_id,
score=item.score,
match_kind=item.match_kind,
review_status=item.review_status,
subject_name=item.run.subject_snapshot.primary_name,
subject_type=item.run.subject_snapshot.subject_type,
entry_name=item.entry.primary_name,
source_entry_id=item.entry.source_entry_id,
list_source_version=(
item.run.list_snapshot.source_version
),
created_at=item.created_at,
)
for item in items
]
)
@router.get(
"/sanctions/review-queue/{candidate_id}",
response_model=CandidateDetailResponse,
)
def api_get_candidate(
candidate_id: str,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> CandidateDetailResponse:
try:
item = get_candidate(
session,
principal,
candidate_id=candidate_id,
)
except RiskSanctionsError as exc:
raise _http_error(exc) from exc
return _candidate_detail_response(item)
@router.post(
"/sanctions/review-queue/{candidate_id}/dispositions",
response_model=DispositionCreateResponse,
)
def api_record_disposition(
candidate_id: str,
payload: DispositionCreateRequest,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> DispositionCreateResponse:
try:
candidate, item = record_disposition(
session,
principal,
candidate_id=candidate_id,
disposition=DispositionInput(
decision=payload.decision,
reason=payload.reason,
evidence_refs=tuple(payload.evidence_refs),
exception_scope=payload.exception_scope,
expires_at=payload.expires_at,
review_at=payload.review_at,
override_reason=payload.override_reason,
),
)
except RiskSanctionsError as exc:
raise _http_error(exc) from exc
audit_from_principal(
session,
principal,
action="risk_compliance.sanctions_disposition.recorded",
scope="tenant",
object_type="risk_screening_candidate",
object_id=candidate.id,
details={
"disposition_id": item.id,
"decision": item.decision,
"scope": item.scope,
"separation_status": item.separation_status,
"supersedes_id": item.supersedes_id,
},
)
session.commit()
return DispositionCreateResponse(
candidate=_candidate_response(candidate),
disposition=_disposition_response(item),
)
@router.get(
"/sanctions/review-queue/{candidate_id}/dispositions",
response_model=DispositionListResponse,
)
def api_list_dispositions(
candidate_id: str,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> DispositionListResponse:
try:
items = list_candidate_dispositions(
session,
principal,
candidate_id=candidate_id,
)
except RiskSanctionsError as exc:
raise _http_error(exc) from exc
return DispositionListResponse(
dispositions=[
_disposition_response(item)
for item in items
]
)
def _list_snapshot_response(item) -> ListSnapshotResponse:
return ListSnapshotResponse.model_validate(
item,
from_attributes=True,
)
def _subject_response(item) -> SubjectSnapshotResponse:
return SubjectSnapshotResponse.model_validate(
item,
from_attributes=True,
)
def _entry_response(item) -> dict:
return {
"id": item.id,
"source_entry_id": item.source_entry_id,
"subject_type": item.subject_type,
"primary_name": item.primary_name,
"normalized_name": item.normalized_name,
"original_script_name": item.original_script_name,
"reference_number": item.reference_number,
"listed_on": item.listed_on,
"programmes": list(item.programmes),
"measures": list(item.measures),
"raw_evidence_locator": item.raw_evidence_locator,
"details": dict(item.details),
"aliases": [
{
"name": value.name,
"normalized_name": value.normalized_name,
"quality": value.quality,
"alias_type": value.alias_type,
}
for value in item.aliases
],
"identifiers": [
{
"identifier_type": value.identifier_type,
"value": value.value,
"issuing_country": value.issuing_country,
"note": value.note,
}
for value in item.identifiers
],
"dates": [
{
"date_type": value.date_type,
"value": value.value,
"precision": value.precision,
}
for value in item.dates
],
"addresses": [
{
"street": value.street,
"city": value.city,
"region": value.region,
"postal_code": value.postal_code,
"country": value.country,
"note": value.note,
}
for value in item.addresses
],
}
def _candidate_response(item) -> CandidateResponse:
return CandidateResponse.model_validate(
{
"id": item.id,
"score": item.score,
"match_kind": item.match_kind,
"evidence": list(item.evidence),
"review_status": item.review_status,
"current_disposition_id": item.current_disposition_id,
"entry": _entry_response(item.entry),
}
)
def _run_summary(item) -> ScreeningRunSummaryResponse:
return ScreeningRunSummaryResponse(
id=item.id,
subject_name=item.subject_snapshot.primary_name,
subject_type=item.subject_snapshot.subject_type,
list_snapshot_id=item.list_snapshot_id,
list_source_version=item.list_snapshot.source_version,
matcher_version=item.matcher_version,
normalization_version=item.normalization_version,
policy_version=item.policy_version,
status=item.status,
outcome=item.outcome,
candidate_count=item.candidate_count,
started_at=item.started_at,
completed_at=item.completed_at,
created_at=item.created_at,
)
def _run_response(item) -> ScreeningRunResponse:
return ScreeningRunResponse(
id=item.id,
idempotency_key=item.idempotency_key,
request_hash=item.request_hash,
matcher_version=item.matcher_version,
normalization_version=item.normalization_version,
policy_version=item.policy_version,
policy_snapshot=dict(item.policy_snapshot),
status=item.status,
outcome=item.outcome,
candidate_count=item.candidate_count,
started_at=item.started_at,
completed_at=item.completed_at,
created_by=item.created_by,
created_at=item.created_at,
subject=_subject_response(item.subject_snapshot),
list_snapshot=_list_snapshot_response(item.list_snapshot),
candidates=[
_candidate_response(value)
for value in sorted(
item.candidates,
key=lambda candidate: (
-candidate.score,
candidate.id,
),
)
],
)
def _candidate_detail_response(item) -> CandidateDetailResponse:
return CandidateDetailResponse(
candidate=_candidate_response(item),
run=_run_summary(item.run),
subject=_subject_response(item.run.subject_snapshot),
list_snapshot=_list_snapshot_response(
item.run.list_snapshot
),
)
def _disposition_response(item) -> DispositionResponse:
return DispositionResponse.model_validate(
item,
from_attributes=True,
)
__all__ = ["router"]

547
src/govoplan_risk_compliance/backend/sanctions_catalog.py Normal file
View File

@@ -0,0 +1,547 @@
from __future__ import annotations
from datetime import date
from typing import Any
from defusedxml import ElementTree
from sqlalchemy import or_, select
from sqlalchemy.orm import Session
from govoplan_core.auth import ApiPrincipal, has_scope
from govoplan_core.core.sanctions import sanctions_snapshot_provider
from govoplan_core.db.base import utcnow
from govoplan_risk_compliance.backend.db.models import (
RiskSanctionsAddress,
RiskSanctionsAlias,
RiskSanctionsDate,
RiskSanctionsEntry,
RiskSanctionsIdentifier,
RiskSanctionsListSnapshot,
)
from govoplan_risk_compliance.backend.normalization import (
NORMALIZATION_VERSION,
normalize_identifier,
normalize_name,
)
from govoplan_risk_compliance.backend.permissions import (
SANCTIONS_ADMIN_SCOPE,
SANCTIONS_READ_SCOPE,
)
MAX_SANCTIONS_ENTRIES = 5_000
class RiskSanctionsError(ValueError):
pass
class RiskSanctionsAccessError(RiskSanctionsError):
pass
class RiskSanctionsNotFoundError(RiskSanctionsError):
pass
class RiskSanctionsConflictError(RiskSanctionsError):
pass
def import_connector_snapshot(
session: Session,
principal: ApiPrincipal,
*,
registry: object | None,
connector_snapshot_ref: str,
) -> tuple[RiskSanctionsListSnapshot, bool]:
_require_scope(principal, SANCTIONS_ADMIN_SCOPE)
existing = session.scalar(
select(RiskSanctionsListSnapshot).where(
RiskSanctionsListSnapshot.tenant_id
== principal.tenant_id,
RiskSanctionsListSnapshot.connector_snapshot_ref
== connector_snapshot_ref,
)
)
if existing is not None:
return existing, False
provider = sanctions_snapshot_provider(registry)
if provider is None:
raise RiskSanctionsConflictError(
"The Connectors sanctions snapshot capability is unavailable."
)
payload = provider.read_snapshot(
session,
principal,
snapshot_ref=connector_snapshot_ref,
)
source = payload.snapshot
if source.tenant_id != principal.tenant_id:
raise RiskSanctionsAccessError(
"Sanctions snapshot belongs to another tenant."
)
parsed = _parse_un_snapshot(
payload.content,
parser_version=source.parser_version,
)
item = RiskSanctionsListSnapshot(
tenant_id=principal.tenant_id,
visibility="tenant",
connector_snapshot_ref=source.ref,
provider_id=source.provider_id,
publisher=source.publisher,
jurisdiction=source.jurisdiction,
list_type=source.list_type,
source_id=source.source_id,
source_version=source.source_version,
publication_at=source.publication_at,
effective_at=source.effective_at,
acquired_at=source.acquired_at,
sha256=source.sha256,
connector_run_id=source.connector_run_id,
raw_evidence_ref=source.raw_evidence_ref,
source_parser_version=source.parser_version,
normalization_version=NORMALIZATION_VERSION,
signature_evidence=dict(source.signature_evidence),
provenance={
"connector_contract_version": source.contract_version,
"transport_evidence": dict(source.transport_evidence),
"licence_notes": source.licence_notes,
"trust_notes": source.trust_notes,
},
entry_count=len(parsed),
status="active",
imported_by=_actor_id(principal),
imported_at=utcnow(),
)
session.add(item)
session.flush()
for parsed_entry in parsed:
entry = RiskSanctionsEntry(
snapshot_id=item.id,
source_entry_id=parsed_entry["source_entry_id"],
subject_type=parsed_entry["subject_type"],
primary_name=parsed_entry["primary_name"],
normalized_name=normalize_name(
parsed_entry["primary_name"]
),
original_script_name=parsed_entry[
"original_script_name"
],
reference_number=parsed_entry["reference_number"],
listed_on=parsed_entry["listed_on"],
programmes=parsed_entry["programmes"],
measures=parsed_entry["measures"],
raw_evidence_locator=parsed_entry[
"raw_evidence_locator"
],
details=parsed_entry["details"],
)
session.add(entry)
session.flush()
session.add_all(
[
RiskSanctionsAlias(
entry_id=entry.id,
name=alias["name"],
normalized_name=normalize_name(alias["name"]),
quality=alias["quality"],
alias_type=alias["alias_type"],
)
for alias in parsed_entry["aliases"]
]
)
session.add_all(
[
RiskSanctionsIdentifier(
entry_id=entry.id,
identifier_type=identifier["identifier_type"],
value=identifier["value"],
normalized_value=normalize_identifier(
identifier["value"]
),
issuing_country=identifier["issuing_country"],
note=identifier["note"],
)
for identifier in parsed_entry["identifiers"]
]
)
session.add_all(
[
RiskSanctionsDate(
entry_id=entry.id,
date_type=value["date_type"],
value=value["value"],
precision=value["precision"],
)
for value in parsed_entry["dates"]
]
)
session.add_all(
[
RiskSanctionsAddress(
entry_id=entry.id,
street=address["street"],
city=address["city"],
region=address["region"],
postal_code=address["postal_code"],
country=address["country"],
note=address["note"],
)
for address in parsed_entry["addresses"]
]
)
session.flush()
return item, True
def list_list_snapshots(
session: Session,
principal: ApiPrincipal,
) -> tuple[RiskSanctionsListSnapshot, ...]:
_require_scope(principal, SANCTIONS_READ_SCOPE)
return tuple(
session.scalars(
select(RiskSanctionsListSnapshot)
.where(
or_(
RiskSanctionsListSnapshot.tenant_id
== principal.tenant_id,
RiskSanctionsListSnapshot.visibility == "global",
)
)
.order_by(
RiskSanctionsListSnapshot.acquired_at.desc(),
RiskSanctionsListSnapshot.id.desc(),
)
)
)
def get_list_snapshot(
session: Session,
principal: ApiPrincipal,
*,
snapshot_id: str,
) -> RiskSanctionsListSnapshot:
_require_scope(principal, SANCTIONS_READ_SCOPE)
item = session.scalar(
select(RiskSanctionsListSnapshot).where(
RiskSanctionsListSnapshot.id == snapshot_id,
or_(
RiskSanctionsListSnapshot.tenant_id
== principal.tenant_id,
RiskSanctionsListSnapshot.visibility == "global",
),
)
)
if item is None:
raise RiskSanctionsNotFoundError(
"Sanctions list snapshot was not found."
)
return item
def _parse_un_snapshot(
content: bytes,
*,
parser_version: str,
) -> tuple[dict[str, Any], ...]:
if parser_version != "unsc-xml-v1":
raise RiskSanctionsConflictError(
f"Unsupported sanctions parser version: {parser_version}"
)
try:
root = ElementTree.fromstring(content)
except Exception as exc:
raise RiskSanctionsConflictError(
"Stored sanctions source evidence is malformed."
) from exc
if _local_name(root.tag) != "CONSOLIDATED_LIST":
raise RiskSanctionsConflictError(
"Stored sanctions evidence has an unsupported schema."
)
entries: list[dict[str, Any]] = []
for section_name, item_name, subject_type in (
("INDIVIDUALS", "INDIVIDUAL", "person"),
("ENTITIES", "ENTITY", "entity"),
):
section = _child(root, section_name)
if section is None:
raise RiskSanctionsConflictError(
f"Sanctions evidence is missing {section_name}."
)
for index, element in enumerate(
_children(section, item_name),
start=1,
):
entries.append(
_parsed_entry(
element,
subject_type=subject_type,
locator=f"{section_name}/{item_name}[{index}]",
)
)
if len(entries) > MAX_SANCTIONS_ENTRIES:
raise RiskSanctionsConflictError(
"Sanctions source exceeds the entry safety limit."
)
return tuple(entries)
def _parsed_entry(
element,
*,
subject_type: str,
locator: str,
) -> dict[str, Any]:
name = " ".join(
value
for value in (
_text(element, "FIRST_NAME"),
_text(element, "SECOND_NAME"),
_text(element, "THIRD_NAME"),
_text(element, "FOURTH_NAME"),
)
if value
).strip()
if not name:
raise RiskSanctionsConflictError(
f"Sanctions entry at {locator} has no name."
)
source_entry_id = (
_text(element, "REFERENCE_NUMBER")
or _text(element, "DATAID")
)
if not source_entry_id:
raise RiskSanctionsConflictError(
f"Sanctions entry at {locator} has no stable source ID."
)
alias_tag = (
"INDIVIDUAL_ALIAS"
if subject_type == "person"
else "ENTITY_ALIAS"
)
address_tag = (
"INDIVIDUAL_ADDRESS"
if subject_type == "person"
else "ENTITY_ADDRESS"
)
aliases = [
{
"name": alias_name,
"quality": _text(alias, "QUALITY"),
"alias_type": "aka",
}
for alias in _children(element, alias_tag)
if (alias_name := _text(alias, "ALIAS_NAME"))
and normalize_name(alias_name) != normalize_name(name)
]
identifiers = [
{
"identifier_type": (
_text(document, "TYPE_OF_DOCUMENT")
or "document"
),
"value": number,
"issuing_country": _text(
document,
"ISSUING_COUNTRY",
),
"note": _text(document, "NOTE"),
}
for document in _children(
element,
"INDIVIDUAL_DOCUMENT",
)
if (number := _text(document, "NUMBER"))
]
dates = [
_parsed_date(value)
for value in _children(
element,
"INDIVIDUAL_DATE_OF_BIRTH",
)
]
addresses = [
{
"street": _text(address, "STREET"),
"city": _text(address, "CITY"),
"region": (
_text(address, "STATE_PROVINCE")
or _text(address, "REGION")
),
"postal_code": _text(address, "ZIP_CODE"),
"country": _text(address, "COUNTRY"),
"note": _text(address, "NOTE"),
}
for address in _children(element, address_tag)
if any(
_text(address, field)
for field in (
"STREET",
"CITY",
"STATE_PROVINCE",
"REGION",
"ZIP_CODE",
"COUNTRY",
"NOTE",
)
)
]
programme = _text(element, "UN_LIST_TYPE")
return {
"source_entry_id": source_entry_id,
"subject_type": subject_type,
"primary_name": name,
"original_script_name": _text(
element,
"NAME_ORIGINAL_SCRIPT",
),
"reference_number": _text(
element,
"REFERENCE_NUMBER",
),
"listed_on": _parse_date(_text(element, "LISTED_ON")),
"programmes": [programme] if programme else [],
"measures": [],
"raw_evidence_locator": locator,
"details": {
"titles": _texts(element, "TITLE/VALUE"),
"designations": _texts(
element,
"DESIGNATION/VALUE",
),
"nationalities": _texts(
element,
"NATIONALITY/VALUE",
),
"source_data_id": _text(element, "DATAID"),
"source_version": _text(element, "VERSIONNUM"),
},
"aliases": aliases,
"identifiers": identifiers,
"dates": dates,
"addresses": addresses,
}
def _parsed_date(element) -> dict[str, str]:
exact = _text(element, "DATE")
if exact:
return {
"date_type": "birth",
"value": exact,
"precision": "day",
}
year = _text(element, "YEAR")
if year:
return {
"date_type": "birth",
"value": year,
"precision": "year",
}
from_year = _text(element, "FROM_YEAR")
to_year = _text(element, "TO_YEAR")
return {
"date_type": "birth",
"value": "-".join(
value
for value in (from_year, to_year)
if value
)
or "unknown",
"precision": "range",
}
def _child(element, name: str):
return next(
(
child
for child in element
if _local_name(child.tag) == name
),
None,
)
def _children(element, name: str):
return tuple(
child
for child in element
if _local_name(child.tag) == name
)
def _text(element, path: str) -> str | None:
current = element
for name in path.split("/"):
current = _child(current, name)
if current is None:
return None
clean = " ".join(str(current.text or "").split())
return clean or None
def _texts(element, path: str) -> list[str]:
parent_name, child_name = path.split("/", 1)
parent = _child(element, parent_name)
if parent is None:
return []
return [
value
for item in _children(parent, child_name)
if (value := " ".join(str(item.text or "").split()))
]
def _parse_date(value: str | None) -> date | None:
if not value:
return None
try:
return date.fromisoformat(value[:10])
except ValueError:
return None
def _local_name(value: str) -> str:
return value.rsplit("}", 1)[-1]
def _require_scope(
principal: ApiPrincipal,
scope: str,
) -> None:
if not isinstance(principal, ApiPrincipal):
raise RiskSanctionsAccessError(
"A tenant API principal is required."
)
if not (
has_scope(principal, scope)
or has_scope(principal, SANCTIONS_ADMIN_SCOPE)
):
raise RiskSanctionsAccessError(
f"Missing scope: {scope}"
)
def _actor_id(principal: ApiPrincipal) -> str | None:
return (
principal.account_id
or principal.membership_id
or principal.identity_id
)
__all__ = [
"MAX_SANCTIONS_ENTRIES",
"RiskSanctionsAccessError",
"RiskSanctionsConflictError",
"RiskSanctionsError",
"RiskSanctionsNotFoundError",
"get_list_snapshot",
"import_connector_snapshot",
"list_list_snapshots",
]

340
src/govoplan_risk_compliance/backend/schemas.py Normal file
View File

@@ -0,0 +1,340 @@
from __future__ import annotations
from datetime import date, datetime
from typing import Any, Literal
from pydantic import BaseModel, Field, model_validator
class ConnectorSnapshotResponse(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
byte_count: int
sha256: str
parser_version: str
raw_evidence_ref: str
connector_run_id: str
licence_notes: str | None = None
trust_notes: str | None = None
class ConnectorSnapshotListResponse(BaseModel):
available: bool
snapshots: list[ConnectorSnapshotResponse]
class ListSnapshotImportRequest(BaseModel):
connector_snapshot_ref: str = Field(
min_length=1,
max_length=300,
)
class ListSnapshotResponse(BaseModel):
id: str
visibility: 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
sha256: str
connector_run_id: str
raw_evidence_ref: str
source_parser_version: str
normalization_version: str
signature_evidence: dict[str, Any]
provenance: dict[str, Any]
entry_count: int
status: str
imported_by: str | None
imported_at: datetime
class ListSnapshotImportResponse(BaseModel):
snapshot: ListSnapshotResponse
created: bool
class ListSnapshotListResponse(BaseModel):
snapshots: list[ListSnapshotResponse]
class ScreeningIdentifierInput(BaseModel):
type: str = Field(default="document", max_length=100)
value: str = Field(min_length=1, max_length=1000)
class ScreeningAddressInput(BaseModel):
street: str | None = Field(default=None, max_length=1000)
city: str | None = Field(default=None, max_length=500)
region: str | None = Field(default=None, max_length=500)
postal_code: str | None = Field(default=None, max_length=100)
country: str | None = Field(default=None, max_length=255)
class ScreeningSubjectInput(BaseModel):
subject_type: Literal["person", "entity"]
subject_ref: str | None = Field(default=None, max_length=500)
primary_name: str | None = Field(default=None, max_length=1000)
aliases: list[str] = Field(default_factory=list, max_length=100)
identifiers: list[ScreeningIdentifierInput] = Field(
default_factory=list,
max_length=100,
)
dates: list[str] = Field(default_factory=list, max_length=100)
addresses: list[ScreeningAddressInput] = Field(
default_factory=list,
max_length=100,
)
@model_validator(mode="after")
def validate_lookup_fields(self) -> "ScreeningSubjectInput":
if not str(self.primary_name or "").strip() and not self.identifiers:
raise ValueError("A name or identifier is required.")
return self
class ScreeningPolicyInput(BaseModel):
fuzzy_threshold: float = Field(default=0.88, ge=0.8, le=1)
max_snapshot_age_days: int = Field(default=7, ge=1, le=365)
max_candidates: int = Field(default=100, ge=1, le=100)
class ScreeningRunCreateRequest(BaseModel):
list_snapshot_id: str = Field(min_length=1, max_length=36)
idempotency_key: str = Field(min_length=1, max_length=255)
subject: ScreeningSubjectInput
policy: ScreeningPolicyInput = Field(
default_factory=ScreeningPolicyInput
)
class SubjectSnapshotResponse(BaseModel):
id: str
subject_ref: str | None
subject_type: str
primary_name: str | None
normalized_name: str | None
aliases: list[str]
identifiers: list[dict[str, str]]
dates: list[str]
addresses: list[dict[str, str]]
fingerprint: str
submitted_by: str | None
class SanctionsAliasResponse(BaseModel):
name: str
normalized_name: str
quality: str | None
alias_type: str
class SanctionsIdentifierResponse(BaseModel):
identifier_type: str
value: str
issuing_country: str | None
note: str | None
class SanctionsDateResponse(BaseModel):
date_type: str
value: str
precision: str
class SanctionsAddressResponse(BaseModel):
street: str | None
city: str | None
region: str | None
postal_code: str | None
country: str | None
note: str | None
class SanctionsEntryResponse(BaseModel):
id: str
source_entry_id: str
subject_type: str
primary_name: str
normalized_name: str
original_script_name: str | None
reference_number: str | None
listed_on: date | None
programmes: list[str]
measures: list[str]
raw_evidence_locator: str
details: dict[str, Any]
aliases: list[SanctionsAliasResponse]
identifiers: list[SanctionsIdentifierResponse]
dates: list[SanctionsDateResponse]
addresses: list[SanctionsAddressResponse]
class CandidateResponse(BaseModel):
id: str
score: int
match_kind: str
evidence: list[dict[str, Any]]
review_status: str
current_disposition_id: str | None
entry: SanctionsEntryResponse
class ScreeningRunSummaryResponse(BaseModel):
id: str
subject_name: str | None
subject_type: str
list_snapshot_id: str
list_source_version: str
matcher_version: str
normalization_version: str
policy_version: str
status: str
outcome: str
candidate_count: int
started_at: datetime
completed_at: datetime | None
created_at: datetime
class ScreeningRunResponse(BaseModel):
id: str
idempotency_key: str
request_hash: str
matcher_version: str
normalization_version: str
policy_version: str
policy_snapshot: dict[str, Any]
status: str
outcome: str
candidate_count: int
started_at: datetime
completed_at: datetime | None
created_by: str | None
created_at: datetime
subject: SubjectSnapshotResponse
list_snapshot: ListSnapshotResponse
candidates: list[CandidateResponse]
class ScreeningRunCreateResponse(BaseModel):
run: ScreeningRunResponse
created: bool
class ScreeningRunListResponse(BaseModel):
runs: list[ScreeningRunSummaryResponse]
class ReviewQueueItemResponse(BaseModel):
id: str
run_id: str
score: int
match_kind: str
review_status: str
subject_name: str | None
subject_type: str
entry_name: str
source_entry_id: str
list_source_version: str
created_at: datetime
class ReviewQueueResponse(BaseModel):
candidates: list[ReviewQueueItemResponse]
class CandidateDetailResponse(BaseModel):
candidate: CandidateResponse
run: ScreeningRunSummaryResponse
subject: SubjectSnapshotResponse
list_snapshot: ListSnapshotResponse
class DispositionCreateRequest(BaseModel):
decision: Literal[
"true_match",
"false_positive",
"needs_information",
"escalated",
]
reason: str = Field(min_length=3, max_length=10_000)
evidence_refs: list[str] = Field(
default_factory=list,
max_length=100,
)
exception_scope: Literal[
"candidate",
"subject_entry",
] = "candidate"
expires_at: datetime | None = None
review_at: datetime | None = None
override_reason: str | None = Field(
default=None,
max_length=10_000,
)
class DispositionResponse(BaseModel):
id: str
candidate_id: str
decision: str
reason: str
evidence_refs: list[str]
scope: str
expires_at: datetime | None
review_at: datetime | None
actor_account_id: str | None
actor_membership_id: str | None
actor_authority: dict[str, Any]
separation_status: str
override_reason: str | None
supersedes_id: str | None
created_at: datetime
class DispositionCreateResponse(BaseModel):
candidate: CandidateResponse
disposition: DispositionResponse
class DispositionListResponse(BaseModel):
dispositions: list[DispositionResponse]
__all__ = [
"CandidateDetailResponse",
"CandidateResponse",
"ConnectorSnapshotListResponse",
"ConnectorSnapshotResponse",
"DispositionCreateRequest",
"DispositionCreateResponse",
"DispositionListResponse",
"DispositionResponse",
"ListSnapshotImportRequest",
"ListSnapshotImportResponse",
"ListSnapshotListResponse",
"ListSnapshotResponse",
"ReviewQueueItemResponse",
"ReviewQueueResponse",
"ScreeningRunCreateRequest",
"ScreeningRunCreateResponse",
"ScreeningRunListResponse",
"ScreeningRunResponse",
"ScreeningRunSummaryResponse",
"SubjectSnapshotResponse",
]

648
src/govoplan_risk_compliance/backend/screening.py Normal file
View File

@@ -0,0 +1,648 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from difflib import SequenceMatcher
import hashlib
import json
from typing import Any
from sqlalchemy import or_, select
from sqlalchemy.orm import Session, selectinload
from govoplan_core.auth import ApiPrincipal, has_scope
from govoplan_core.db.base import utcnow
from govoplan_risk_compliance.backend.db.models import (
RiskSanctionsEntry,
RiskSanctionsListSnapshot,
RiskScreeningCandidate,
RiskScreeningException,
RiskScreeningRun,
RiskScreeningSubjectSnapshot,
)
from govoplan_risk_compliance.backend.normalization import (
NORMALIZATION_VERSION,
normalize_identifier,
normalize_name,
subject_fingerprint,
)
from govoplan_risk_compliance.backend.permissions import (
SANCTIONS_ADMIN_SCOPE,
SANCTIONS_READ_SCOPE,
SANCTIONS_SCREEN_SCOPE,
)
from govoplan_risk_compliance.backend.sanctions_catalog import (
RiskSanctionsAccessError,
RiskSanctionsConflictError,
RiskSanctionsNotFoundError,
)
MATCHER_VERSION = "sanctions-matcher-v1"
POLICY_VERSION = "sanctions-policy-v1"
MAX_MATCH_ENTRIES = 5_000
MAX_CANDIDATES = 100
DEFAULT_FUZZY_THRESHOLD = 0.88
DEFAULT_MAX_SNAPSHOT_AGE_DAYS = 7
@dataclass(frozen=True, slots=True)
class ScreeningSubject:
subject_type: str
primary_name: str | None = None
subject_ref: str | None = None
aliases: tuple[str, ...] = ()
identifiers: tuple[dict[str, str], ...] = ()
dates: tuple[str, ...] = ()
addresses: tuple[dict[str, str], ...] = ()
def __post_init__(self) -> None:
if self.subject_type not in {"person", "entity"}:
raise RiskSanctionsConflictError(
"Screening subject type must be person or entity."
)
if not normalize_name(self.primary_name) and not any(
normalize_identifier(item.get("value"))
for item in self.identifiers
):
raise RiskSanctionsConflictError(
"A screening subject needs a name or identifier."
)
@dataclass(frozen=True, slots=True)
class ScreeningPolicy:
fuzzy_threshold: float = DEFAULT_FUZZY_THRESHOLD
max_snapshot_age_days: int = DEFAULT_MAX_SNAPSHOT_AGE_DAYS
max_candidates: int = MAX_CANDIDATES
def __post_init__(self) -> None:
if not 0.8 <= self.fuzzy_threshold <= 1:
raise RiskSanctionsConflictError(
"Fuzzy threshold must be between 0.8 and 1."
)
if not 1 <= self.max_snapshot_age_days <= 365:
raise RiskSanctionsConflictError(
"Snapshot age must be between 1 and 365 days."
)
if not 1 <= self.max_candidates <= MAX_CANDIDATES:
raise RiskSanctionsConflictError(
f"Candidate limit must be between 1 and {MAX_CANDIDATES}."
)
def to_dict(self) -> dict[str, Any]:
return {
"fuzzy_threshold": self.fuzzy_threshold,
"max_snapshot_age_days": self.max_snapshot_age_days,
"max_candidates": self.max_candidates,
"fuzzy_auto_confirms": False,
}
def run_screening(
session: Session,
principal: ApiPrincipal,
*,
list_snapshot_id: str,
idempotency_key: str,
subject: ScreeningSubject,
policy: ScreeningPolicy | None = None,
) -> tuple[RiskScreeningRun, bool]:
_require_scope(principal, SANCTIONS_SCREEN_SCOPE)
clean_key = idempotency_key.strip()
if not clean_key or len(clean_key) > 255:
raise RiskSanctionsConflictError(
"Idempotency key must contain 1 to 255 characters."
)
effective_policy = policy or ScreeningPolicy()
canonical_subject = _canonical_subject(subject)
request_hash = _request_hash(
list_snapshot_id=list_snapshot_id,
subject=canonical_subject,
policy=effective_policy,
)
existing = session.scalar(
select(RiskScreeningRun).where(
RiskScreeningRun.tenant_id == principal.tenant_id,
RiskScreeningRun.idempotency_key == clean_key,
)
)
if existing is not None:
if existing.request_hash != request_hash:
raise RiskSanctionsConflictError(
"Idempotency key was already used for another screening request."
)
return get_screening_run(
session,
principal,
run_id=existing.id,
), False
list_snapshot = _visible_list_snapshot(
session,
principal,
snapshot_id=list_snapshot_id,
)
actor_id = _actor_id(principal)
subject_snapshot = RiskScreeningSubjectSnapshot(
tenant_id=principal.tenant_id,
subject_ref=_bounded(subject.subject_ref, 500),
subject_type=subject.subject_type,
primary_name=_bounded(subject.primary_name, 1000),
normalized_name=canonical_subject["primary_name"],
aliases=list(canonical_subject["aliases"]),
identifiers=list(canonical_subject["identifiers"]),
dates=list(canonical_subject["dates"]),
addresses=list(canonical_subject["addresses"]),
fingerprint=subject_fingerprint(canonical_subject),
submitted_by=actor_id,
)
now = utcnow()
run = RiskScreeningRun(
tenant_id=principal.tenant_id,
subject_snapshot=subject_snapshot,
list_snapshot_id=list_snapshot.id,
idempotency_key=clean_key,
request_hash=request_hash,
matcher_version=MATCHER_VERSION,
normalization_version=NORMALIZATION_VERSION,
policy_version=POLICY_VERSION,
policy_snapshot=effective_policy.to_dict(),
status="running",
outcome="insufficient",
candidate_count=0,
started_at=now,
created_by=actor_id,
)
session.add(run)
session.flush()
list_state = _list_state(
list_snapshot,
now=now,
max_age_days=effective_policy.max_snapshot_age_days,
)
if list_state in {"unavailable", "stale"}:
run.status = "completed"
run.outcome = list_state
run.completed_at = utcnow()
session.flush()
return get_screening_run(
session,
principal,
run_id=run.id,
), True
entries = tuple(
session.scalars(
select(RiskSanctionsEntry)
.where(RiskSanctionsEntry.snapshot_id == list_snapshot.id)
.options(
selectinload(RiskSanctionsEntry.aliases),
selectinload(RiskSanctionsEntry.identifiers),
selectinload(RiskSanctionsEntry.dates),
selectinload(RiskSanctionsEntry.addresses),
)
.limit(MAX_MATCH_ENTRIES + 1)
)
)
if len(entries) > MAX_MATCH_ENTRIES:
run.status = "completed"
run.outcome = "unavailable"
run.policy_snapshot = {
**run.policy_snapshot,
"failure": "entry_limit_exceeded",
}
run.completed_at = utcnow()
session.flush()
return get_screening_run(
session,
principal,
run_id=run.id,
), True
candidates = sorted(
(
candidate
for entry in entries
if (
candidate := _match_entry(
subject_snapshot,
entry,
fuzzy_threshold=effective_policy.fuzzy_threshold,
)
)
is not None
),
key=lambda item: (-item["score"], item["entry"].source_entry_id),
)[: effective_policy.max_candidates]
active_exceptions = _active_exceptions(
session,
tenant_id=principal.tenant_id,
subject_fingerprint_value=subject_snapshot.fingerprint,
entry_refs={
item["entry"].source_entry_id
for item in candidates
},
now=now,
)
for item in candidates:
entry = item["entry"]
evidence = list(item["evidence"])
exception = active_exceptions.get(entry.source_entry_id)
review_status = "pending"
if exception is not None:
review_status = "exception_review"
evidence.append(
{
"kind": "prior_exception",
"exception_id": exception.id,
"expires_at": exception.expires_at.isoformat(),
"effect": "review_required",
}
)
session.add(
RiskScreeningCandidate(
tenant_id=principal.tenant_id,
run_id=run.id,
entry_id=entry.id,
score=item["score"],
match_kind=item["match_kind"],
evidence=evidence,
review_status=review_status,
)
)
run.candidate_count = len(candidates)
run.status = "completed"
run.outcome = "potential" if candidates else "clear"
run.completed_at = utcnow()
session.flush()
return get_screening_run(
session,
principal,
run_id=run.id,
), True
def get_screening_run(
session: Session,
principal: ApiPrincipal,
*,
run_id: str,
) -> RiskScreeningRun:
_require_scope(principal, SANCTIONS_READ_SCOPE)
item = session.scalar(
select(RiskScreeningRun)
.where(
RiskScreeningRun.id == run_id,
RiskScreeningRun.tenant_id == principal.tenant_id,
)
.options(
selectinload(RiskScreeningRun.subject_snapshot),
selectinload(RiskScreeningRun.list_snapshot),
selectinload(RiskScreeningRun.candidates).options(
selectinload(RiskScreeningCandidate.entry).options(
selectinload(RiskSanctionsEntry.aliases),
selectinload(RiskSanctionsEntry.identifiers),
selectinload(RiskSanctionsEntry.dates),
selectinload(RiskSanctionsEntry.addresses),
)
),
)
)
if item is None:
raise RiskSanctionsNotFoundError(
"Screening run was not found."
)
return item
def list_screening_runs(
session: Session,
principal: ApiPrincipal,
*,
limit: int = 100,
) -> tuple[RiskScreeningRun, ...]:
_require_scope(principal, SANCTIONS_READ_SCOPE)
return tuple(
session.scalars(
select(RiskScreeningRun)
.where(RiskScreeningRun.tenant_id == principal.tenant_id)
.options(
selectinload(RiskScreeningRun.subject_snapshot),
selectinload(RiskScreeningRun.list_snapshot),
)
.order_by(
RiskScreeningRun.created_at.desc(),
RiskScreeningRun.id.desc(),
)
.limit(max(1, min(limit, 500)))
)
)
def _visible_list_snapshot(
session: Session,
principal: ApiPrincipal,
*,
snapshot_id: str,
) -> RiskSanctionsListSnapshot:
item = session.scalar(
select(RiskSanctionsListSnapshot).where(
RiskSanctionsListSnapshot.id == snapshot_id,
or_(
RiskSanctionsListSnapshot.tenant_id
== principal.tenant_id,
RiskSanctionsListSnapshot.visibility == "global",
),
)
)
if item is None:
raise RiskSanctionsNotFoundError(
"Sanctions list snapshot was not found."
)
return item
def _canonical_subject(subject: ScreeningSubject) -> dict[str, Any]:
aliases = sorted(
{
normalized
for value in subject.aliases
if (normalized := normalize_name(value))
}
)
identifiers = sorted(
(
{
"type": _bounded(
item.get("type") or item.get("identifier_type"),
100,
)
or "document",
"value": normalized,
}
for item in subject.identifiers
if (
normalized := normalize_identifier(item.get("value"))
)
),
key=lambda item: (item["type"], item["value"]),
)
dates = sorted(
{
str(value).strip()[:100]
for value in subject.dates
if str(value).strip()
}
)
addresses = sorted(
(
{
key: clean
for key in (
"street",
"city",
"region",
"postal_code",
"country",
)
if (clean := normalize_name(item.get(key)))
}
for item in subject.addresses
),
key=lambda item: json.dumps(item, sort_keys=True),
)
return {
"subject_type": subject.subject_type,
"primary_name": normalize_name(subject.primary_name),
"aliases": aliases,
"identifiers": identifiers,
"dates": dates,
"addresses": addresses,
}
def _request_hash(
*,
list_snapshot_id: str,
subject: dict[str, Any],
policy: ScreeningPolicy,
) -> str:
encoded = json.dumps(
{
"list_snapshot_id": list_snapshot_id,
"subject": subject,
"policy": policy.to_dict(),
"matcher_version": MATCHER_VERSION,
"normalization_version": NORMALIZATION_VERSION,
"policy_version": POLICY_VERSION,
},
sort_keys=True,
separators=(",", ":"),
ensure_ascii=False,
).encode("utf-8")
return hashlib.sha256(encoded).hexdigest()
def _match_entry(
subject: RiskScreeningSubjectSnapshot,
entry: RiskSanctionsEntry,
*,
fuzzy_threshold: float,
) -> dict[str, Any] | None:
if subject.subject_type != entry.subject_type:
return None
subject_names = {
name
for name in (
subject.normalized_name,
*(normalize_name(value) for value in subject.aliases),
)
if name
}
entry_names = {
("primary_name", entry.normalized_name),
*(
("alias", alias.normalized_name)
for alias in entry.aliases
if alias.normalized_name
),
}
subject_identifiers = {
item.get("value", "")
for item in subject.identifiers
if item.get("value")
}
entry_identifiers = {
item.normalized_value
for item in entry.identifiers
if item.normalized_value
}
identifier_matches = sorted(
subject_identifiers & entry_identifiers
)
if identifier_matches:
return {
"entry": entry,
"score": 100,
"match_kind": "identifier_exact",
"evidence": [
{
"kind": "identifier_exact",
"subject_field": "identifiers",
"list_field": "identifiers",
"source_entry_ref": entry.source_entry_id,
"raw_evidence_locator": entry.raw_evidence_locator,
}
],
}
exact = next(
(
field
for field, entry_name in entry_names
if entry_name in subject_names
),
None,
)
if exact is not None:
score = 98 if exact == "alias" else 100
return {
"entry": entry,
"score": score,
"match_kind": (
"alias_normalized"
if exact == "alias"
else "name_normalized"
),
"evidence": [
{
"kind": "normalized_name",
"subject_field": "name",
"list_field": exact,
"source_entry_ref": entry.source_entry_id,
"raw_evidence_locator": entry.raw_evidence_locator,
}
],
}
best: tuple[float, str] | None = None
for subject_name in subject_names:
if len(subject_name) < 5:
continue
for field, entry_name in entry_names:
if len(entry_name) < 5:
continue
ratio = SequenceMatcher(
None,
subject_name,
entry_name,
autojunk=False,
).ratio()
if best is None or ratio > best[0]:
best = (ratio, field)
if best is None or best[0] < fuzzy_threshold:
return None
return {
"entry": entry,
"score": int(round(best[0] * 100)),
"match_kind": "name_fuzzy",
"evidence": [
{
"kind": "fuzzy_name",
"subject_field": "name",
"list_field": best[1],
"similarity": round(best[0], 4),
"threshold": fuzzy_threshold,
"auto_confirmed": False,
"source_entry_ref": entry.source_entry_id,
"raw_evidence_locator": entry.raw_evidence_locator,
}
],
}
def _active_exceptions(
session: Session,
*,
tenant_id: str,
subject_fingerprint_value: str,
entry_refs: set[str],
now: datetime,
) -> dict[str, RiskScreeningException]:
if not entry_refs:
return {}
items = session.scalars(
select(RiskScreeningException).where(
RiskScreeningException.tenant_id == tenant_id,
RiskScreeningException.subject_fingerprint
== subject_fingerprint_value,
RiskScreeningException.source_entry_ref.in_(entry_refs),
RiskScreeningException.status == "active",
RiskScreeningException.starts_at <= now,
RiskScreeningException.expires_at > now,
)
)
return {item.source_entry_ref: item for item in items}
def _list_state(
snapshot: RiskSanctionsListSnapshot,
*,
now: datetime,
max_age_days: int,
) -> str:
if snapshot.status != "active":
return "unavailable"
acquired_at = snapshot.acquired_at
if acquired_at.tzinfo is None:
acquired_at = acquired_at.replace(tzinfo=timezone.utc)
compare_now = now
if compare_now.tzinfo is None:
compare_now = compare_now.replace(tzinfo=timezone.utc)
if compare_now - acquired_at > timedelta(days=max_age_days):
return "stale"
return "active"
def _bounded(value: object, length: int) -> str | None:
clean = " ".join(str(value or "").split())
return clean[:length] or None
def _require_scope(
principal: ApiPrincipal,
scope: str,
) -> None:
if not isinstance(principal, ApiPrincipal):
raise RiskSanctionsAccessError(
"A tenant API principal is required."
)
if not (
has_scope(principal, scope)
or has_scope(principal, SANCTIONS_ADMIN_SCOPE)
):
raise RiskSanctionsAccessError(f"Missing scope: {scope}")
def _actor_id(principal: ApiPrincipal) -> str:
return (
principal.account_id
or principal.membership_id
or principal.identity_id
)
__all__ = [
"DEFAULT_FUZZY_THRESHOLD",
"DEFAULT_MAX_SNAPSHOT_AGE_DAYS",
"MATCHER_VERSION",
"MAX_CANDIDATES",
"POLICY_VERSION",
"ScreeningPolicy",
"ScreeningSubject",
"get_screening_run",
"list_screening_runs",
"run_screening",
]

51
tests/test_manifest.py
View File

@@ -2,22 +2,53 @@ from __future__ import annotations
import unittest
from govoplan_risk_compliance.backend.manifest import ADMIN_SCOPE, READ_SCOPE, WRITE_SCOPE, get_manifest
from govoplan_risk_compliance.backend.manifest import (
ADMIN_SCOPE,
READ_SCOPE,
SANCTIONS_ADMIN_SCOPE,
SANCTIONS_READ_SCOPE,
SANCTIONS_REVIEW_SCOPE,
SANCTIONS_SCREEN_SCOPE,
WRITE_SCOPE,
get_manifest,
)
class ManifestSeedTests(unittest.TestCase):
def test_manifest_registers_seed_contract(self) -> None:
class ManifestTests(unittest.TestCase):
def test_manifest_registers_runtime_contract(self) -> None:
manifest = get_manifest()
self.assertEqual(manifest.id, "risk-compliance")
self.assertEqual(manifest.id, "risk_compliance")
self.assertEqual(manifest.name, "Risk Compliance")
self.assertEqual(manifest.dependencies, ("access",))
self.assertEqual({permission.scope for permission in manifest.permissions}, {READ_SCOPE, WRITE_SCOPE, ADMIN_SCOPE})
self.assertEqual({role.slug for role in manifest.role_templates}, {"risk_compliance_manager", "risk_compliance_viewer"})
self.assertTrue(manifest.documentation)
self.assertIsNone(manifest.route_factory)
self.assertIsNone(manifest.migration_spec)
self.assertIsNone(manifest.frontend)
self.assertIn("connectors", manifest.optional_dependencies)
self.assertEqual(
{
permission.scope
for permission in manifest.permissions
},
{
READ_SCOPE,
WRITE_SCOPE,
ADMIN_SCOPE,
SANCTIONS_READ_SCOPE,
SANCTIONS_SCREEN_SCOPE,
SANCTIONS_REVIEW_SCOPE,
SANCTIONS_ADMIN_SCOPE,
},
)
self.assertIn(
"risk_compliance_reviewer",
{role.slug for role in manifest.role_templates},
)
self.assertIsNotNone(manifest.route_factory)
self.assertIsNotNone(manifest.migration_spec)
self.assertIsNotNone(manifest.frontend)
self.assertEqual(
"connectors.sanctions_snapshots",
manifest.requires_interfaces[0].name,
)
self.assertTrue(manifest.requires_interfaces[0].optional)
if __name__ == "__main__":

57
tests/test_migrations.py Normal file
View File

@@ -0,0 +1,57 @@
from __future__ import annotations
import tempfile
import unittest
from pathlib import Path
from alembic.runtime.migration import MigrationContext
from sqlalchemy import create_engine, inspect
from govoplan_core.db.migrations import migrate_database
from govoplan_risk_compliance.backend.manifest import get_manifest
class RiskComplianceMigrationTests(unittest.TestCase):
def test_baseline_creates_screening_evidence_tables(self) -> None:
with tempfile.TemporaryDirectory(
prefix="govoplan-risk-migration-"
) as directory:
url = f"sqlite:///{Path(directory) / 'risk.db'}"
migrate_database(
database_url=url,
enabled_modules=("risk_compliance",),
manifest_factories=(get_manifest,),
)
engine = create_engine(url)
try:
with engine.connect() as connection:
self.assertIn(
"a8b9c0d1e2f3",
set(
MigrationContext.configure(
connection
).get_current_heads()
),
)
tables = set(
inspect(connection).get_table_names()
)
self.assertIn(
"risk_sanctions_list_snapshots",
tables,
)
self.assertIn("risk_screening_runs", tables)
self.assertIn(
"risk_screening_dispositions",
tables,
)
self.assertIn(
"risk_screening_exceptions",
tables,
)
finally:
engine.dispose()
if __name__ == "__main__":
unittest.main()

383
tests/test_sanctions_screening.py Normal file
View File

@@ -0,0 +1,383 @@
from __future__ import annotations
from datetime import timedelta
from types import SimpleNamespace
import hashlib
import unittest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from govoplan_core.auth import ApiPrincipal
from govoplan_core.core.access import PrincipalRef
from govoplan_core.core.sanctions import (
CAPABILITY_CONNECTORS_SANCTIONS_SNAPSHOTS,
SanctionsSnapshotPayload,
SanctionsSnapshotReference,
)
from govoplan_core.db.base import Base, utcnow
from govoplan_risk_compliance.backend.db.models import (
RiskSanctionsAddress,
RiskSanctionsAlias,
RiskSanctionsDate,
RiskSanctionsEntry,
RiskSanctionsIdentifier,
RiskSanctionsListSnapshot,
RiskScreeningCandidate,
RiskScreeningDisposition,
RiskScreeningException,
RiskScreeningRun,
RiskScreeningSubjectSnapshot,
)
from govoplan_risk_compliance.backend.permissions import (
SANCTIONS_ADMIN_SCOPE,
SANCTIONS_READ_SCOPE,
SANCTIONS_REVIEW_SCOPE,
SANCTIONS_SCREEN_SCOPE,
)
from govoplan_risk_compliance.backend.review import (
DispositionInput,
record_disposition,
)
from govoplan_risk_compliance.backend.sanctions_catalog import (
RiskSanctionsAccessError,
RiskSanctionsConflictError,
import_connector_snapshot,
)
from govoplan_risk_compliance.backend.screening import (
MATCHER_VERSION,
ScreeningPolicy,
ScreeningSubject,
run_screening,
)
UN_XML = b"""<?xml version="1.0" encoding="UTF-8"?>
<CONSOLIDATED_LIST dateGenerated="2026-07-29T00:00:00Z">
<INDIVIDUALS>
<INDIVIDUAL>
<DATAID>42</DATAID>
<VERSIONNUM>1</VERSIONNUM>
<FIRST_NAME>Example</FIRST_NAME>
<SECOND_NAME>Person</SECOND_NAME>
<UN_LIST_TYPE>Example programme</UN_LIST_TYPE>
<REFERENCE_NUMBER>QI.42.26</REFERENCE_NUMBER>
<LISTED_ON>2026-01-02</LISTED_ON>
<INDIVIDUAL_ALIAS>
<QUALITY>Good</QUALITY>
<ALIAS_NAME>Example Alias</ALIAS_NAME>
</INDIVIDUAL_ALIAS>
<INDIVIDUAL_DOCUMENT>
<TYPE_OF_DOCUMENT>Passport</TYPE_OF_DOCUMENT>
<NUMBER>P-123 456</NUMBER>
</INDIVIDUAL_DOCUMENT>
<INDIVIDUAL_DATE_OF_BIRTH>
<DATE>1980-01-02</DATE>
</INDIVIDUAL_DATE_OF_BIRTH>
<INDIVIDUAL_ADDRESS>
<CITY>Example City</CITY>
<COUNTRY>Example Country</COUNTRY>
</INDIVIDUAL_ADDRESS>
</INDIVIDUAL>
</INDIVIDUALS>
<ENTITIES>
<ENTITY>
<DATAID>84</DATAID>
<VERSIONNUM>1</VERSIONNUM>
<FIRST_NAME>Example Trading Company</FIRST_NAME>
<REFERENCE_NUMBER>QE.84.26</REFERENCE_NUMBER>
<LISTED_ON>2026-01-03</LISTED_ON>
</ENTITY>
</ENTITIES>
</CONSOLIDATED_LIST>
"""
TABLES = (
RiskSanctionsListSnapshot.__table__,
RiskSanctionsEntry.__table__,
RiskSanctionsAlias.__table__,
RiskSanctionsIdentifier.__table__,
RiskSanctionsDate.__table__,
RiskSanctionsAddress.__table__,
RiskScreeningSubjectSnapshot.__table__,
RiskScreeningRun.__table__,
RiskScreeningCandidate.__table__,
RiskScreeningDisposition.__table__,
RiskScreeningException.__table__,
)
def principal(
account_id: str = "account-1",
*,
scopes: tuple[str, ...] = (
SANCTIONS_READ_SCOPE,
SANCTIONS_SCREEN_SCOPE,
SANCTIONS_REVIEW_SCOPE,
SANCTIONS_ADMIN_SCOPE,
),
) -> ApiPrincipal:
return ApiPrincipal(
principal=PrincipalRef(
account_id=account_id,
membership_id=f"membership-{account_id}",
tenant_id="tenant-1",
scopes=frozenset(scopes),
),
account=SimpleNamespace(id=account_id),
user=SimpleNamespace(id=account_id),
)
class _SnapshotProvider:
def __init__(self) -> None:
digest = hashlib.sha256(UN_XML).hexdigest()
self.snapshot = SanctionsSnapshotReference(
ref="sanctions-snapshot:fixture",
tenant_id="tenant-1",
provider_id="synthetic.un_fixture",
publisher="United Nations Security Council fixture",
jurisdiction="UN",
list_type="consolidated_sanctions",
source_id="synthetic-un-v1",
source_version="fixture-v1",
publication_at=utcnow(),
effective_at=utcnow(),
acquired_at=utcnow(),
content_type="application/xml",
byte_count=len(UN_XML),
sha256=digest,
parser_version="unsc-xml-v1",
raw_evidence_ref="connector-evidence:fixture",
connector_run_id="run-fixture",
)
def list_snapshots(self, session, principal, *, limit=100):
del session, principal, limit
return (self.snapshot,)
def get_snapshot(self, session, principal, *, snapshot_ref):
del session, principal
return self.snapshot if snapshot_ref == self.snapshot.ref else None
def read_snapshot(self, session, principal, *, snapshot_ref):
del session, principal
if snapshot_ref != self.snapshot.ref:
raise ValueError("not found")
return SanctionsSnapshotPayload(
snapshot=self.snapshot,
content=UN_XML,
)
class _Registry:
def __init__(self, provider) -> None:
self.provider = provider
def has_capability(self, name: str) -> bool:
return name == CAPABILITY_CONNECTORS_SANCTIONS_SNAPSHOTS
def capability(self, name: str):
return self.provider if self.has_capability(name) else None
class SanctionsScreeningTests(unittest.TestCase):
def setUp(self) -> None:
self.engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(self.engine, tables=TABLES)
self.session = Session(self.engine)
self.provider = _SnapshotProvider()
self.registry = _Registry(self.provider)
self.list_snapshot, _ = import_connector_snapshot(
self.session,
principal(),
registry=self.registry,
connector_snapshot_ref=self.provider.snapshot.ref,
)
self.session.commit()
def tearDown(self) -> None:
self.session.close()
self.engine.dispose()
def test_import_is_immutable_idempotent_and_normalized(self) -> None:
second, created = import_connector_snapshot(
self.session,
principal(),
registry=self.registry,
connector_snapshot_ref=self.provider.snapshot.ref,
)
self.assertFalse(created)
self.assertEqual(self.list_snapshot.id, second.id)
self.assertEqual(2, second.entry_count)
person = self.session.query(RiskSanctionsEntry).filter_by(
source_entry_id="QI.42.26"
).one()
self.assertEqual("example person", person.normalized_name)
self.assertEqual("example alias", person.aliases[0].normalized_name)
self.assertEqual("p123456", person.identifiers[0].normalized_value)
self.assertEqual(
"INDIVIDUALS/INDIVIDUAL[1]",
person.raw_evidence_locator,
)
def test_screening_is_versioned_explainable_and_idempotent(self) -> None:
item, created = run_screening(
self.session,
principal(),
list_snapshot_id=self.list_snapshot.id,
idempotency_key="screening-1",
subject=ScreeningSubject(
subject_type="person",
primary_name="Exampel Person",
),
)
self.assertTrue(created)
self.assertEqual("potential", item.outcome)
self.assertEqual(MATCHER_VERSION, item.matcher_version)
self.assertEqual(1, item.candidate_count)
self.assertEqual("name_fuzzy", item.candidates[0].match_kind)
self.assertEqual("pending", item.candidates[0].review_status)
self.assertFalse(
item.candidates[0].evidence[0]["auto_confirmed"]
)
replay, replay_created = run_screening(
self.session,
principal(),
list_snapshot_id=self.list_snapshot.id,
idempotency_key="screening-1",
subject=ScreeningSubject(
subject_type="person",
primary_name="Exampel Person",
),
)
self.assertFalse(replay_created)
self.assertEqual(item.id, replay.id)
with self.assertRaises(RiskSanctionsConflictError):
run_screening(
self.session,
principal(),
list_snapshot_id=self.list_snapshot.id,
idempotency_key="screening-1",
subject=ScreeningSubject(
subject_type="person",
primary_name="Different Person",
),
)
def test_identifier_match_is_exact(self) -> None:
item, _ = run_screening(
self.session,
principal(),
list_snapshot_id=self.list_snapshot.id,
idempotency_key="identifier-1",
subject=ScreeningSubject(
subject_type="person",
identifiers=(
{
"type": "passport",
"value": "P123-456",
},
),
),
)
self.assertEqual("identifier_exact", item.candidates[0].match_kind)
self.assertEqual(100, item.candidates[0].score)
def test_stale_list_has_explicit_outcome(self) -> None:
self.list_snapshot.acquired_at = utcnow() - timedelta(days=30)
item, _ = run_screening(
self.session,
principal(),
list_snapshot_id=self.list_snapshot.id,
idempotency_key="stale-1",
subject=ScreeningSubject(
subject_type="person",
primary_name="Example Person",
),
policy=ScreeningPolicy(max_snapshot_age_days=7),
)
self.assertEqual("stale", item.outcome)
self.assertEqual(0, item.candidate_count)
def test_review_is_separated_and_exception_requires_review(self) -> None:
submitter = principal(
scopes=(
SANCTIONS_READ_SCOPE,
SANCTIONS_SCREEN_SCOPE,
)
)
item, _ = run_screening(
self.session,
submitter,
list_snapshot_id=self.list_snapshot.id,
idempotency_key="review-1",
subject=ScreeningSubject(
subject_type="person",
primary_name="Example Person",
),
)
candidate = item.candidates[0]
with self.assertRaises(RiskSanctionsAccessError):
record_disposition(
self.session,
principal(
scopes=(SANCTIONS_REVIEW_SCOPE,)
),
candidate_id=candidate.id,
disposition=DispositionInput(
decision="false_positive",
reason="Independent evidence excludes this subject.",
),
)
reviewer = principal(
"account-2",
scopes=(SANCTIONS_REVIEW_SCOPE,),
)
reviewed, disposition = record_disposition(
self.session,
reviewer,
candidate_id=candidate.id,
disposition=DispositionInput(
decision="false_positive",
reason="Independent evidence excludes this subject.",
exception_scope="subject_entry",
expires_at=utcnow() + timedelta(days=30),
),
)
self.assertEqual("false_positive", reviewed.review_status)
self.assertEqual("independent", disposition.separation_status)
self.assertEqual(
1,
self.session.query(RiskScreeningException).count(),
)
repeated, _ = run_screening(
self.session,
submitter,
list_snapshot_id=self.list_snapshot.id,
idempotency_key="review-2",
subject=ScreeningSubject(
subject_type="person",
primary_name="Example Person",
),
)
self.assertEqual(
"exception_review",
repeated.candidates[0].review_status,
)
self.assertEqual(
"prior_exception",
repeated.candidates[0].evidence[-1]["kind"],
)
if __name__ == "__main__":
unittest.main()

28
webui/package.json Normal file
View File

@@ -0,0 +1,28 @@
{
"name": "@govoplan/risk-compliance-webui",
"version": "0.1.8",
"private": true,
"type": "module",
"main": "src/index.ts",
"module": "src/index.ts",
"types": "src/index.ts",
"exports": {
".": {
"types": "./src/index.ts",
"import": "./src/index.ts"
},
"./styles/risk-compliance.css": "./src/styles/risk-compliance.css"
},
"peerDependencies": {
"@govoplan/core-webui": "^0.1.14",
"lucide-react": "^1.23.0",
"react": "^19.0.0",
"react-dom": "^19.0.0",
"react-router-dom": ">=7.18.2 <8"
},
"peerDependenciesMeta": {
"@govoplan/core-webui": {
"optional": true
}
}
}

253
webui/src/api/riskCompliance.ts Normal file
View File

@@ -0,0 +1,253 @@
import {
apiFetch,
apiPath,
type ApiSettings
} from "@govoplan/core-webui";
export type ConnectorSnapshot = {
ref: string;
provider_id: string;
publisher: string;
jurisdiction: string;
source_version: string;
publication_at?: string | null;
acquired_at: string;
byte_count: number;
sha256: string;
parser_version: string;
};
export type ListSnapshot = {
id: string;
visibility: string;
provider_id: string;
publisher: string;
jurisdiction: string;
list_type: string;
source_id: string;
source_version: string;
publication_at?: string | null;
effective_at?: string | null;
acquired_at: string;
sha256: string;
connector_run_id: string;
raw_evidence_ref: string;
source_parser_version: string;
normalization_version: string;
signature_evidence: Record<string, unknown>;
provenance: Record<string, unknown>;
entry_count: number;
status: string;
imported_by?: string | null;
imported_at: string;
};
export type SubjectSnapshot = {
id: string;
subject_ref?: string | null;
subject_type: string;
primary_name?: string | null;
normalized_name?: string | null;
aliases: string[];
identifiers: Array<Record<string, string>>;
dates: string[];
addresses: Array<Record<string, string>>;
fingerprint: string;
submitted_by?: string | null;
};
export type SanctionsEntry = {
id: string;
source_entry_id: string;
subject_type: string;
primary_name: string;
original_script_name?: string | null;
reference_number?: string | null;
listed_on?: string | null;
programmes: string[];
raw_evidence_locator: string;
aliases: Array<{
name: string;
quality?: string | null;
}>;
identifiers: Array<{
identifier_type: string;
value: string;
issuing_country?: string | null;
}>;
dates: Array<{
date_type: string;
value: string;
precision: string;
}>;
addresses: Array<{
street?: string | null;
city?: string | null;
region?: string | null;
postal_code?: string | null;
country?: string | null;
}>;
};
export type ScreeningCandidate = {
id: string;
score: number;
match_kind: string;
evidence: Array<Record<string, unknown>>;
review_status: string;
current_disposition_id?: string | null;
entry: SanctionsEntry;
};
export type ScreeningRun = {
id: string;
matcher_version: string;
normalization_version: string;
policy_version: string;
policy_snapshot: Record<string, unknown>;
status: string;
outcome: string;
candidate_count: number;
started_at: string;
completed_at?: string | null;
created_at: string;
subject: SubjectSnapshot;
list_snapshot: ListSnapshot;
candidates: ScreeningCandidate[];
};
export type ReviewQueueItem = {
id: string;
run_id: string;
score: number;
match_kind: string;
review_status: string;
subject_name?: string | null;
subject_type: string;
entry_name: string;
source_entry_id: string;
list_source_version: string;
created_at: string;
};
export type CandidateDetail = {
candidate: ScreeningCandidate;
subject: SubjectSnapshot;
list_snapshot: ListSnapshot;
run: {
id: string;
matcher_version: string;
normalization_version: string;
policy_version: string;
outcome: string;
created_at: string;
};
};
export async function listConnectorSnapshots(
settings: ApiSettings
) {
return apiFetch<{
available: boolean;
snapshots: ConnectorSnapshot[];
}>(
settings,
"/api/v1/risk-compliance/sanctions/source-snapshots"
);
}
export async function importListSnapshot(
settings: ApiSettings,
connectorSnapshotRef: string
) {
return apiFetch<{
snapshot: ListSnapshot;
created: boolean;
}>(
settings,
"/api/v1/risk-compliance/sanctions/list-snapshots/import",
{
method: "POST",
body: JSON.stringify({
connector_snapshot_ref: connectorSnapshotRef
})
}
);
}
export async function listListSnapshots(settings: ApiSettings) {
return apiFetch<{ snapshots: ListSnapshot[] }>(
settings,
"/api/v1/risk-compliance/sanctions/list-snapshots"
);
}
export async function runScreening(
settings: ApiSettings,
input: {
list_snapshot_id: string;
idempotency_key: string;
subject: {
subject_type: "person" | "entity";
primary_name: string;
identifiers?: Array<{type: string;value: string;}>;
};
}
) {
return apiFetch<{run: ScreeningRun;created: boolean;}>(
settings,
"/api/v1/risk-compliance/sanctions/screenings",
{
method: "POST",
body: JSON.stringify(input)
}
);
}
export async function listReviewQueue(
settings: ApiSettings,
reviewStatus = "pending"
) {
return apiFetch<{ candidates: ReviewQueueItem[] }>(
settings,
apiPath(
"/api/v1/risk-compliance/sanctions/review-queue",
{ review_status: reviewStatus, limit: 500 }
)
);
}
export async function getCandidate(
settings: ApiSettings,
candidateId: string
) {
return apiFetch<CandidateDetail>(
settings,
`/api/v1/risk-compliance/sanctions/review-queue/${encodeURIComponent(candidateId)}`
);
}
export async function createDisposition(
settings: ApiSettings,
candidateId: string,
input: {
decision: string;
reason: string;
exception_scope: "candidate" | "subject_entry";
expires_at?: string | null;
override_reason?: string | null;
}
) {
return apiFetch<{
candidate: ScreeningCandidate;
disposition: {id: string;decision: string;};
}>(
settings,
`/api/v1/risk-compliance/sanctions/review-queue/${encodeURIComponent(candidateId)}/dispositions`,
{
method: "POST",
body: JSON.stringify(input)
}
);
}

810
webui/src/features/riskCompliance/RiskCompliancePage.tsx Normal file
View File

@@ -0,0 +1,810 @@
import {
CheckCircle2,
Database,
Play,
RefreshCw,
Scale,
Upload
} from "lucide-react";
import {
useCallback,
useEffect,
useMemo,
useState,
type FormEvent
} from "react";
import {
Button,
Dialog,
DismissibleAlert,
FormField,
IconButton,
LoadingIndicator,
SegmentedControl,
StatusBadge,
ToggleSwitch,
hasScope,
type PlatformRouteContext
} from "@govoplan/core-webui";
import {
createDisposition,
getCandidate,
importListSnapshot,
listConnectorSnapshots,
listListSnapshots,
listReviewQueue,
runScreening,
type CandidateDetail,
type ConnectorSnapshot,
type ListSnapshot,
type ReviewQueueItem,
type ScreeningRun
} from "../../api/riskCompliance";
type ViewMode = "sources" | "screen" | "review";
export default function RiskCompliancePage({
settings,
auth
}: PlatformRouteContext) {
const [view, setView] = useState<ViewMode>("review");
const [sourceSnapshots, setSourceSnapshots] = useState<
ConnectorSnapshot[]
>([]);
const [sourcesAvailable, setSourcesAvailable] = useState(false);
const [listSnapshots, setListSnapshots] = useState<ListSnapshot[]>([]);
const [queue, setQueue] = useState<ReviewQueueItem[]>([]);
const [selectedCandidateId, setSelectedCandidateId] = useState("");
const [candidate, setCandidate] = useState<CandidateDetail | null>(null);
const [run, setRun] = useState<ScreeningRun | null>(null);
const [loading, setLoading] = useState(true);
const [busy, setBusy] = useState(false);
const [error, setError] = useState("");
const [notice, setNotice] = useState("");
const canAdmin = hasScope(
auth,
"risk_compliance:sanctions:admin"
);
const canScreen = hasScope(
auth,
"risk_compliance:sanctions:screen"
);
const canReview = hasScope(
auth,
"risk_compliance:sanctions:review"
);
const refresh = useCallback(async () => {
setLoading(true);
setError("");
try {
const [sources, lists, reviewQueue] = await Promise.all([
listConnectorSnapshots(settings),
listListSnapshots(settings),
canReview
? listReviewQueue(settings)
: Promise.resolve({ candidates: [] })
]);
setSourcesAvailable(sources.available);
setSourceSnapshots(sources.snapshots);
setListSnapshots(lists.snapshots);
setQueue(reviewQueue.candidates);
setSelectedCandidateId((current) =>
current &&
reviewQueue.candidates.some((item) => item.id === current)
? current
: reviewQueue.candidates[0]?.id ?? ""
);
} catch (reason) {
setError(errorMessage(reason));
} finally {
setLoading(false);
}
}, [canReview, settings]);
useEffect(() => {
void refresh();
}, [refresh]);
useEffect(() => {
if (!selectedCandidateId) {
setCandidate(null);
return;
}
let cancelled = false;
void getCandidate(settings, selectedCandidateId)
.then((item) => {
if (!cancelled) setCandidate(item);
})
.catch((reason) => {
if (!cancelled) setError(errorMessage(reason));
});
return () => {
cancelled = true;
};
}, [selectedCandidateId, settings]);
async function importSnapshot(item: ConnectorSnapshot) {
setBusy(true);
setError("");
try {
const result = await importListSnapshot(settings, item.ref);
setNotice(
result.created
? `Imported ${result.snapshot.entry_count} list entries.`
: "This immutable snapshot was already imported."
);
await refresh();
} catch (reason) {
setError(errorMessage(reason));
} finally {
setBusy(false);
}
}
return (
<main className="risk-page">
<div className="risk-toolbar">
<SegmentedControl
value={view}
onChange={setView}
ariaLabel="Risk Compliance view"
options={[
{
id: "sources",
label: (
<>
<Database size={15} />
Sources
</>
)
},
{
id: "screen",
label: (
<>
<Play size={15} />
Screen
</>
),
disabled: !canScreen
},
{
id: "review",
label: (
<>
<Scale size={15} />
Review
{queue.length > 0 && (
<span className="risk-count">{queue.length}</span>
)}
</>
),
disabled: !canReview
}
]}
/>
<span className="risk-toolbar-spacer" />
{loading && <LoadingIndicator size="sm" label="Loading" />}
<IconButton
label="Refresh"
icon={<RefreshCw size={16} />}
onClick={() => void refresh()}
disabled={loading || busy}
/>
</div>
{(error || notice) && (
<div className="risk-alerts">
{error && (
<DismissibleAlert
tone="danger"
resetKey={error}
onDismiss={() => setError("")}
>
{error}
</DismissibleAlert>
)}
{notice && !error && (
<DismissibleAlert
tone="success"
resetKey={notice}
onDismiss={() => setNotice("")}
>
{notice}
</DismissibleAlert>
)}
</div>
)}
<div className="risk-workspace">
{view === "sources" && (
<SourcesPane
available={sourcesAvailable}
sources={sourceSnapshots}
imported={listSnapshots}
canImport={canAdmin}
busy={busy}
onImport={importSnapshot}
/>
)}
{view === "screen" && (
<ScreenPane
settings={settings}
snapshots={listSnapshots}
run={run}
onRun={setRun}
onError={setError}
/>
)}
{view === "review" && (
<ReviewPane
queue={queue}
selectedId={selectedCandidateId}
detail={candidate}
busy={busy}
onSelect={setSelectedCandidateId}
onBusy={setBusy}
onError={setError}
onNotice={setNotice}
onRefresh={refresh}
settings={settings}
/>
)}
</div>
</main>
);
}
function SourcesPane({
available,
sources,
imported,
canImport,
busy,
onImport
}: {
available: boolean;
sources: ConnectorSnapshot[];
imported: ListSnapshot[];
canImport: boolean;
busy: boolean;
onImport: (item: ConnectorSnapshot) => Promise<void>;
}) {
const importedRefs = useMemo(
() => new Set(imported.map((item) => item.sha256)),
[imported]
);
return (
<section className="risk-source-layout">
<div className="risk-panel">
<header>
<div>
<strong>Connector evidence</strong>
<span>Immutable acquired source snapshots</span>
</div>
</header>
{!available && (
<div className="risk-empty">
The Connectors sanctions source capability is not enabled.
</div>
)}
<div className="risk-list">
{sources.map((item) => {
const isImported = importedRefs.has(item.sha256);
return (
<div className="risk-list-row" key={item.ref}>
<div className="risk-list-main">
<strong>{item.publisher}</strong>
<span>
{item.source_version} · {formatDate(item.acquired_at)}
</span>
<code>{item.sha256.slice(0, 16)}…</code>
</div>
{isImported ? (
<StatusBadge status="active" label="Imported" />
) : (
<IconButton
label="Import snapshot"
icon={<Upload size={16} />}
disabled={!canImport || busy}
onClick={() => void onImport(item)}
/>
)}
</div>
);
})}
</div>
</div>
<div className="risk-panel">
<header>
<div>
<strong>Screening catalogues</strong>
<span>Normalized, immutable list versions</span>
</div>
</header>
<div className="risk-list">
{imported.map((item) => (
<div className="risk-list-row" key={item.id}>
<div className="risk-list-main">
<strong>{item.publisher}</strong>
<span>
{item.entry_count.toLocaleString()} entries ·{" "}
{item.normalization_version}
</span>
<code>{item.source_version}</code>
</div>
<StatusBadge status={item.status} />
</div>
))}
{!imported.length && (
<div className="risk-empty">
No source snapshot has been imported.
</div>
)}
</div>
</div>
</section>
);
}
function ScreenPane({
settings,
snapshots,
run,
onRun,
onError
}: {
settings: PlatformRouteContext["settings"];
snapshots: ListSnapshot[];
run: ScreeningRun | null;
onRun: (run: ScreeningRun | null) => void;
onError: (message: string) => void;
}) {
const [listId, setListId] = useState(snapshots[0]?.id ?? "");
const [subjectType, setSubjectType] = useState<"person" | "entity">(
"person"
);
const [name, setName] = useState("");
const [identifier, setIdentifier] = useState("");
const [submitting, setSubmitting] = useState(false);
useEffect(() => {
if (!snapshots.some((item) => item.id === listId)) {
setListId(snapshots[0]?.id ?? "");
}
}, [listId, snapshots]);
async function submit(event: FormEvent) {
event.preventDefault();
setSubmitting(true);
onError("");
try {
const response = await runScreening(settings, {
list_snapshot_id: listId,
idempotency_key: crypto.randomUUID(),
subject: {
subject_type: subjectType,
primary_name: name,
identifiers: identifier.trim()
? [{ type: "document", value: identifier.trim() }]
: []
}
});
onRun(response.run);
} catch (reason) {
onError(errorMessage(reason));
} finally {
setSubmitting(false);
}
}
return (
<section className="risk-screen-layout">
<form className="risk-panel risk-screen-form" onSubmit={submit}>
<header>
<div>
<strong>New screening</strong>
<span>Use only the data needed for comparison</span>
</div>
</header>
<div className="risk-form-body">
<FormField label="List snapshot">
<select
value={listId}
onChange={(event) => setListId(event.target.value)}
required
>
{snapshots.map((item) => (
<option value={item.id} key={item.id}>
{item.publisher} · {item.source_version}
</option>
))}
</select>
</FormField>
<FormField label="Subject type">
<SegmentedControl
value={subjectType}
onChange={setSubjectType}
width="fill"
size="equal"
options={[
{ id: "person", label: "Person" },
{ id: "entity", label: "Entity" }
]}
/>
</FormField>
<FormField label="Name">
<input
value={name}
onChange={(event) => setName(event.target.value)}
maxLength={1000}
/>
</FormField>
<FormField label="Identifier (optional)">
<input
value={identifier}
onChange={(event) => setIdentifier(event.target.value)}
maxLength={1000}
/>
</FormField>
<Button
type="submit"
variant="primary"
disabled={
submitting ||
!listId ||
(!name.trim() && !identifier.trim())
}
>
<Play size={16} />
Run screening
</Button>
</div>
</form>
<div className="risk-panel risk-run-result">
<header>
<div>
<strong>Result</strong>
<span>Version-pinned candidate evidence</span>
</div>
{run && <StatusBadge status={run.outcome} />}
</header>
{!run && (
<div className="risk-empty">
Run a screening to inspect the result.
</div>
)}
{run && (
<div className="risk-result-body">
<div className="risk-metrics">
<span>
<strong>{run.candidate_count}</strong>
candidates
</span>
<span>
<strong>{run.matcher_version}</strong>
matcher
</span>
<span>
<strong>{run.list_snapshot.source_version}</strong>
list
</span>
</div>
{run.candidates.map((item) => (
<div className="risk-candidate-summary" key={item.id}>
<span className="risk-score">{item.score}</span>
<div>
<strong>{item.entry.primary_name}</strong>
<span>
{item.match_kind} · {item.entry.source_entry_id}
</span>
</div>
<StatusBadge status={item.review_status} />
</div>
))}
{run.outcome === "clear" && (
<div className="risk-clear">
<CheckCircle2 size={18} />
No candidate met the configured threshold.
</div>
)}
</div>
)}
</div>
</section>
);
}
function ReviewPane({
queue,
selectedId,
detail,
busy,
onSelect,
onBusy,
onError,
onNotice,
onRefresh,
settings
}: {
queue: ReviewQueueItem[];
selectedId: string;
detail: CandidateDetail | null;
busy: boolean;
onSelect: (id: string) => void;
onBusy: (value: boolean) => void;
onError: (message: string) => void;
onNotice: (message: string) => void;
onRefresh: () => Promise<void>;
settings: PlatformRouteContext["settings"];
}) {
const [dialogOpen, setDialogOpen] = useState(false);
const [decision, setDecision] = useState("false_positive");
const [reason, setReason] = useState("");
const [reusable, setReusable] = useState(false);
const [expiresAt, setExpiresAt] = useState("");
async function submitDisposition(event: FormEvent) {
event.preventDefault();
if (!detail) return;
onBusy(true);
onError("");
try {
await createDisposition(settings, detail.candidate.id, {
decision,
reason,
exception_scope: reusable ? "subject_entry" : "candidate",
expires_at:
reusable && expiresAt
? new Date(`${expiresAt}T23:59:59`).toISOString()
: null
});
setDialogOpen(false);
setReason("");
setReusable(false);
onNotice("The disposition was recorded as append-only evidence.");
await onRefresh();
} catch (reasonValue) {
onError(errorMessage(reasonValue));
} finally {
onBusy(false);
}
}
return (
<section className="risk-review-layout">
<aside className="risk-panel risk-review-queue">
<header>
<div>
<strong>Review queue</strong>
<span>{queue.length} candidates need review</span>
</div>
</header>
<div className="risk-list">
{queue.map((item) => (
<button
type="button"
className={
item.id === selectedId
? "risk-queue-row selected"
: "risk-queue-row"
}
key={item.id}
onClick={() => onSelect(item.id)}
>
<span className="risk-score">{item.score}</span>
<span className="risk-list-main">
<strong>{item.subject_name || "Identifier-only subject"}</strong>
<span>{item.entry_name}</span>
</span>
<StatusBadge status={item.review_status} />
</button>
))}
{!queue.length && (
<div className="risk-empty">
No candidates currently need review.
</div>
)}
</div>
</aside>
<div className="risk-panel risk-evidence">
<header>
<div>
<strong>Candidate evidence</strong>
<span>Subject and immutable list entry comparison</span>
</div>
{detail && (
<Button
variant="primary"
onClick={() => setDialogOpen(true)}
>
Record disposition
</Button>
)}
</header>
{!detail && (
<div className="risk-empty">
Select a candidate from the queue.
</div>
)}
{detail && (
<div className="risk-evidence-body">
<div className="risk-comparison">
<EvidenceColumn
title="Screening subject"
name={detail.subject.primary_name || "Identifier-only"}
type={detail.subject.subject_type}
aliases={detail.subject.aliases}
identifiers={detail.subject.identifiers.map(
(item) => item.value || ""
)}
dates={detail.subject.dates}
/>
<EvidenceColumn
title="Sanctions list entry"
name={detail.candidate.entry.primary_name}
type={detail.candidate.entry.subject_type}
aliases={detail.candidate.entry.aliases.map(
(item) => item.name
)}
identifiers={detail.candidate.entry.identifiers.map(
(item) =>
`${item.identifier_type}: ${item.value}`
)}
dates={detail.candidate.entry.dates.map(
(item) => item.value
)}
/>
</div>
<div className="risk-match-evidence">
<strong>
{detail.candidate.score}% ·{" "}
{detail.candidate.match_kind}
</strong>
<span>
List {detail.list_snapshot.source_version} ·{" "}
{detail.run.matcher_version} ·{" "}
{detail.run.normalization_version}
</span>
<code>
{detail.candidate.entry.raw_evidence_locator}
</code>
</div>
</div>
)}
</div>
<Dialog
open={dialogOpen}
title="Record screening disposition"
onClose={() => setDialogOpen(false)}
closeDisabled={busy}
footer={
<>
<Button
type="button"
onClick={() => setDialogOpen(false)}
disabled={busy}
>
Cancel
</Button>
<Button
form="risk-disposition-form"
type="submit"
variant="primary"
disabled={
busy ||
reason.trim().length < 3 ||
(reusable && !expiresAt)
}
>
Record
</Button>
</>
}
>
<form
id="risk-disposition-form"
className="risk-disposition-form"
onSubmit={submitDisposition}
>
<FormField label="Decision">
<select
value={decision}
onChange={(event) => setDecision(event.target.value)}
>
<option value="false_positive">False positive</option>
<option value="true_match">Confirmed match</option>
<option value="needs_information">
More information needed
</option>
<option value="escalated">Escalate</option>
</select>
</FormField>
<FormField label="Reason">
<textarea
value={reason}
onChange={(event) => setReason(event.target.value)}
rows={5}
maxLength={10000}
required
/>
</FormField>
{decision === "false_positive" && (
<>
<ToggleSwitch
checked={reusable}
onChange={(event) => setReusable(event.target.checked)}
label="Apply as a time-bounded exception to this subject and list entry"
/>
{reusable && (
<FormField label="Exception expires">
<input
type="date"
value={expiresAt}
onChange={(event) => setExpiresAt(event.target.value)}
required
/>
</FormField>
)}
</>
)}
</form>
</Dialog>
</section>
);
}
function EvidenceColumn({
title,
name,
type,
aliases,
identifiers,
dates
}: {
title: string;
name: string;
type: string;
aliases: string[];
identifiers: string[];
dates: string[];
}) {
return (
<div className="risk-evidence-column">
<span className="risk-eyebrow">{title}</span>
<strong>{name}</strong>
<span>{type}</span>
<EvidenceValues label="Aliases" values={aliases} />
<EvidenceValues label="Identifiers" values={identifiers} />
<EvidenceValues label="Dates" values={dates} />
</div>
);
}
function EvidenceValues({
label,
values
}: {
label: string;
values: string[];
}) {
return (
<div className="risk-evidence-values">
<span>{label}</span>
<strong>{values.filter(Boolean).join(", ") || "None"}</strong>
</div>
);
}
function formatDate(value: string) {
return new Intl.DateTimeFormat(undefined, {
dateStyle: "medium",
timeStyle: "short"
}).format(new Date(value));
}
function errorMessage(reason: unknown) {
return reason instanceof Error
? reason.message
: "The Risk Compliance operation failed.";
}

2
webui/src/index.ts Normal file
View File

@@ -0,0 +1,2 @@
export { default, riskComplianceModule } from "./module";
export * from "./api/riskCompliance";

71
webui/src/module.ts Normal file
View File

@@ -0,0 +1,71 @@
import { createElement, lazy } from "react";
import type { PlatformWebModule } from "@govoplan/core-webui";
import "./styles/risk-compliance.css";
const RiskCompliancePage = lazy(
() => import("./features/riskCompliance/RiskCompliancePage")
);
export const riskComplianceModule: PlatformWebModule = {
id: "risk_compliance",
label: "Risk Compliance",
version: "0.1.8",
dependencies: ["access"],
optionalDependencies: [
"audit",
"policy",
"records",
"inspections",
"files",
"tasks",
"notifications",
"connectors",
"views",
"workflow"
],
navItems: [
{
to: "/risk-compliance",
label: "Risk Compliance",
iconName: "shield-check",
anyOf: ["risk_compliance:sanctions:read"],
order: 115,
surfaceId: "risk_compliance.navigation"
}
],
routes: [
{
path: "/risk-compliance",
anyOf: ["risk_compliance:sanctions:read"],
order: 115,
surfaceId: "risk_compliance.workspace",
render: (context) => createElement(RiskCompliancePage, context)
}
],
viewSurfaces: [
{
id: "risk_compliance.sanctions.sources",
moduleId: "risk_compliance",
kind: "section",
label: "Sanctions source snapshots",
order: 20
},
{
id: "risk_compliance.sanctions.screening",
moduleId: "risk_compliance",
kind: "section",
label: "Sanctions screening",
order: 30
},
{
id: "risk_compliance.sanctions.review",
moduleId: "risk_compliance",
kind: "section",
label: "Sanctions review queue",
order: 40
}
]
};
export default riskComplianceModule;

353
webui/src/styles/risk-compliance.css Normal file
View File

@@ -0,0 +1,353 @@
.risk-page {
display: flex;
min-width: 0;
min-height: 0;
height: 100%;
flex-direction: column;
background: var(--panel);
}
.risk-toolbar {
display: flex;
align-items: center;
gap: 10px;
flex: 0 0 auto;
min-height: 50px;
border-bottom: var(--border-line);
background: var(--panel-header);
padding: 8px 14px;
}
.risk-toolbar .segmented-control-option {
gap: 6px;
}
.risk-toolbar-spacer {
flex: 1;
}
.risk-count {
min-width: 18px;
border-radius: 9px;
background: var(--accent);
color: #fff;
padding: 1px 5px;
font-size: 10px;
text-align: center;
}
.risk-alerts {
flex: 0 0 auto;
padding: 10px 14px 0;
}
.risk-workspace {
min-width: 0;
min-height: 0;
flex: 1;
overflow: hidden;
padding: 14px;
}
.risk-source-layout,
.risk-screen-layout,
.risk-review-layout {
display: grid;
min-width: 0;
min-height: 0;
height: 100%;
gap: 12px;
}
.risk-source-layout,
.risk-screen-layout {
grid-template-columns: minmax(320px, 1fr) minmax(360px, 1.35fr);
}
.risk-review-layout {
grid-template-columns: minmax(300px, 0.7fr) minmax(480px, 1.6fr);
}
.risk-panel {
display: flex;
min-width: 0;
min-height: 0;
flex-direction: column;
border: var(--border-line);
border-radius: 6px;
background: var(--surface);
overflow: hidden;
}
.risk-panel > header {
display: flex;
align-items: center;
gap: 10px;
min-height: 56px;
flex: 0 0 auto;
border-bottom: var(--border-line);
background: var(--panel-header);
padding: 9px 12px;
}
.risk-panel > header > div:first-child {
display: grid;
min-width: 0;
gap: 2px;
flex: 1;
}
.risk-panel > header strong {
color: var(--text-strong);
font-size: 13px;
}
.risk-panel > header span {
color: var(--muted);
font-size: 11px;
}
.risk-list,
.risk-result-body,
.risk-evidence-body {
min-height: 0;
flex: 1;
overflow: auto;
}
.risk-list-row,
.risk-queue-row,
.risk-candidate-summary {
display: flex;
align-items: center;
gap: 10px;
min-height: 58px;
border: 0;
border-bottom: var(--border-line);
background: transparent;
color: var(--text);
padding: 8px 11px;
}
.risk-queue-row {
width: 100%;
cursor: pointer;
text-align: left;
font: inherit;
}
.risk-queue-row:hover,
.risk-queue-row.selected {
background: var(--sidebar-hover-bg);
}
.risk-list-main {
display: grid;
min-width: 0;
gap: 3px;
flex: 1;
}
.risk-list-main strong,
.risk-list-main span,
.risk-list-main code {
overflow: hidden;
text-overflow: ellipsis;
white-space: nowrap;
}
.risk-list-main strong {
color: var(--text-strong);
font-size: 13px;
}
.risk-list-main span,
.risk-list-main code {
color: var(--muted);
font-size: 11px;
}
.risk-empty {
color: var(--muted);
padding: 24px;
text-align: center;
}
.risk-form-body {
display: grid;
align-content: start;
gap: 14px;
overflow: auto;
padding: 16px;
}
.risk-form-body input,
.risk-form-body select,
.risk-disposition-form input,
.risk-disposition-form select,
.risk-disposition-form textarea {
width: 100%;
box-sizing: border-box;
}
.risk-form-body .btn {
justify-self: start;
}
.risk-metrics {
display: grid;
grid-template-columns: repeat(3, minmax(0, 1fr));
border-bottom: var(--border-line);
}
.risk-metrics > span {
display: grid;
min-width: 0;
gap: 4px;
border-right: var(--border-line);
color: var(--muted);
padding: 12px;
font-size: 11px;
}
.risk-metrics > span:last-child {
border-right: 0;
}
.risk-metrics strong {
overflow: hidden;
color: var(--text-strong);
text-overflow: ellipsis;
white-space: nowrap;
font-size: 14px;
}
.risk-score {
display: inline-grid;
width: 38px;
height: 32px;
flex: 0 0 38px;
place-items: center;
border: 1px solid var(--warning-border);
border-radius: 4px;
background: var(--warning-soft);
color: var(--text-strong);
font-weight: 700;
}
.risk-candidate-summary > div {
display: grid;
min-width: 0;
gap: 3px;
flex: 1;
}
.risk-candidate-summary span {
color: var(--muted);
font-size: 11px;
}
.risk-clear {
display: flex;
align-items: center;
gap: 8px;
color: var(--success);
padding: 18px;
}
.risk-comparison {
display: grid;
grid-template-columns: repeat(2, minmax(0, 1fr));
border-bottom: var(--border-line);
}
.risk-evidence-column {
display: grid;
align-content: start;
gap: 5px;
min-width: 0;
border-right: var(--border-line);
padding: 16px;
}
.risk-evidence-column:last-child {
border-right: 0;
}
.risk-evidence-column > strong {
color: var(--text-strong);
font-size: 16px;
}
.risk-evidence-column > span {
color: var(--muted);
font-size: 12px;
}
.risk-eyebrow {
text-transform: uppercase;
font-size: 10px !important;
font-weight: 700;
}
.risk-evidence-values {
display: grid;
gap: 3px;
margin-top: 10px;
}
.risk-evidence-values span {
color: var(--muted);
font-size: 10px;
text-transform: uppercase;
}
.risk-evidence-values strong {
overflow-wrap: anywhere;
color: var(--text);
font-size: 12px;
font-weight: 500;
}
.risk-match-evidence {
display: grid;
gap: 6px;
padding: 16px;
}
.risk-match-evidence span,
.risk-match-evidence code {
color: var(--muted);
font-size: 11px;
}
.risk-disposition-form {
display: grid;
gap: 14px;
}
@media (max-width: 920px) {
.risk-workspace {
overflow: auto;
}
.risk-source-layout,
.risk-screen-layout,
.risk-review-layout {
height: auto;
grid-template-columns: minmax(0, 1fr);
}
.risk-panel {
min-height: 340px;
}
.risk-comparison {
grid-template-columns: minmax(0, 1fr);
}
.risk-evidence-column {
border-right: 0;
border-bottom: var(--border-line);
}
}