From f3d3fdc156756d6030e4982773f7e0646ac997e6 Mon Sep 17 00:00:00 2001 From: SanctionsKit Date: Sat, 3 Oct 2026 23:16:09 -0600 Subject: [PATCH 1/4] Add isolated Kestra 2.0 batch fixture candidate Add a synthetic batch-screening flow and native fixture tests for Kestra 2.0.4. Run the engine in an unprivileged Docker container with no external network, real API keys or published ports. Keep the example marked as a candidate until all native tests pass. --- .github/workflows/kestra.yml | 48 ++++ integrations/kestra/.dockerignore | 5 + integrations/kestra/.gitignore | 6 + integrations/kestra/LICENSE | 21 ++ integrations/kestra/README.md | 82 ++++++ integrations/kestra/runtime-lock.json | 37 +++ .../kestra/sanctionskit-batch-evidence.yaml | 217 ++++++++++++++++ integrations/kestra/tests/Dockerfile | 15 ++ integrations/kestra/tests/bootstrap.py | 13 + integrations/kestra/tests/container_runner.py | 53 ++++ .../tests/fixtures/synthetic-evidence.json | 142 +++++++++++ integrations/kestra/tests/local-kestra.yml | 25 ++ integrations/kestra/tests/native_test.py | 238 ++++++++++++++++++ 13 files changed, 902 insertions(+) create mode 100644 .github/workflows/kestra.yml create mode 100644 integrations/kestra/.dockerignore create mode 100644 integrations/kestra/.gitignore create mode 100644 integrations/kestra/LICENSE create mode 100644 integrations/kestra/README.md create mode 100644 integrations/kestra/runtime-lock.json create mode 100644 integrations/kestra/sanctionskit-batch-evidence.yaml create mode 100644 integrations/kestra/tests/Dockerfile create mode 100644 integrations/kestra/tests/bootstrap.py create mode 100644 integrations/kestra/tests/container_runner.py create mode 100644 integrations/kestra/tests/fixtures/synthetic-evidence.json create mode 100644 integrations/kestra/tests/local-kestra.yml create mode 100644 integrations/kestra/tests/native_test.py diff --git a/.github/workflows/kestra.yml b/.github/workflows/kestra.yml new file mode 100644 index 0000000..1559410 --- /dev/null +++ b/.github/workflows/kestra.yml @@ -0,0 +1,48 @@ +name: Kestra 2.0 candidate fixture + +on: + push: + paths: + - "integrations/kestra/**" + - ".github/workflows/kestra.yml" + pull_request: + paths: + - "integrations/kestra/**" + - ".github/workflows/kestra.yml" + workflow_dispatch: + +permissions: + contents: read + +jobs: + native-synthetic-fixture: + runs-on: ubuntu-24.04 + timeout-minutes: 15 + defaults: + run: + working-directory: integrations/kestra + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + - name: Build pinned official runtime and fixture image + run: docker build -f tests/Dockerfile -t sanctionskit-kestra-fixture . + - name: Execute native flow with only container loopback available + run: | + mkdir -p test-results + docker run --rm --network none --cap-drop ALL \ + --security-opt no-new-privileges:true --read-only \ + --memory 3g --cpus 2 --pids-limit 512 \ + --user "$(id -u):$(id -g)" \ + --tmpfs /tmp:rw,nosuid,nodev,size=1g,mode=1777 \ + --tmpfs /work:rw,nosuid,nodev,size=1g,mode=1777 \ + --mount "type=bind,source=$PWD/test-results,target=/evidence" \ + sanctionskit-kestra-fixture + - name: Keep native fixture evidence + if: always() + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: kestra-2-native-fixture-${{ github.run_id }} + path: integrations/kestra/test-results/ + retention-days: 7 + if-no-files-found: warn diff --git a/integrations/kestra/.dockerignore b/integrations/kestra/.dockerignore new file mode 100644 index 0000000..2e85623 --- /dev/null +++ b/integrations/kestra/.dockerignore @@ -0,0 +1,5 @@ +__pycache__ +**/__pycache__ +tests/native-results.json +tests/last-failure.json +test-results diff --git a/integrations/kestra/.gitignore b/integrations/kestra/.gitignore new file mode 100644 index 0000000..ac6495d --- /dev/null +++ b/integrations/kestra/.gitignore @@ -0,0 +1,6 @@ +__pycache__/ +tests/native-results.json +tests/last-failure.json +test-results/ +synthetic-review-bundle.json +synthetic-row-summary.csv diff --git a/integrations/kestra/LICENSE b/integrations/kestra/LICENSE new file mode 100644 index 0000000..94d712c --- /dev/null +++ b/integrations/kestra/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 SanctionsKit + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/integrations/kestra/README.md b/integrations/kestra/README.md new file mode 100644 index 0000000..63f99f6 --- /dev/null +++ b/integrations/kestra/README.md @@ -0,0 +1,82 @@ +# Kestra 2.0 batch-screening candidate + +This is a candidate for native fixture testing on Kestra 2.0.4. It has passed the +2.0.4 server's flow import validation. It has not yet passed native execution on +that version, so it is not ready to use as a published integration. + +The flow submits two fictional contacts to the SanctionsKit synthetic sandbox, +waits for the batch to finish, and collects retained evidence with a CSV row +summary. Completed, failed and cancelled rows stay visible. A successful collection +does not approve onboarding or mean that every row screened successfully. + +The maintenance-version implementation passed 24 native mock scenarios on Kestra +1.3.41. That result does not establish 2.0 compatibility. This candidate replaces +`ForEach` with `Loop`, uses explicit per-iteration outputs, removes `pluginDefaults`, +updates the boolean input and HTTP timeout fields, and uses the 2.0 output API in +the tests. Native execution is the next gate. + +## Isolated fixture test + +The CI job builds an image from official Python and Eclipse Temurin images pinned +by digest, downloads Kestra 2.0.4 with a checked SHA-256, and installs serdes 2.0.6 +with a checked artifact hash. [runtime-lock.json](runtime-lock.json) records these +sources. Network access is needed only while building the image. + +Execution uses Docker `--network none`, with no published ports or Docker socket. +The test runner refuses to start unless its only network interface is `lo`. It +runs as an unprivileged user with all Linux capabilities dropped, no new privileges, +a read-only root filesystem, and temporary writable directories. The only service +key is the invented string `local-mock-only`; no GitHub secret or real API key is +needed. No hosted SanctionsKit request can leave the container. + +From this directory on a machine with Docker: + +```sh +docker build -f tests/Dockerfile -t sanctionskit-kestra-fixture . +mkdir -p test-results +docker run --rm --network none --cap-drop ALL \ + --security-opt no-new-privileges:true --read-only \ + --memory 3g --cpus 2 --pids-limit 512 \ + --user "$(id -u):$(id -g)" \ + --tmpfs /tmp:rw,nosuid,nodev,size=1g,mode=1777 \ + --tmpfs /work:rw,nosuid,nodev,size=1g,mode=1777 \ + --mount "type=bind,source=$PWD/test-results,target=/evidence" \ + sanctionskit-kestra-fixture +``` + +The fixture imports the actual flow and changes only the API host, polling interval +and polling budget. Request bodies, evidence checks, HTTP limits and exports stay +unchanged. It expects all 24 scenarios to pass, including completion, rate-limit +retry, failed/cancelled rows, timeouts, HTTP errors, unsafe IDs, mismatched evidence, +pagination, row order and counters. It checks the parent execution and native Loop +iteration executions, uses the 2.0 outputs API, downloads the JSON/CSV files, and +checks logs and outputs for the fake key. + +The server log, per-case results, isolation report and generated files go into +`test-results/`. A parser or import pass is not a substitute for this run. + +## Intended sandbox setup after qualification + +Create a free [developer workspace](https://www.sanctionskit.com/signup?workflow=api) +and a sandbox key with `batches:write` and `results:read`. Save the key as the Kestra +secret `SANCTIONSKIT_API_KEY`. The sandbox uses synthetic records and needs no card. +Production screening needs a plan and payment method. + +The two fixed contacts keep their partial birth years as strings. Each row requests +`sandbox@1` and standard retention. Use a fresh `request_key` for a new business +batch and the same key/body for retries. Redirects are disabled. A 202 only confirms +acceptance; the bounded loop waits for a terminal batch state. + +This starter expects one complete ordered page. Unexpected pagination, counters, +IDs, contacts or retained requests stop the flow. It fetches same-ID evidence only +for completed rows and preserves unsuccessful row outcomes. A `potential_match` +needs review; `no_match` means no candidates within the selected coverage and +matching rules. Neither is an account approval or legal clearance. + +The daily trigger is disabled. Before adapting this to larger inputs or enabling +scheduling, read the [batch API guide](https://www.sanctionskit.com/docs/batches) and +choose a durable business-batch key. No external notifications, account decisions +or database writes are part of this example. + +No Kestra catalog submission or placement is claimed. Upstream blueprint licensing +and catalog acceptance remain separate from company source publication. diff --git a/integrations/kestra/runtime-lock.json b/integrations/kestra/runtime-lock.json new file mode 100644 index 0000000..5629c89 --- /dev/null +++ b/integrations/kestra/runtime-lock.json @@ -0,0 +1,37 @@ +{ + "checked_at": "2026-10-04T04:51:28.638389+00:00", + "kestra": { + "version": "2.0.4", + "url": "https://github.com/kestra-io/kestra/releases/download/v2.0.4/kestra-2.0.4", + "sha256": "88ee8d44c3d4a234c9d04689e72b8ba903ae8f9f2c6ae5f94bad89f3bcd68513" + }, + "plugin": { + "coordinates": "io.kestra.plugin:plugin-serdes:2.0.6", + "sha256": "1394bf5ec0e7a57e446adc2b326e4ac72651d3fe525853d4d1c92bef60fe117e", + "maven_sha1_matches": true + }, + "docker_images": [ + { + "image": "docker.io/library/python", + "tag": "3.13-slim-trixie", + "index_digest": "sha256:bb2988715db2cf7ace7b53f38f3cffbef7c7046a656bee66245eb0ed386e2e81", + "linux_amd64_digest": "sha256:b92e6b9bb1ea9d826e9956fd8d30bf18bc384c130231d74fdf0ba3460c290b0b" + }, + { + "image": "docker.io/library/eclipse-temurin", + "tag": "25-jre-noble", + "index_digest": "sha256:398f810215757dc1926390014272579fb0e57c41ef1c8aa4f64ae761613a168b", + "linux_amd64_digest": "sha256:78498f30dd330b06755c1b134039dcbf8324bedb385136819dd519b842a3816a" + } + ], + "actions": { + "checkout": { + "version": "v7.0.1", + "sha": "3d3c42e5aac5ba805825da76410c181273ba90b1" + }, + "upload_artifact": { + "version": "v7.0.1", + "sha": "043fb46d1a93c77aae656e7c1c64a875d1fc6a0a" + } + } +} diff --git a/integrations/kestra/sanctionskit-batch-evidence.yaml b/integrations/kestra/sanctionskit-batch-evidence.yaml new file mode 100644 index 0000000..374d793 --- /dev/null +++ b/integrations/kestra/sanctionskit-batch-evidence.yaml @@ -0,0 +1,217 @@ +id: sanctionskit-batch-evidence +namespace: company.sanctionskit +description: | + Candidate for Kestra 2.0.4 native fixture CI. Not yet qualified for use. + Run two fictional contacts and collect row-level retained evidence. +inputs: + - id: request_key + type: STRING + required: true + validator: ^[A-Za-z0-9_:.-]{8,128}$ + description: A fresh key for a new batch. Reuse it when retrying the same batch. + - id: synthetic + type: BOOL + defaults: false + description: Confirm that this example contains fictional contacts only. +variables: + api_base: https://www.sanctionskit.com/api/v1 + subjects: + - subject: + name: Alex Morgan + entityType: person + birthDate: "1984" + package: sandbox@1 + retention: standard + reference: example-customer-001 + - subject: + name: Taylor Jordan + entityType: person + birthDate: "1979" + package: sandbox@1 + retention: standard + reference: example-customer-002 +tasks: + - id: require_synthetic + type: io.kestra.plugin.core.execution.Assert + conditions: + - "{{ inputs.synthetic == true }}" + errorMessage: Confirm the fictional contacts before running this sandbox example. + - id: submit_batch + type: io.kestra.plugin.core.http.Request + method: POST + uri: "{{ vars.api_base }}/batches" + headers: + Authorization: Bearer {{ secret('SANCTIONSKIT_API_KEY') }} + Content-Type: application/json + Idempotency-Key: "{{ inputs.request_key }}" + body: | + {"name":"Kestra fictional onboarding example","subjects":{{ vars.subjects | toJson }}} + timeout: PT20S + options: + followRedirects: false + allowFailed: false + timeout: + connectTimeout: PT5S + readIdleTimeout: PT10S + - id: check_acceptance + type: io.kestra.plugin.core.execution.Assert + conditions: + - "{{ outputs.submit_batch.code == 202 }}" + - "{{ fromJson(outputs.submit_batch.body).data.id | toJson | jq('test(\"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$\")') | first }}" + - "{{ fromJson(outputs.submit_batch.body).data.total == 2 }}" + - "{{ fromJson(outputs.submit_batch.body).data.status == 'pending' }}" + errorMessage: Expected acceptance of exactly two rows. Acceptance is not a completed screening. + - id: wait_for_batch + type: io.kestra.plugin.core.flow.LoopUntil + condition: "{{ (fromJson(outputs.poll_batch.body).data.status == 'completed' or fromJson(outputs.poll_batch.body).data.status == 'failed' or fromJson(outputs.poll_batch.body).data.status == 'cancelled') }}" + failOnMaxReached: true + checkFrequency: + interval: PT2S + maxIterations: 30 + maxDuration: PT2M + tasks: + - id: poll_batch + type: io.kestra.plugin.core.http.Request + method: GET + uri: "{{ vars.api_base }}/batches/{{ fromJson(outputs.submit_batch.body).data.id }}?limit=100&offset=0" + timeout: PT20S + options: + followRedirects: false + allowFailed: false + timeout: + connectTimeout: PT5S + readIdleTimeout: PT10S + headers: + Authorization: Bearer {{ secret('SANCTIONSKIT_API_KEY') }} + Content-Type: application/json + - id: check_poll_identity + type: io.kestra.plugin.core.execution.Assert + conditions: + - "{{ outputs.poll_batch.code == 200 }}" + - "{{ fromJson(outputs.poll_batch.body).data.id == fromJson(outputs.submit_batch.body).data.id }}" + - "{{ fromJson(outputs.poll_batch.body).data.environment == 'sandbox' }}" + - "{{ fromJson(outputs.poll_batch.body).data.total == 2 }}" + - "{{ (fromJson(outputs.poll_batch.body).data.status == 'pending' or fromJson(outputs.poll_batch.body).data.status == 'processing' or fromJson(outputs.poll_batch.body).data.status == 'completed' or fromJson(outputs.poll_batch.body).data.status == 'failed' or fromJson(outputs.poll_batch.body).data.status == 'cancelled') }}" + errorMessage: Batch identity, environment or state did not match the sandbox request. + - id: check_complete_page + type: io.kestra.plugin.core.execution.Assert + conditions: + - "{{ fromJson(outputs.poll_batch.body).data.nextOffset == null }}" + - "{{ fromJson(outputs.poll_batch.body).data.rows | length == 2 }}" + - "{{ fromJson(outputs.poll_batch.body).data.rows[0].row_number == 1 }}" + - "{{ fromJson(outputs.poll_batch.body).data.rows[1].row_number == 2 }}" + - "{{ fromJson(outputs.poll_batch.body).data.completed == (outputs.poll_batch.body | jq('.data.rows | map(select(.status == \"completed\")) | length') | first) }}" + - "{{ fromJson(outputs.poll_batch.body).data.failed == (outputs.poll_batch.body | jq('.data.rows | map(select(.status == \"failed\")) | length') | first) }}" + errorMessage: This two-row starter requires one complete ordered page. Stop before using partial or unexpected rows. + - id: collect_evidence + type: io.kestra.plugin.core.flow.Loop + values: + - "0" + - "1" + concurrencyLimit: 1 + tasks: + - id: check_row + type: io.kestra.plugin.core.execution.Assert + conditions: + - "{{ (fromJson(outputs.poll_batch.body).data.rows[item.value | number].status == 'completed' or fromJson(outputs.poll_batch.body).data.rows[item.value | number].status == 'failed' or fromJson(outputs.poll_batch.body).data.rows[item.value | number].status == 'cancelled') }}" + errorMessage: A terminal batch still contains an unfinished row. + - id: completed_row + type: io.kestra.plugin.core.flow.If + condition: "{{ fromJson(outputs.poll_batch.body).data.rows[item.value | number].status == 'completed' }}" + then: + - id: check_screening_id + type: io.kestra.plugin.core.execution.Assert + conditions: + - "{{ fromJson(outputs.poll_batch.body).data.rows[item.value | number].screening_id | toJson | jq('test(\"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$\")') | first }}" + - id: fetch_evidence + type: io.kestra.plugin.core.http.Request + method: GET + uri: "{{ vars.api_base }}/results/{{ fromJson(outputs.poll_batch.body).data.rows[item.value | number].screening_id }}/evidence" + timeout: PT20S + options: + followRedirects: false + allowFailed: false + timeout: + connectTimeout: PT5S + readIdleTimeout: PT10S + headers: + Authorization: Bearer {{ secret('SANCTIONSKIT_API_KEY') }} + Content-Type: application/json + - id: check_evidence + type: io.kestra.plugin.core.execution.Assert + conditions: + - "{{ outputs.fetch_evidence.code == 200 }}" + - "{{ fromJson(outputs.fetch_evidence.body).format == 'sanctionskit-evidence@1' }}" + - "{{ fromJson(outputs.fetch_evidence.body).result.id == fromJson(outputs.poll_batch.body).data.rows[item.value | number].screening_id }}" + - "{{ fromJson(outputs.fetch_evidence.body).result.environment == 'sandbox' }}" + - "{{ (fromJson(outputs.fetch_evidence.body).result.status == 'no_match' or fromJson(outputs.fetch_evidence.body).result.status == 'potential_match') }}" + - "{{ fromJson(outputs.fetch_evidence.body).retention == 'standard' }}" + - "{{ fromJson(outputs.fetch_evidence.body).retainedInputs == true }}" + - "{{ fromJson(outputs.fetch_evidence.body).result.versions.package == 'sandbox@1' }}" + - "{{ fromJson(outputs.fetch_evidence.body).result.status == 'no_match' ? fromJson(outputs.fetch_evidence.body).result.matches | length == 0 : fromJson(outputs.fetch_evidence.body).result.matches | length > 0 }}" + - "{{ fromJson(outputs.fetch_evidence.body).reference == vars.subjects[item.value | number].reference }}" + - "{{ fromJson(outputs.fetch_evidence.body).subject == vars.subjects[item.value | number].subject }}" + - "{{ fromJson(outputs.fetch_evidence.body).request == vars.subjects[item.value | number] }}" + errorMessage: Retained evidence does not match this sandbox batch row and fictional contact. + fetchType: FETCH + outputs: + - id: evidence + type: JSON + required: false + value: "{{ outputs.fetch_evidence is defined ? fromJson(outputs.fetch_evidence.body) : null }}" + - id: write_review_bundle + type: io.kestra.plugin.core.storage.Write + extension: .json + content: | + { + "synthetic":true, + "onboardingApproved":false, + "requestKey":{{ inputs.request_key | toJson }}, + "batch":{{ fromJson(outputs.poll_batch.body).data | toJson }}, + "evidenceByRow":{{ outputs.collect_evidence.outputs | toJson | jq('map(select(.outputs.evidence != null) | {key: ((.item.value | tonumber) + 1 | tostring), value: .outputs.evidence}) | from_entries') | first | toJson }}, + "nextStep":"Review each row and its evidence. A name-screening result does not approve onboarding." + } + - id: write_row_summary + type: io.kestra.plugin.core.storage.Write + extension: .json + content: | + {% set evidenceByRow = outputs.collect_evidence.outputs | toJson | jq('map(select(.outputs.evidence != null) | {key: ((.item.value | tonumber) + 1 | tostring), value: .outputs.evidence}) | from_entries') | first %} + [ + {% for row in fromJson(outputs.poll_batch.body).data.rows %} + {"row_number":{{ row.row_number }}, + "reference":{{ vars.subjects[row.row_number - 1].reference | toJson }}, + "row_status":{{ row.status | toJson }}, + "screening_id":{{ row.screening_id | toJson }}, + "screening_status":{{ (row.status == 'completed' ? evidenceByRow[row.row_number | string].result.status : '') | toJson }}, + "error_code":{{ (row.error == null ? '' : row.error.code) | toJson }}, + "synthetic":true,"onboarding_approved":false}{% if not loop.last %},{% endif %} + {% endfor %} + ] + - id: summary_to_ion + type: io.kestra.plugin.serdes.json.JsonToIon + from: "{{ outputs.write_row_summary.uri }}" + newLine: false + version: 2.0.6 + - id: summary_to_csv + type: io.kestra.plugin.serdes.csv.IonToCsv + from: "{{ outputs.summary_to_ion.uri }}" + header: true + version: 2.0.6 +outputs: + - id: review_bundle + type: FILE + value: "{{ outputs.write_review_bundle.uri | default(null) }}" + - id: row_summary + type: FILE + value: "{{ outputs.summary_to_csv.uri | default(null) }}" + - id: batch_status + type: STRING + value: "{{ outputs.poll_batch is defined ? fromJson(outputs.poll_batch.body).data.status : 'not_collected' }}" +triggers: + - id: optional_daily_example + type: io.kestra.plugin.core.trigger.Schedule + cron: 0 9 * * * + disabled: true + inputs: + synthetic: true + request_key: kestra-example-{{ trigger.date | date('yyyyMMdd') }} diff --git a/integrations/kestra/tests/Dockerfile b/integrations/kestra/tests/Dockerfile new file mode 100644 index 0000000..af792e2 --- /dev/null +++ b/integrations/kestra/tests/Dockerfile @@ -0,0 +1,15 @@ +FROM eclipse-temurin:25-jre-noble@sha256:398f810215757dc1926390014272579fb0e57c41ef1c8aa4f64ae761613a168b AS java +FROM python:3.13-slim-trixie@sha256:bb2988715db2cf7ace7b53f38f3cffbef7c7046a656bee66245eb0ed386e2e81 +COPY --from=java /opt/java/openjdk /opt/java/openjdk +ENV JAVA_HOME=/opt/java/openjdk +ENV PATH=/opt/java/openjdk/bin:$PATH +ENV PYTHONDONTWRITEBYTECODE=1 +ENV PYTHONUNBUFFERED=1 +COPY . /opt/example +RUN python /opt/example/tests/bootstrap.py +RUN java -jar /opt/kestra/kestra-2.0.4 plugins install --locally \ + -p /opt/kestra/plugins -c /opt/example/tests/local-kestra.yml \ + io.kestra.plugin:plugin-serdes:2.0.6 +RUN python -c "import hashlib,pathlib; p=pathlib.Path('/opt/kestra/plugins/io_kestra_plugin__plugin-serdes__2_0_6.jar'); assert hashlib.sha256(p.read_bytes()).hexdigest() == '1394bf5ec0e7a57e446adc2b326e4ac72651d3fe525853d4d1c92bef60fe117e'" +WORKDIR /work +ENTRYPOINT ["python", "/opt/example/tests/container_runner.py"] diff --git a/integrations/kestra/tests/bootstrap.py b/integrations/kestra/tests/bootstrap.py new file mode 100644 index 0000000..3c02dff --- /dev/null +++ b/integrations/kestra/tests/bootstrap.py @@ -0,0 +1,13 @@ +"""Download one pinned official Kestra runtime during image build.""" +import hashlib +from pathlib import Path +import urllib.request + +url = 'https://github.com/kestra-io/kestra/releases/download/v2.0.4/kestra-2.0.4' +expected = '88ee8d44c3d4a234c9d04689e72b8ba903ae8f9f2c6ae5f94bad89f3bcd68513' +path = Path('/opt/kestra/kestra-2.0.4') +path.parent.mkdir(parents=True, exist_ok=True) +with urllib.request.urlopen(url, timeout=60) as response, path.open('wb') as output: + while chunk := response.read(1024 * 1024): + output.write(chunk) +assert hashlib.sha256(path.read_bytes()).hexdigest() == expected, 'Runtime checksum mismatch' diff --git a/integrations/kestra/tests/container_runner.py b/integrations/kestra/tests/container_runner.py new file mode 100644 index 0000000..0a49c84 --- /dev/null +++ b/integrations/kestra/tests/container_runner.py @@ -0,0 +1,53 @@ +"""Run the engine and fixture tests inside a network-none disposable container.""" +import json +import os +from pathlib import Path +import shutil +import subprocess +import sys +import time +import urllib.request + +# A network-none Linux container has only loopback. Refuse any other interface. +interfaces = sorted(p.name for p in Path('/sys/class/net').iterdir()) +assert interfaces == ['lo'], f'Expected isolated loopback-only container, got {interfaces}' +assert os.geteuid() != 0, 'Run the test container as an unprivileged user' +root = Path('/work/example') +shutil.copytree('/opt/example', root) +work = Path('/work/runtime') +work.mkdir() +log = Path('/evidence/kestra-server.log') +config = root / 'tests/local-kestra.yml' +command = ['java', '-Xmx1g', '-Djava.awt.headless=true', '-Duser.home=/work/runtime', '-jar', '/opt/kestra/kestra-2.0.4', 'server', 'local', '-c', str(config), '-p', '/opt/kestra/plugins', '--port=28101', '--worker-thread=4', '--no-tutorials'] +env = dict(os.environ, SECRET_SANCTIONSKIT_API_KEY='bG9jYWwtbW9jay1vbmx5') +result = 1 +with log.open('w') as stream: + process = subprocess.Popen(command, cwd=work, env=env, stdout=stream, stderr=subprocess.STDOUT) + try: + deadline = time.monotonic() + 120 + while True: + assert process.poll() is None, 'Kestra exited before readiness' + assert time.monotonic() < deadline, 'Kestra readiness timed out' + try: + with urllib.request.urlopen('http://127.0.0.1:28101/ping', timeout=2) as response: + if response.status == 200: + break + except (OSError, urllib.error.URLError): + time.sleep(1) + result = subprocess.run([sys.executable, str(root / 'tests/native_test.py')], cwd=root, timeout=360).returncode + finally: + process.terminate() + try: + process.wait(timeout=30) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=5) + for name in ['tests/native-results.json', 'tests/last-failure.json', 'synthetic-review-bundle.json', 'synthetic-row-summary.csv']: + source = root / name + if source.exists(): + shutil.copy2(source, Path('/evidence') / source.name) + report = {'network_interfaces': interfaces, 'runtime': 'Kestra2.0.4', 'fixture_tests_exit': result, 'hosted_api_calls': 0, 'service_credentials_used': False} + Path('/evidence/container-boundary.json').write_text(json.dumps(report, indent=2) + '\n') + # The only key involved is fake, but logs still must not expose it. + assert 'local-mock-only' not in log.read_text(), 'Mock key leaked to server log' +sys.exit(result) diff --git a/integrations/kestra/tests/fixtures/synthetic-evidence.json b/integrations/kestra/tests/fixtures/synthetic-evidence.json new file mode 100644 index 0000000..de12c1f --- /dev/null +++ b/integrations/kestra/tests/fixtures/synthetic-evidence.json @@ -0,0 +1,142 @@ +{ + "format": "sanctionskit-evidence@1", + "result": { + "id": "00000000-0000-4000-8000-000000000001", + "status": "potential_match", + "matches": [ + { + "score": 96, + "record": { + "id": "sandbox:person-001", + "list": "Synthetic demonstration records", + "names": [ + { + "kind": "primary", + "value": "Alex Morgan" + }, + { + "kind": "alias", + "value": "\u0410\u043b\u0435\u043a\u0441 \u041c\u043e\u0440\u0433\u0430\u043d" + } + ], + "sourceId": "sandbox-synthetic", + "addresses": [], + "authority": "SanctionsKit synthetic examples", + "sourceUrl": "https://sanctionskit.com/demo", + "birthDates": [ + { + "value": "1984", + "original": "1984", + "precision": "year" + } + ], + "entityType": "person", + "extensions": { + "synthetic": true + }, + "identifiers": [ + { + "type": "passport", + "value": "DEMO-001", + "issuer": "Fictional authority" + } + ], + "designations": [ + { + "wording": "Invented example, not a government designation.", + "category": "other", + "programs": [ + "Synthetic demonstration only" + ], + "legalReferences": [] + } + ], + "nationalities": [] + }, + "evidence": [ + { + "field": "name", + "method": "normalized_name_exact", + "queryValue": "Alex Morgan", + "explanation": "Name similarity is a review signal, not a probability or finding of wrongdoing.", + "sourceValue": "Alex Morgan", + "contribution": 96 + } + ], + "conflicts": [], + "assessment": { + "name": { + "method": "normalized_name_exact", + "aliasKind": "primary", + "similarity": 96, + "aliasQuality": "unknown" + }, + "country": { + "status": "not_supplied" + }, + "birthDate": { + "status": "not_supplied" + }, + "alertBasis": "name", + "entityType": { + "status": "agreement" + }, + "reviewTier": "name_only", + "identifiers": { + "status": "not_supplied", + "matched": 0, + "conflicts": 0, + "issuerUnknown": 0, + "issuerAgreements": 0 + }, + "hasConflicts": false, + "sameNameCandidates": 1 + } + } + ], + "coverage": [ + { + "fresh": true, + "version": "1", + "sourceId": "sandbox-synthetic", + "retrievedAt": "2026-09-17T12:00:00.000Z" + } + ], + "matching": { + "profile": "standard", + "version": "review-1.2.0", + "nameThreshold": 78, + "retrievalVersion": "indexed-1.1.1", + "corroboratedNameThreshold": 75 + }, + "versions": { + "policy": "review-1.2.0", + "dataset": "sandbox-synthetic-v1", + "package": "sandbox@1", + "matchingEngine": "deterministic-1.2.0" + }, + "createdAt": "2026-09-17T12:00:00.000Z", + "disclaimer": "Potential matches support analyst review. No match means no match in the selected sources and dataset, not legal clearance. Scores and evidence categories support review; they are not probabilities of identity or wrongdoing.", + "environment": "sandbox" + }, + "retention": "standard", + "expires_at": null, + "subject": { + "name": "Alex Morgan", + "entityType": "person", + "birthDate": "1984" + }, + "reference": "example-customer-001", + "request": { + "subject": { + "name": "Alex Morgan", + "entityType": "person", + "birthDate": "1984" + }, + "package": "sandbox@1", + "reference": "example-customer-001", + "retention": "standard" + }, + "retainedInputs": true, + "replayLimit": null +} diff --git a/integrations/kestra/tests/local-kestra.yml b/integrations/kestra/tests/local-kestra.yml new file mode 100644 index 0000000..c35979e --- /dev/null +++ b/integrations/kestra/tests/local-kestra.yml @@ -0,0 +1,25 @@ +# Disposable local test instance. These are invented test credentials. +micronaut: + server: + host: 127.0.0.1 + port: 28101 + executors: + io: + type: fixed + nThreads: 8 +kestra: + anonymous-usage-report: + enabled: false + ui-anonymous-usage-report: + enabled: false + tutorial-flows: + enabled: false + plugins: + auto-install: + enabled: false + server: + basic-auth: + username: local@sanctionskit.invalid + password: LocalMockOnly123! + controller: + port: 28103 diff --git a/integrations/kestra/tests/native_test.py b/integrations/kestra/tests/native_test.py new file mode 100644 index 0000000..a86f463 --- /dev/null +++ b/integrations/kestra/tests/native_test.py @@ -0,0 +1,238 @@ +"""Execute the actual flow in a loopback-only Kestra 2.0.4 instance. + +Only fictional fixtures are sent to an in-process local HTTP server. This file +never contacts the hosted SanctionsKit API and never needs a real API key. +""" +import base64 +import copy +import csv +import io +import json +import os +from pathlib import Path +import re +import threading +import time +import urllib.error +import urllib.parse +import urllib.request +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +ROOT = Path(__file__).resolve().parents[1] +KESTRA = os.getenv('KESTRA_TEST_URL', 'http://127.0.0.1:28101') +assert urllib.parse.urlsplit(KESTRA).hostname in ('127.0.0.1', 'localhost') +AUTH = 'Basic ' + base64.b64encode(b'local@sanctionskit.invalid:LocalMockOnly123!').decode() +MOCK_KEY = 'local-mock-only' +BATCH = '11111111-1111-4111-8111-111111111111' +IDS = ['22222222-2222-4222-8222-222222222222', '33333333-3333-4333-8333-333333333333'] +SUBJECTS = [ + {'subject': {'name': 'Alex Morgan', 'entityType': 'person', 'birthDate': '1984'}, 'package': 'sandbox@1', 'retention': 'standard', 'reference': 'example-customer-001'}, + {'subject': {'name': 'Taylor Jordan', 'entityType': 'person', 'birthDate': '1979'}, 'package': 'sandbox@1', 'retention': 'standard', 'reference': 'example-customer-002'}, +] +TEMPLATE = json.loads((ROOT / 'tests/fixtures/synthetic-evidence.json').read_text()) + +class State: + mode = 'success' + calls = [] + polls = 0 + +class Mock(BaseHTTPRequestHandler): + def log_message(self, *_): + pass + + def respond(self, code, body): + self.send_response(code) + self.send_header('Content-Type', 'application/json') + self.end_headers() + self.wfile.write(json.dumps(body).encode()) + + def request_record(self): + assert self.headers.get('Authorization') == 'Bearer ' + MOCK_KEY + State.calls.append({'method': self.command, 'path': self.path}) + + def do_POST(self): + self.request_record() + assert self.path == '/api/v1/batches' + body = json.loads(self.rfile.read(int(self.headers['Content-Length']))) + assert body == {'name': 'Kestra fictional onboarding example', 'subjects': SUBJECTS} + assert self.headers['Idempotency-Key'] == 'native-test-' + State.mode + State.calls[-1]['idempotency_key'] = self.headers['Idempotency-Key'] + State.calls[-1]['body'] = body + if State.mode == 'rate_limited_once' and len(State.calls) == 1: + self.respond(429, {'error': {'code': 'synthetic_retry', 'message': 'Try once more'}}) + return + if State.mode in ('unauthorized', 'rate_limited', 'server_error', 'conflict'): + self.respond({'unauthorized': 401, 'rate_limited': 429, 'server_error': 503, 'conflict': 409}[State.mode], {'error': {'code': 'synthetic_test_error', 'message': 'Fictional test response'}}) + return + if State.mode == 'redirect': + self.send_response(302) + self.send_header('Location', 'http://127.0.0.1:28102/do-not-follow') + self.end_headers() + return + self.respond(202, {'data': {'id': '../not-an-id' if State.mode == 'bad_batch_id' else BATCH, 'status': 'pending', 'total': 2}}) + + def do_GET(self): + self.request_record() + if self.path.startswith('/api/v1/batches/'): + assert self.path == f'/api/v1/batches/{BATCH}?limit=100&offset=0' + State.polls += 1 + status = 'processing' if State.polls == 1 or State.mode == 'timeout' else 'completed' + rows = [{'row_number': i + 1, 'status': 'pending' if status == 'processing' else 'completed', 'screening_id': None if status == 'processing' else IDS[i], 'error': None} for i in range(2)] + if status == 'completed' and State.mode in ('mixed', 'all_failed', 'cancelled'): + indexes = [1] if State.mode == 'mixed' else [0, 1] + for i in indexes: + rows[i].update(status='cancelled' if State.mode == 'cancelled' else 'failed', screening_id=None, error={'code': 'synthetic_row_error', 'message': 'Fictional row outcome'}) + if State.mode == 'cancelled': status = 'cancelled' + if State.mode == 'all_failed': status = 'failed' + if status == 'completed' and State.mode == 'unfinished': + rows[1].update(status='pending', screening_id=None) + if status == 'completed' and State.mode == 'bad_uuid': rows[0]['screening_id'] = '../not-an-id' + if status == 'completed' and State.mode == 'row_order': rows.reverse() + self.respond(200, {'data': {'id': IDS[0] if State.mode == 'wrong_batch' else BATCH, 'name': 'Kestra fictional onboarding example', 'environment': 'production' if State.mode == 'wrong_environment' else 'sandbox', 'status': status, 'total': 2, 'completed': sum(r['status'] == 'completed' for r in rows) + (1 if State.mode == 'counter_mismatch' else 0), 'failed': sum(r['status'] == 'failed' for r in rows), 'created_at': '2026-01-01T00:00:00Z', 'cancelled_at': '2026-01-01T00:01:00Z' if State.mode == 'cancelled' else None, 'rows': rows, 'nextOffset': 2 if State.mode == 'extra_page' else None}}) + return + m = re.fullmatch('/api/v1/results/([a-f0-9-]+)/evidence', self.path) + assert m and m[1] in IDS + i = IDS.index(m[1]) + evidence = copy.deepcopy(TEMPLATE) + evidence['result']['id'] = IDS[i] + evidence['subject'] = SUBJECTS[i]['subject'] + evidence['reference'] = SUBJECTS[i]['reference'] + evidence['request'] = SUBJECTS[i] + if i == 1: + evidence['result'].update(status='no_match', matches=[]) + if State.mode == 'wrong_evidence': evidence['result']['id'] = BATCH + if State.mode == 'wrong_package': evidence['result']['versions']['package'] = 'wrong@1' + if State.mode == 'wrong_match_count': evidence['result']['matches'] = [] + if State.mode == 'wrong_contact': evidence['subject'] = {**evidence['subject'], 'birthDate': '1984-01-01'} + self.respond(200, evidence) + + +def api(path, data=None, content_type=None, method=None): + headers = {'Authorization': AUTH} + if content_type: headers['Content-Type'] = content_type + req = urllib.request.Request(KESTRA + '/api/v1/main' + path, data=data, headers=headers, method=method) + try: + with urllib.request.urlopen(req, timeout=15) as r: + payload = r.read() + return json.loads(payload) if 'json' in r.headers.get('Content-Type', '') else payload + except urllib.error.HTTPError as e: + raise RuntimeError(f'Local Kestra HTTP {e.code}: {e.read().decode()[:2000]}') from None + + +def execute(mode): + State.mode, State.calls, State.polls = mode, [], 0 + boundary = 'kestra-local-synthetic-test' + fields = {'request_key': 'native-test-' + mode, 'synthetic': 'false' if mode == 'no_confirmation' else 'true'} + body = ''.join(f'--{boundary}\r\nContent-Disposition: form-data; name="{key}"\r\n\r\n{value}\r\n' for key, value in fields.items()) + f'--{boundary}--\r\n' + execution = api('/executions/company.sanctionskit/sanctionskit-batch-evidence', body.encode(), 'multipart/form-data; boundary=' + boundary, 'POST') + execution_id = execution['id'] + deadline = time.monotonic() + 60 + while execution['state']['current'] not in ('SUCCESS', 'FAILED', 'KILLED', 'WARNING', 'CANCELLED'): + assert time.monotonic() < deadline, 'Native execution exceeded test deadline' + time.sleep(.25) + execution = api('/executions/' + execution_id) + return execution + + +def main(): + modes = os.getenv('KESTRA_TEST_CASES', 'success,rate_limited_once,mixed,all_failed,cancelled,no_confirmation,timeout,unauthorized,rate_limited,server_error,conflict,wrong_batch,wrong_environment,extra_page,row_order,unfinished,bad_uuid,wrong_evidence,wrong_contact,wrong_package,wrong_match_count,redirect,bad_batch_id,counter_mismatch').split(',') + flow = (ROOT / 'sanctionskit-batch-evidence.yaml').read_text() + # Only endpoint, polling interval and polling budget change for local tests. + # Request bodies, evidence checks, HTTP limits and exports are unchanged. + assert flow.count('https://www.sanctionskit.com/api/v1') == 1 + flow = flow.replace('https://www.sanctionskit.com/api/v1', 'http://127.0.0.1:28102/api/v1').replace('interval: PT2S', 'interval: PT0.1S').replace('maxIterations: 30', 'maxIterations: 3').replace('maxDuration: PT2M', 'maxDuration: PT10S') + exists = any(f['id'] == 'sanctionskit-batch-evidence' and f['namespace'] == 'company.sanctionskit' for f in api('/flows/search?' + urllib.parse.urlencode({'filters[namespace][EQUALS]': 'company.sanctionskit', 'size': 100}))['results']) + api('/flows/company.sanctionskit/sanctionskit-batch-evidence' if exists else '/flows', flow.encode(), 'application/x-yaml', 'PUT' if exists else 'POST') + server = ThreadingHTTPServer(('127.0.0.1', 28102), Mock) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + records = [] + try: + for mode in modes: + execution = execute(mode) + state = execution['state']['current'] + expected = 'SUCCESS' if mode in ('success', 'rate_limited_once', 'mixed', 'all_failed', 'cancelled') else 'FAILED' + record = {'case': mode, 'execution_id': execution['id'], 'state': state, 'expected': expected, 'calls': State.calls, 'polls': State.polls} + records.append(record) + if state != expected: + (ROOT / 'tests/last-failure.json').write_text(json.dumps(execution, indent=2)) + print(json.dumps(record), flush=True) + raise AssertionError(f'{mode}: expected {expected}, got {state}; see last-failure.json') + family = [execution] + if any(t['taskId'] == 'collect_evidence' for t in execution.get('taskRunList', [])): + query = urllib.parse.urlencode({'filters[parentId][EQUALS]': execution['id'], 'filters[kind][EQUALS]': 'LOOP', 'size': 100}) + children = api('/executions/search?' + query) + index_deadline = time.monotonic() + 5 + while not children['results'] and time.monotonic() < index_deadline: + time.sleep(.25) + children = api('/executions/search?' + query) + assert children['total'] <= 100, 'Unexpected iteration pagination' + for child in children['results']: + family.append(api('/executions/' + child['id'])) + assert len(family) > 1, 'Expected native Loop iteration executions' + record['loop_iterations_observed'] = len(family) - 1 + for member in family: + logs = api('/logs/' + member['id']) + assert MOCK_KEY not in json.dumps(logs), 'Mock key leaked into task logs' + assert MOCK_KEY not in json.dumps(member), 'Mock key leaked into execution metadata' + for task in member.get('taskRunList', []): + if task['state']['current'] == 'SUCCESS' and task['taskId'] in ('submit_batch', 'poll_batch', 'fetch_evidence'): + task_outputs = api('/outputs/tasks/' + member['id'] + '/' + task['id']) + assert MOCK_KEY not in json.dumps(task_outputs), 'Mock key leaked into HTTP outputs' + + if state == 'FAILED': + expected_task = { + 'no_confirmation': 'require_synthetic', 'timeout': 'wait_for_batch', + 'unauthorized': 'submit_batch', 'rate_limited': 'submit_batch', + 'server_error': 'submit_batch', 'conflict': 'submit_batch', + 'wrong_batch': 'check_poll_identity', 'wrong_environment': 'check_poll_identity', + 'extra_page': 'check_complete_page', 'row_order': 'check_complete_page', + 'unfinished': 'check_row', 'bad_uuid': 'check_screening_id', + 'wrong_evidence': 'check_evidence', 'wrong_contact': 'check_evidence', + 'wrong_package': 'check_evidence', 'wrong_match_count': 'check_evidence', + 'redirect': 'check_acceptance', 'bad_batch_id': 'check_acceptance', + 'counter_mismatch': 'check_complete_page', + }[mode] + failed_tasks = [t['taskId'] for member in family for t in member.get('taskRunList', []) if t['state']['current'] == 'FAILED'] + assert expected_task in failed_tasks, (mode, expected_task, failed_tasks) + record['intended_failure_task'] = expected_task + record['mock_key_absent_from_logs_and_outputs'] = True + if state == 'SUCCESS': + outputs = api('/outputs/executions/' + execution['id']) + assert MOCK_KEY not in json.dumps(outputs) + for key in ['review_bundle', 'row_summary']: + uri = outputs[key] + payload = api('/executions/' + execution['id'] + '/file?path=' + urllib.parse.quote(uri, safe='')) + assert MOCK_KEY not in payload.decode() + if key == 'review_bundle': + bundle = json.loads(payload) + assert bundle['synthetic'] is True and bundle['onboardingApproved'] is False + count = {'success': 2, 'rate_limited_once': 2, 'mixed': 1, 'all_failed': 0, 'cancelled': 0}[mode] + assert len(bundle['evidenceByRow']) == count + assert len(bundle['batch']['rows']) == 2 + for row, e in bundle['evidenceByRow'].items(): + index = int(row) - 1 + assert e['result']['id'] == IDS[index] + assert e['request'] == SUBJECTS[index] + if mode == 'success': (ROOT / 'synthetic-review-bundle.json').write_bytes(payload) + else: + rows = list(csv.DictReader(io.StringIO(payload.decode()))) + assert len(rows) == 2 and [r['reference'] for r in rows] == [s['reference'] for s in SUBJECTS] + if mode == 'success': (ROOT / 'synthetic-row-summary.csv').write_bytes(payload) + if mode == 'no_confirmation': assert not State.calls + if mode in ('unauthorized', 'conflict'): assert len(State.calls) == 1 + if mode in ('rate_limited', 'server_error'): assert len(State.calls) == 2 + if mode == 'rate_limited_once': + posts = [c for c in State.calls if c['method'] == 'POST'] + assert len(posts) == 2 and posts[0] == posts[1] + if mode in ('bad_uuid', 'extra_page', 'row_order', 'wrong_batch', 'wrong_environment', 'timeout'): assert not any('/results/' in c['path'] for c in State.calls) + if mode == 'timeout': assert State.polls == 3 + if mode in ('redirect', 'bad_batch_id'): assert len(State.calls) == 1 + if mode == 'counter_mismatch': assert not any('/results/' in c['path'] for c in State.calls) + print(f'{mode}: {state}, {len(State.calls)} mock requests', flush=True) + finally: + server.shutdown() + (ROOT / 'tests/native-results.json').write_text(json.dumps({'runtime': 'Kestra 2.0.4', 'plugin': 'io.kestra.plugin:plugin-serdes:2.0.6', 'java': 'Official Temurin 25 JRE image pinned by digest', 'boundary': 'Native engine against loopback fictional HTTP mock only', 'cases': records}, indent=2) + '\n') + +if __name__ == '__main__': + main() From 346a8ad88673dfcb13d870b066f196d0af3ba121 Mon Sep 17 00:00:00 2001 From: SanctionsKit Date: Sat, 3 Oct 2026 23:29:24 -0600 Subject: [PATCH 2/4] Resolve container hostname through isolated loopback Give the network-none fixture container a fixed hostname mapped to 127.0.0.1. Assert that resolution before starting Kestra and preserve the existing network and privilege limits. --- .github/workflows/kestra.yml | 1 + integrations/kestra/README.md | 5 ++++- integrations/kestra/tests/container_runner.py | 7 ++++++- 3 files changed, 11 insertions(+), 2 deletions(-) diff --git a/.github/workflows/kestra.yml b/.github/workflows/kestra.yml index 1559410..47bfc46 100644 --- a/.github/workflows/kestra.yml +++ b/.github/workflows/kestra.yml @@ -31,6 +31,7 @@ jobs: run: | mkdir -p test-results docker run --rm --network none --cap-drop ALL \ + --hostname kestra-fixture --add-host kestra-fixture:127.0.0.1 \ --security-opt no-new-privileges:true --read-only \ --memory 3g --cpus 2 --pids-limit 512 \ --user "$(id -u):$(id -g)" \ diff --git a/integrations/kestra/README.md b/integrations/kestra/README.md index 63f99f6..fc0061d 100644 --- a/integrations/kestra/README.md +++ b/integrations/kestra/README.md @@ -23,7 +23,9 @@ with a checked artifact hash. [runtime-lock.json](runtime-lock.json) records the sources. Network access is needed only while building the image. Execution uses Docker `--network none`, with no published ports or Docker socket. -The test runner refuses to start unless its only network interface is `lo`. It +The fixed hostname resolves to `127.0.0.1` through the container's hosts file, so +Java can identify the local host without DNS. The test runner checks that mapping +and refuses to start unless its only network interface is `lo`. It runs as an unprivileged user with all Linux capabilities dropped, no new privileges, a read-only root filesystem, and temporary writable directories. The only service key is the invented string `local-mock-only`; no GitHub secret or real API key is @@ -35,6 +37,7 @@ From this directory on a machine with Docker: docker build -f tests/Dockerfile -t sanctionskit-kestra-fixture . mkdir -p test-results docker run --rm --network none --cap-drop ALL \ + --hostname kestra-fixture --add-host kestra-fixture:127.0.0.1 \ --security-opt no-new-privileges:true --read-only \ --memory 3g --cpus 2 --pids-limit 512 \ --user "$(id -u):$(id -g)" \ diff --git a/integrations/kestra/tests/container_runner.py b/integrations/kestra/tests/container_runner.py index 0a49c84..328dd50 100644 --- a/integrations/kestra/tests/container_runner.py +++ b/integrations/kestra/tests/container_runner.py @@ -3,6 +3,7 @@ import os from pathlib import Path import shutil +import socket import subprocess import sys import time @@ -12,6 +13,10 @@ interfaces = sorted(p.name for p in Path('/sys/class/net').iterdir()) assert interfaces == ['lo'], f'Expected isolated loopback-only container, got {interfaces}' assert os.geteuid() != 0, 'Run the test container as an unprivileged user' +hostname = socket.gethostname() +assert hostname == 'kestra-fixture', 'Run the test container with --hostname kestra-fixture' +hostname_addresses = sorted({address[4][0] for address in socket.getaddrinfo(hostname, None, type=socket.SOCK_STREAM)}) +assert hostname_addresses == ['127.0.0.1'], f'Expected hostname to resolve only to loopback, got {hostname_addresses}' root = Path('/work/example') shutil.copytree('/opt/example', root) work = Path('/work/runtime') @@ -46,7 +51,7 @@ source = root / name if source.exists(): shutil.copy2(source, Path('/evidence') / source.name) - report = {'network_interfaces': interfaces, 'runtime': 'Kestra2.0.4', 'fixture_tests_exit': result, 'hosted_api_calls': 0, 'service_credentials_used': False} + report = {'network_interfaces': interfaces, 'hostname': hostname, 'hostname_addresses': hostname_addresses, 'runtime': 'Kestra2.0.4', 'fixture_tests_exit': result, 'hosted_api_calls': 0, 'service_credentials_used': False} Path('/evidence/container-boundary.json').write_text(json.dumps(report, indent=2) + '\n') # The only key involved is fake, but logs still must not expose it. assert 'local-mock-only' not in log.read_text(), 'Mock key leaked to server log' From 3f9c56d940e80598c7dec22863637faeda1c10ba Mon Sep 17 00:00:00 2001 From: SanctionsKit Date: Sat, 3 Oct 2026 23:53:35 -0600 Subject: [PATCH 3/4] Preserve empty evidence exports for failed batch rows Treat an absent Loop output collection as empty when every row has failed or been cancelled. Keep row errors in the exported review bundle and add bounded fixture diagnostics for unexpected native failures. --- integrations/kestra/.gitignore | 1 + integrations/kestra/README.md | 4 +- .../kestra/sanctionskit-batch-evidence.yaml | 4 +- integrations/kestra/tests/container_runner.py | 2 +- integrations/kestra/tests/native_test.py | 47 ++++++++++++++++++- 5 files changed, 52 insertions(+), 6 deletions(-) diff --git a/integrations/kestra/.gitignore b/integrations/kestra/.gitignore index ac6495d..0f67333 100644 --- a/integrations/kestra/.gitignore +++ b/integrations/kestra/.gitignore @@ -1,6 +1,7 @@ __pycache__/ tests/native-results.json tests/last-failure.json +tests/last-failure-diagnostics.json test-results/ synthetic-review-bundle.json synthetic-row-summary.csv diff --git a/integrations/kestra/README.md b/integrations/kestra/README.md index fc0061d..c25dafa 100644 --- a/integrations/kestra/README.md +++ b/integrations/kestra/README.md @@ -56,7 +56,9 @@ iteration executions, uses the 2.0 outputs API, downloads the JSON/CSV files, an checks logs and outputs for the fake key. The server log, per-case results, isolation report and generated files go into -`test-results/`. A parser or import pass is not a substitute for this run. +`test-results/`. A failed assertion also saves bounded parent and Loop task logs +and outputs, with fixture credentials redacted. A parser or import pass is not +a substitute for this run. ## Intended sandbox setup after qualification diff --git a/integrations/kestra/sanctionskit-batch-evidence.yaml b/integrations/kestra/sanctionskit-batch-evidence.yaml index 374d793..a4d5abf 100644 --- a/integrations/kestra/sanctionskit-batch-evidence.yaml +++ b/integrations/kestra/sanctionskit-batch-evidence.yaml @@ -168,14 +168,14 @@ tasks: "onboardingApproved":false, "requestKey":{{ inputs.request_key | toJson }}, "batch":{{ fromJson(outputs.poll_batch.body).data | toJson }}, - "evidenceByRow":{{ outputs.collect_evidence.outputs | toJson | jq('map(select(.outputs.evidence != null) | {key: ((.item.value | tonumber) + 1 | tostring), value: .outputs.evidence}) | from_entries') | first | toJson }}, + "evidenceByRow":{{ (outputs.collect_evidence.outputs | default([])) | toJson | jq('map(select(.outputs.evidence != null) | {key: ((.item.value | tonumber) + 1 | tostring), value: .outputs.evidence}) | from_entries') | first | toJson }}, "nextStep":"Review each row and its evidence. A name-screening result does not approve onboarding." } - id: write_row_summary type: io.kestra.plugin.core.storage.Write extension: .json content: | - {% set evidenceByRow = outputs.collect_evidence.outputs | toJson | jq('map(select(.outputs.evidence != null) | {key: ((.item.value | tonumber) + 1 | tostring), value: .outputs.evidence}) | from_entries') | first %} + {% set evidenceByRow = (outputs.collect_evidence.outputs | default([])) | toJson | jq('map(select(.outputs.evidence != null) | {key: ((.item.value | tonumber) + 1 | tostring), value: .outputs.evidence}) | from_entries') | first %} [ {% for row in fromJson(outputs.poll_batch.body).data.rows %} {"row_number":{{ row.row_number }}, diff --git a/integrations/kestra/tests/container_runner.py b/integrations/kestra/tests/container_runner.py index 328dd50..405d387 100644 --- a/integrations/kestra/tests/container_runner.py +++ b/integrations/kestra/tests/container_runner.py @@ -47,7 +47,7 @@ except subprocess.TimeoutExpired: process.kill() process.wait(timeout=5) - for name in ['tests/native-results.json', 'tests/last-failure.json', 'synthetic-review-bundle.json', 'synthetic-row-summary.csv']: + for name in ['tests/native-results.json', 'tests/last-failure.json', 'tests/last-failure-diagnostics.json', 'synthetic-review-bundle.json', 'synthetic-row-summary.csv']: source = root / name if source.exists(): shutil.copy2(source, Path('/evidence') / source.name) diff --git a/integrations/kestra/tests/native_test.py b/integrations/kestra/tests/native_test.py index a86f463..1b294a0 100644 --- a/integrations/kestra/tests/native_test.py +++ b/integrations/kestra/tests/native_test.py @@ -107,18 +107,52 @@ def do_GET(self): self.respond(200, evidence) -def api(path, data=None, content_type=None, method=None): +def api(path, data=None, content_type=None, method=None, timeout=15): headers = {'Authorization': AUTH} if content_type: headers['Content-Type'] = content_type req = urllib.request.Request(KESTRA + '/api/v1/main' + path, data=data, headers=headers, method=method) try: - with urllib.request.urlopen(req, timeout=15) as r: + with urllib.request.urlopen(req, timeout=timeout) as r: payload = r.read() return json.loads(payload) if 'json' in r.headers.get('Content-Type', '') else payload except urllib.error.HTTPError as e: raise RuntimeError(f'Local Kestra HTTP {e.code}: {e.read().decode()[:2000]}') from None +def capture_failure(execution): + """Save bounded local task diagnostics without replacing the test failure.""" + def fetch(path): + try: + return api(path, timeout=3) + except Exception as error: + return {'diagnostic_error_type': type(error).__name__} + + family = [execution] + query = urllib.parse.urlencode({'filters[parentId][EQUALS]': execution['id'], 'filters[kind][EQUALS]': 'LOOP', 'size': 3}) + children = fetch('/executions/search?' + query) + # The fixed fixture has two rows. Bound diagnostics even if an engine is faulty. + for child in children.get('results', [])[:2]: + member = fetch('/executions/' + child['id']) + if 'id' in member: + family.append(member) + snapshots = [] + for member in family: + task_outputs = {} + tasks = [t for t in member.get('taskRunList', []) if t['taskId'] == 'collect_evidence' or t['state']['current'] == 'FAILED'] + for task in tasks[:3]: + task_outputs[task['id']] = fetch('/outputs/tasks/' + member['id'] + '/' + task['id']) + snapshots.append({'execution': member, 'logs': fetch('/logs/' + member['id']), 'task_outputs': task_outputs}) + serialized = json.dumps({'case': State.mode, 'loop_search': children, 'snapshots': snapshots}, indent=2) + # These are invented fixture credentials, but diagnostics should not expose them. + fixture_credentials = [MOCK_KEY, AUTH, AUTH.removeprefix('Basic '), 'LocalMockOnly123!'] + credentials_seen = any(value in serialized for value in fixture_credentials) + for value in fixture_credentials: + serialized = serialized.replace(value, '[fixture credential redacted]') + diagnostics = json.loads(serialized) + diagnostics['fixture_credentials_redacted'] = credentials_seen + (ROOT / 'tests/last-failure-diagnostics.json').write_text(json.dumps(diagnostics, indent=2) + '\n') + + def execute(mode): State.mode, State.calls, State.polls = mode, [], 0 boundary = 'kestra-local-synthetic-test' @@ -147,8 +181,10 @@ def main(): thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() records = [] + execution = None try: for mode in modes: + execution = None execution = execute(mode) state = execution['state']['current'] expected = 'SUCCESS' if mode in ('success', 'rate_limited_once', 'mixed', 'all_failed', 'cancelled') else 'FAILED' @@ -230,6 +266,13 @@ def main(): if mode in ('redirect', 'bad_batch_id'): assert len(State.calls) == 1 if mode == 'counter_mismatch': assert not any('/results/' in c['path'] for c in State.calls) print(f'{mode}: {state}, {len(State.calls)} mock requests', flush=True) + except Exception: + if execution is not None: + try: + capture_failure(execution) + except Exception as error: + print(f'Failure diagnostics unavailable: {type(error).__name__}', flush=True) + raise finally: server.shutdown() (ROOT / 'tests/native-results.json').write_text(json.dumps({'runtime': 'Kestra 2.0.4', 'plugin': 'io.kestra.plugin:plugin-serdes:2.0.6', 'java': 'Official Temurin 25 JRE image pinned by digest', 'boundary': 'Native engine against loopback fictional HTTP mock only', 'cases': records}, indent=2) + '\n') From da608596f4e3a58abada98f38f5e5ab278e544ca Mon Sep 17 00:00:00 2001 From: SanctionsKit Date: Tue, 6 Oct 2026 19:06:40 -0600 Subject: [PATCH 4/4] Publish qualified Kestra example documentation Link the Kestra example from the root README and document the verified 24-scenario native fixture run. Keep hosted sandbox execution distinct from the isolated test results. --- README.md | 4 +++- integrations/kestra/README.md | 26 +++++++++++--------------- 2 files changed, 14 insertions(+), 16 deletions(-) diff --git a/README.md b/README.md index f3c3a66..5d6641f 100644 --- a/README.md +++ b/README.md @@ -34,6 +34,8 @@ npm --prefix integrations/windmill test [Node-RED](integrations/node-red/) includes an importable flow built with core nodes. Run a fictional screening, retrieve its evidence, and inspect a review record. It keeps credentials in the process environment and reuses a stable request key for manual retries. The README includes setup instructions and local runtime tests. +[Kestra](integrations/kestra/) submits a fictional batch, waits for its final state, and exports retained evidence and a CSV row summary. It keeps failed and cancelled rows visible for review. The Kestra 2.0.4 flow passed 24 native scenarios against a local HTTP fixture in an isolated container. + ## Make your first request 1. [Create a workspace](https://www.sanctionskit.com/signup), then open [API keys](https://www.sanctionskit.com/dashboard/keys?environment=sandbox). @@ -87,7 +89,7 @@ These scripts display responses for learning. When adapting them to an applicati `REQUEST_KEY` is sent as the `Idempotency-Key` header. Generate it once for a new screening. If a request times out or needs a retry, keep **the same key and the same body**. Rerun the script without repeating the key-generation command. A new subject or changed body needs a new key. -The examples make one attempt and exit with a nonzero status on an HTTP or network error. They do not retry automatically. An error is an incomplete request, never a `no_match` result. See [idempotency](https://www.sanctionskit.com/docs/idempotency) and [error handling](https://www.sanctionskit.com/docs/errors) before adding retries to an application. +The standalone language examples make one attempt and exit with a nonzero status on an HTTP or network error. They do not retry automatically. An error is an incomplete request, never a `no_match` result. See [idempotency](https://www.sanctionskit.com/docs/idempotency) and [error handling](https://www.sanctionskit.com/docs/errors) before adding retries to an application. ## Adapt the example diff --git a/integrations/kestra/README.md b/integrations/kestra/README.md index c25dafa..a0592c5 100644 --- a/integrations/kestra/README.md +++ b/integrations/kestra/README.md @@ -1,19 +1,15 @@ -# Kestra 2.0 batch-screening candidate +# Kestra batch screening and evidence -This is a candidate for native fixture testing on Kestra 2.0.4. It has passed the -2.0.4 server's flow import validation. It has not yet passed native execution on -that version, so it is not ready to use as a published integration. +This Kestra 2.0.4 flow submits two fictional contacts to the SanctionsKit synthetic +sandbox, waits for the batch to finish, and collects retained evidence with a CSV +row summary. Completed, failed and cancelled rows stay visible. A successful +collection does not approve onboarding or mean that every row screened successfully. -The flow submits two fictional contacts to the SanctionsKit synthetic sandbox, -waits for the batch to finish, and collects retained evidence with a CSV row -summary. Completed, failed and cancelled rows stay visible. A successful collection -does not approve onboarding or mean that every row screened successfully. - -The maintenance-version implementation passed 24 native mock scenarios on Kestra -1.3.41. That result does not establish 2.0 compatibility. This candidate replaces -`ForEach` with `Loop`, uses explicit per-iteration outputs, removes `pluginDefaults`, -updates the boolean input and HTTP timeout fields, and uses the 2.0 output API in -the tests. Native execution is the next gate. +All 24 native fixture scenarios passed on Kestra 2.0.4 with serdes 2.0.6 and +Temurin 25 in an isolated container. The [verified CI run](https://github.com/SanctionsKit/sanctions-kit-examples/actions/runs/37181224466) +used a local fictional HTTP server with Docker networking disabled. This verifies +the workflow and its error handling against fixtures; a hosted sandbox run remains +separate. ## Isolated fixture test @@ -60,7 +56,7 @@ The server log, per-case results, isolation report and generated files go into and outputs, with fixture credentials redacted. A parser or import pass is not a substitute for this run. -## Intended sandbox setup after qualification +## Sandbox setup Create a free [developer workspace](https://www.sanctionskit.com/signup?workflow=api) and a sandbox key with `batches:write` and `results:read`. Save the key as the Kestra