Sanctions Screening for Vendors and Customers Against the OFAC Lists, with Compliance Review and Audit Log

id: sanctions-screening-ofac
namespace: company.team
description: |
  Screen vendors and customers against the OFAC sanctions lists: new ones within
  minutes of being added and everyone daily when the lists change. Possible
  matches go to compliance for a decision, false positives are remembered, and
  confirmed matches are blocked with a full audit log.

labels:
  team: compliance
  use-case: sanctions-screening

variables:
  ofac_base_url: https://sanctionslistservice.ofac.treas.gov/api/PublicationPreview/exports

triggers:
  - id: new_counterparties
    type: io.kestra.plugin.jdbc.postgresql.Trigger
    description: Screen new vendors and customers within minutes. Rows with
      screening_status 'new' start a run and are marked 'queued', so each one
      fires only once. Shipped disabled until the database secrets are set.
    interval: PT2M
    disabled: true
    url: "{{ secret('SANCTIONS_DB_URL') }}"
    username: "{{ secret('SANCTIONS_DB_USERNAME') }}"
    password: "{{ secret('SANCTIONS_DB_PASSWORD') }}"
    sql: SELECT id FROM counterparties WHERE screening_status = 'new'
    afterSQL: UPDATE counterparties SET screening_status = 'queued' WHERE
      screening_status = 'new'
    fetchType: FETCH

  - id: daily_rescreen
    type: io.kestra.plugin.core.trigger.Schedule
    description: Re-screen everyone every morning, because OFAC adds names every
      week. Skipped when the lists have not changed since the last full run.
    cron: "0 6 * * *"
    disabled: true
    inputs:
      scope: all

inputs:
  - id: scope
    type: SELECT
    displayName: Who to screen
    description: new screens counterparties added since the last run; all re-screens
      everyone who is not already blocked.
    values:
      - new
      - all
    defaults: new
  - id: review_threshold
    type: INT
    displayName: Review threshold
    description: Fuzzy match score (0 to 100) from which a name pair goes to
      compliance review.
    defaults: 85
  - id: include_consolidated
    type: BOOL
    displayName: Include the non-SDN consolidated lists
    defaults: true
  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: Screen and report only. No review request, no status change, no
      audit log entry and no remembered decisions.
    defaults: false

