305 lines
9.4 KiB
Python
305 lines
9.4 KiB
Python
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",
|
|
]
|