tasks:
  - id: lists
    type: io.kestra.plugin.core.flow.Parallel
    description: Download the current OFAC lists. Saved as files, so names are never
      evaluated as templates. A list that cannot be downloaded fails the run
      instead of screening against an incomplete list.
    tasks:
      - id: sdn
        type: io.kestra.plugin.core.http.Download
        uri: "{{ vars.ofac_base_url }}/SDN.CSV"
        retry:
          type: exponential
          interval: PT30S
          maxInterval: PT5M
          maxAttempts: 4
      - id: sdn_aliases
        type: io.kestra.plugin.core.http.Download
        uri: "{{ vars.ofac_base_url }}/ALT.CSV"
        retry:
          type: exponential
          interval: PT30S
          maxInterval: PT5M
          maxAttempts: 4
      - id: consolidated
        type: io.kestra.plugin.core.http.Download
        uri: "{{ vars.ofac_base_url }}/CONS_PRIM.CSV"
        retry:
          type: exponential
          interval: PT30S
          maxInterval: PT5M
          maxAttempts: 4
      - id: consolidated_aliases
        type: io.kestra.plugin.core.http.Download
        uri: "{{ vars.ofac_base_url }}/CONS_ALT.CSV"
        retry:
          type: exponential
          interval: PT30S
          maxInterval: PT5M
          maxAttempts: 4

  - id: counterparties
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: The counterparties to screen. Blocked ones are never re-screened.
    url: "{{ secret('SANCTIONS_DB_URL') }}"
    username: "{{ secret('SANCTIONS_DB_USERNAME') }}"
    password: "{{ secret('SANCTIONS_DB_PASSWORD') }}"
    sql: |
      SELECT id, name, coalesce(country, '') AS country, coalesce(type, '') AS type, screening_status
      FROM counterparties
      WHERE {{ inputs.scope == 'all' ? "screening_status IS DISTINCT FROM 'blocked'" : "screening_status IN ('new', 'queued')" }}
      ORDER BY id
    fetchType: STORE
    retry:
      type: exponential
      interval: PT10S
      maxInterval: PT1M
      maxAttempts: 3

  - id: screen
    type: io.kestra.plugin.scripts.python.Script
    description: Fuzzy-match every counterparty against every listed name and alias,
      skip pairs compliance already cleared, and list the possible matches.
    dependencies:
      - kestra
      - rapidfuzz
    inputFiles:
      SDN.CSV: "{{ outputs.sdn.uri }}"
      ALT.CSV: "{{ outputs.sdn_aliases.uri }}"
      CONS_PRIM.CSV: "{{ outputs.consolidated.uri }}"
      CONS_ALT.CSV: "{{ outputs.consolidated_aliases.uri }}"
      counterparties.ion: "{{ outputs.counterparties.uri }}"
      decisions.b64: "{{ (kv('sanctions-decisions', errorOnMissing=false) ?? {}) |
        toJson | base64encode }}"
      last_version.txt: "{{ kv('sanctions-list-version', errorOnMissing=false) ?? '' }}"
    outputFiles:
      - hits.json
      - screened.json
    script: |
      import base64, csv, hashlib, json, os, re, unicodedata
      import amazon.ion.simpleion as ion
      from rapidfuzz import fuzz, process
      from kestra import Kestra

      threshold = int({{ inputs.review_threshold }})
      lists = [("SDN", "SDN.CSV", "ALT.CSV")] + ([("Consolidated", "CONS_PRIM.CSV", "CONS_ALT.CSV")] if "{{ inputs.include_consolidated }}" == "true" else [])
      version = hashlib.sha256(b"".join(open(f, "rb").read() for f in ("SDN.CSV", "ALT.CSV", "CONS_PRIM.CSV", "CONS_ALT.CSV"))).hexdigest()[:16]
      SUFFIXES = {"LLC", "LTD", "LIMITED", "INC", "INCORPORATED", "CORP", "CORPORATION", "CO", "COMPANY", "GMBH", "AG", "SA", "SAS", "SARL",
                  "SPA", "SRL", "BV", "NV", "PLC", "PJSC", "OJSC", "JSC", "OOO", "ZAO", "PTE", "PTY", "KG", "AB", "AS", "OY", "LLP", "LP", "THE"}

      def norm(name):
          text = unicodedata.normalize("NFKD", name).encode("ascii", "ignore").decode().upper()
          tokens = [t for t in re.sub(r"[^A-Z0-9 ]+", " ", text).split() if t not in SUFFIXES]
          return " ".join(tokens)

      def plain(text):
          # List and counterparty names end up in Slack and in the review form: keep them plain.
          return re.sub(r"[{}%#|`]", "", str(text)).strip()

      entries, names = {}, {}
      for label, primary, aliases in lists:
          for r in csv.reader(open(primary, encoding="latin-1")):
              if len(r) > 3 and r[0].strip().isdigit():
                  key = f"{label}:{r[0].strip()}"
                  entries[key] = {"list": label, "uid": r[0].strip(), "name": r[1].strip(), "type": r[2].strip().replace("-0-", "entity"), "program": r[3].strip().replace("] [", ", ")}
                  names[(key, r[1].strip())] = norm(r[1])
          for r in csv.reader(open(aliases, encoding="latin-1")):
              key = f"{label}:{r[0].strip()}" if r and r[0].strip().isdigit() else None
              if key in entries and len(r) > 3:
                  names[(key, r[3].strip())] = norm(r[3])
      choices = {k: v for k, v in names.items() if len(v) >= 3}

      decisions = json.loads(base64.b64decode(open("decisions.b64").read() or "e30=") or b"{}")
      rows = []
      if os.path.getsize("counterparties.ion"):
          with open("counterparties.ion", "rb") as f:
              rows = list(ion.load(f, single_value=False))
      hits, screened, skipped = [], [], 0
      for row in rows:
          cp_id, cp_name = int(row["id"]), str(row["name"])
          screened.append({"id": cp_id, "name": plain(cp_name)})
          query = norm(cp_name)
          if len(query) < 3:
              continue
          best = {}
          # With a dict of choices, extract returns (matched value, score, dict key).
          for _, score, (key, listed_name) in process.extract(query, choices, scorer=fuzz.token_sort_ratio, score_cutoff=threshold, limit=10) + (
                  process.extract(query, choices, scorer=fuzz.token_set_ratio, score_cutoff=max(threshold, 90), limit=10) if len(query.split()) >= 2 else []):
              if score > best.get(key, (0,))[0]:
                  best[key] = (round(score), listed_name)
          for key, (score, listed_name) in best.items():
              pair = f"{cp_id}|{key}"
              if decisions.get(pair, {}).get("decision") == "false positive":
                  skipped += 1
                  continue
              e = entries[key]
              hits.append({"pair": pair, "counterparty_id": cp_id, "counterparty": plain(cp_name), "country": plain(row["country"]),
                           "list": e["list"], "list_entry": e["uid"], "listed_name": plain(e["name"]), "matched_on": plain(listed_name),
                           "type": e["type"], "program": plain(e["program"]), "score": score})
      hits.sort(key=lambda h: -h["score"])
      json.dump(hits, open("hits.json", "w")); json.dump(screened, open("screened.json", "w"))
      Kestra.outputs({
          "list_version": version,
          "list_changed": version != open("last_version.txt").read().strip(),
          "screened": len(screened),
          "hits": len(hits),
          "previously_cleared": skipped,
          "review_options": [f"{h['pair']} · {h['counterparty']} ≈ {h['listed_name']}"
                             + (f" (alias {h['matched_on']})" if h['matched_on'] != h['listed_name'] else "")
                             + f" · score {h['score']} · {h['list']} {h['program']}" for h in hits],
          "summary": (f"🛡️ *Sanctions screening ({{ inputs.scope }})*: {len(screened)} counterparties against {len(entries):,} listed parties "
                      f"and their aliases. {len(hits)} possible match(es) to review, {skipped} pair(s) cleared before."),
      })

  - id: skip_if_unchanged
    type: io.kestra.plugin.core.flow.If
    description: A daily re-screen with unchanged lists and nobody new has nothing to add.
    condition: "{{ inputs.scope == 'all' and not outputs.screen.vars.list_changed
      and outputs.screen.vars.hits == 0 }}"
    then:
      - id: nothing_new
        type: io.kestra.plugin.core.execution.Exit
        state: SUCCESS

  - id: post_summary
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: "{{ outputs.screen.vars.summary }} Details: {{ kestra.url ?? ''
      }}/ui/main/executions/{{ flow.namespace }}/{{ flow.id }}/{{ execution.id
      }}"
    retry:
      type: exponential
      interval: PT10S
      maxInterval: PT1M
      maxAttempts: 3

  - id: review
    type: io.kestra.plugin.core.flow.If
    description: Ask compliance about the possible matches only, and only outside a dry run.
    condition: "{{ not inputs.dry_run and outputs.screen.vars.hits > 0 }}"
    then:
      - id: ask_compliance
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: "🔎 {{ outputs.screen.vars.hits }} possible sanctions match(es)
          need a decision: {{ kestra.url ?? '' }}/ui/main/executions/{{
          flow.namespace }}/{{ flow.id }}/{{ execution.id }}"
        retry:
          type: exponential
          interval: PT10S
          maxInterval: PT1M
          maxAttempts: 3

      - id: decision
        type: io.kestra.plugin.core.flow.Pause
        description: Pick the true matches; every other listed pair is recorded as a
          false positive and not asked again. Without a decision within three
          days the execution is cancelled and nothing changes.
        pauseDuration: P3D
        behavior: CANCEL
        onResume:
          - id: true_matches
            type: MULTISELECT
            displayName: True matches (these counterparties will be blocked)
            expression: "{{ outputs.screen.vars.review_options }}"
            required: false
          - id: reason
            type: STRING
            displayName: Reason for the decision (kept in the audit log)

  - id: record
    type: io.kestra.plugin.core.flow.If
    description: Apply the decision, write the audit log and remember it. Nothing
      changes in a dry run.
    condition: "{{ not inputs.dry_run }}"
    then:
      - id: decide
        type: io.kestra.plugin.scripts.python.Script
        description: Turn the reviewer's choice into one audit row per screened
          counterparty and per possible match, and the list of ids to block and
          to clear.
        dependencies:
          - kestra
        inputFiles:
          hits.json: "{{ outputs.screen.outputFiles['hits.json'] }}"
          screened.json: "{{ outputs.screen.outputFiles['screened.json'] }}"
          picked.json: "{{ (outputs.decision.onResume.true_matches ?? []) | toJson }}"
          decisions.b64: "{{ (kv('sanctions-decisions', errorOnMissing=false) ?? {}) |
            toJson | base64encode }}"
        outputFiles:
          - audit.csv
          - decisions.json
        script: |
          import base64, csv, json
          from datetime import datetime, timezone
          from kestra import Kestra

          now = datetime.now(timezone.utc).isoformat(timespec="seconds")
          reason = {{ (outputs.decision.onResume.reason ?? '') | toJson }}
          version = {{ outputs.screen.vars.list_version | toJson }}
          hits = json.load(open("hits.json")); screened = json.load(open("screened.json"))
          picked = {p.split(" · ")[0] for p in json.load(open("picked.json"))}
          decisions = json.loads(base64.b64decode(open("decisions.b64").read() or "e30=") or b"{}")
          blocked = sorted({int(h["counterparty_id"]) for h in hits if h["pair"] in picked})
          rows = []
          for h in hits:
              verdict = "true match" if h["pair"] in picked else "false positive"
              if verdict == "false positive":
                  decisions[h["pair"]] = {"decision": verdict, "reason": reason, "decided_at": now, "listed_name": h["listed_name"]}
              rows.append([h["counterparty_id"], h["counterparty"], "possible match", h["list"], h["list_entry"], h["listed_name"],
                           h["matched_on"], h["program"], h["score"], verdict, reason, version, now])
          flagged = {int(h["counterparty_id"]) for h in hits}
          for s in screened:
              if int(s["id"]) not in flagged:
                  rows.append([s["id"], s["name"], "clear", "", "", "", "", "", "", "", "", version, now])
          with open("audit.csv", "w", newline="") as f:
              csv.writer(f).writerows([["counterparty_id", "counterparty", "outcome", "list", "list_entry", "listed_name", "matched_on",
                                         "program", "score", "decision", "reason", "list_version", "screened_at"]] + rows)
          json.dump(decisions, open("decisions.json", "w"))
          cleared = sorted({int(s["id"]) for s in screened} - set(blocked))
          Kestra.outputs({"blocked_ids": blocked or [-1], "cleared_ids": cleared or [-1], "blocked": len(blocked),
                          "blocked_names": sorted({h["counterparty"] for h in hits if h["pair"] in picked})})

      - id: audit_table
        type: io.kestra.plugin.jdbc.postgresql.Query
        description: Create the audit log table on first use.
        url: "{{ secret('SANCTIONS_DB_URL') }}"
        username: "{{ secret('SANCTIONS_DB_USERNAME') }}"
        password: "{{ secret('SANCTIONS_DB_PASSWORD') }}"
        sql: |
          CREATE TABLE IF NOT EXISTS sanctions_screening_log (
            counterparty_id bigint, counterparty text, outcome text, list text, list_entry text, listed_name text, matched_on text,
            program text, score int, decision text, reason text, list_version text, screened_at timestamptz)
        retry:
          type: exponential
          interval: PT10S
          maxInterval: PT1M
          maxAttempts: 3

      - id: audit_log
        type: io.kestra.plugin.jdbc.postgresql.CopyIn
        description: Bulk-load every result. Names travel as a CSV file, never inside
          SQL text.
        url: "{{ secret('SANCTIONS_DB_URL') }}"
        username: "{{ secret('SANCTIONS_DB_USERNAME') }}"
        password: "{{ secret('SANCTIONS_DB_PASSWORD') }}"
        table: sanctions_screening_log
        columns:
          [
            counterparty_id,
            counterparty,
            outcome,
            list,
            list_entry,
            listed_name,
            matched_on,
            program,
            score,
            decision,
            reason,
            list_version,
            screened_at
          ]
        format: CSV
        header: true
        from: "{{ outputs.decide.outputFiles['audit.csv'] }}"
        retry:
          type: exponential
          interval: PT10S
          maxInterval: PT1M
          maxAttempts: 3

      - id: apply_status
        type: io.kestra.plugin.jdbc.postgresql.Queries
        description: Block confirmed matches and clear everyone else that was screened.
          The ids come from the database and are checked as integers.
        url: "{{ secret('SANCTIONS_DB_URL') }}"
        username: "{{ secret('SANCTIONS_DB_USERNAME') }}"
        password: "{{ secret('SANCTIONS_DB_PASSWORD') }}"
        transaction: true
        sql: |
          UPDATE counterparties SET screening_status = 'blocked' WHERE id IN ({{ outputs.decide.vars.blocked_ids | join(', ') }});
          UPDATE counterparties SET screening_status = 'cleared' WHERE id IN ({{ outputs.decide.vars.cleared_ids | join(', ') }});
        retry:
          type: exponential
          interval: PT10S
          maxInterval: PT1M
          maxAttempts: 3

      - id: remember_decisions
        type: io.kestra.plugin.core.kv.Set
        description: Remember false positives, so the same pair is never sent to review
          again.
        key: sanctions-decisions
        kvType: JSON
        value: "{{ read(outputs.decide.outputFiles['decisions.json']) }}"

      - id: remember_list_version
        type: io.kestra.plugin.core.kv.Set
        key: sanctions-list-version
        value: "{{ outputs.screen.vars.list_version }}"

      - id: alert_blocked
        type: io.kestra.plugin.core.flow.If
        condition: "{{ outputs.decide.vars.blocked > 0 }}"
        then:
          - id: tell_blocked
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            messageText: "⛔ *Sanctions match confirmed*: {{
              outputs.decide.vars.blocked_names | join(', ') }} blocked. Reason:
              {{ outputs.decision.onResume.reason ?? '' }}"
            retry:
              type: exponential
              interval: PT10S
              maxInterval: PT1M
              maxAttempts: 3

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert with the failed task, the error and a link. No status changes
      until a decision is recorded, so the run can simply be retried.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    retry:
      type: constant
      interval: PT10S
      maxAttempts: 3
    messageText: |
      ❌ *Sanctions screening FAILED* (flow {{ flow.namespace }}.{{ flow.id }}, execution {{ execution.id }}).
      Failed task: {{ (errorLogs() | length) > 0 ? errorLogs()[0].taskId : 'unknown' }}. Error: {{ (errorLogs() | length) > 0 ? errorLogs()[0].message : 'see the execution logs' }}
      {{ kestra.url ?? '' }}/ui/main/executions/{{ flow.namespace }}/{{ flow.id }}/{{ execution.id }}

outputs:
  - id: possible_matches
    type: INT
    displayName: Possible matches
    value: "{{ outputs.screen.vars.hits ?? 0 }}"
  - id: blocked
    type: INT
    displayName: Counterparties blocked
    value: "{{ outputs.decide.vars.blocked ?? 0 }}"
  - id: hits_file
    type: STRING
    displayName: Possible matches file
    description: Internal storage URI of hits.json.
    value: "{{ outputs.screen.outputFiles['hits.json'] ?? 'none' }}"