New to Kestra?
Use blueprints to kickstart your first workflows.
TenantShield, Multi-Tenant RAG Security Tester
Test RAG tenant isolation with Qdrant canaries, PostgreSQL evidence, parallel Python probes, Slack alerts, and a failing deployment gate.
Categories
AI
id: tenantshield-multi-tenant-rag-security-tester
namespace: company.ai
description: |
Seed tenant-labelled canary documents into Qdrant, exercise the retrieval boundary
concurrently as two tenants, persist evidence in PostgreSQL, and fail closed when
any query returns another tenant's data.
concurrency:
limit: 1
behavior: QUEUE
inputs:
- id: demo_mode
type: BOOL
defaults: true
description: Start disposable Qdrant and PostgreSQL containers with
deterministic fixtures. Set false to test production endpoints.
- id: inject_leak
type: BOOL
defaults: false
description: Deliberately omit the tenant filter from beta probes to prove the
deployment gate detects and blocks a real cross-tenant retrieval.
- id: qdrant_host
type: STRING
defaults: qdrant.internal
description: Production Qdrant gRPC host. Ignored in demo mode.
- id: qdrant_port
type: INT
defaults: 6334
description: Production Qdrant gRPC port. Ignored in demo mode.
- id: qdrant_tls
type: BOOL
defaults: true
description: Enable TLS for the production Qdrant connection. Ignored in demo mode.
- id: postgres_url
type: STRING
defaults: jdbc:postgresql://postgres.internal:5432/security
description: Production PostgreSQL JDBC URL for the audit ledger. Ignored in demo mode.
- id: collection_prefix
type: STRING
defaults: tenantshield-fixture
description: Prefix for the execution-scoped Qdrant collection; the execution ID
is appended to prevent collisions.
- id: max_results
type: INT
defaults: 3
description: Maximum nearest-neighbour results inspected per canary query. Must
be between 1 and 10.
- id: notify_slack
type: BOOL
defaults: false
description: Send a Slack incident when the gate or its infrastructure fails.
Requires SLACK_WEBHOOK_URL when true.
variables:
collection_name: "{{ inputs.collection_prefix }}-{{ execution.id | lower }}"
resolved_qdrant_host: "{{ inputs.demo_mode ? 'host.docker.internal' : inputs.qdrant_host }}"
resolved_qdrant_port: "{{ inputs.demo_mode ? 16334 : inputs.qdrant_port }}"
resolved_qdrant_tls: "{{ inputs.demo_mode ? false : inputs.qdrant_tls }}"
resolved_qdrant_key: "{{ inputs.demo_mode ? 'tenantshield-demo-key' : secret('QDRANT_API_KEY') }}"
resolved_postgres_url: "{{ inputs.demo_mode ?
'jdbc:postgresql://host.docker.internal:15432/tenantshield' :
inputs.postgres_url }}"
resolved_postgres_user: "{{ inputs.demo_mode ? 'tenantshield' : secret('POSTGRES_USERNAME') }}"
resolved_postgres_password: "{{ inputs.demo_mode ? 'tenantshield-demo-password'
: secret('POSTGRES_PASSWORD') }}"
demo_compose: |
services:
qdrant:
image: qdrant/qdrant:v1.19.2
environment:
QDRANT__SERVICE__API_KEY: tenantshield-demo-key
ports:
- "127.0.0.1:16333:6333"
- "127.0.0.1:16334:6334"
postgres:
image: postgres:17.6-alpine
environment:
POSTGRES_DB: tenantshield
POSTGRES_USER: tenantshield
POSTGRES_PASSWORD: tenantshield-demo-password
ports:
- "127.0.0.1:15432:5432"
healthcheck:
test: ["CMD-SHELL", "pg_isready -U tenantshield -d tenantshield"]
interval: 2s
timeout: 2s
retries: 30
triggers:
- id: nightly_isolation_regression
type: io.kestra.plugin.core.trigger.Schedule
description: Disabled-by-default regression run; enable only after production
secrets and endpoints are configured.
cron: "0 2 * * *"
timezone: UTC
disabled: true
inputs:
demo_mode: false
inject_leak: false
notify_slack: true
tasks:
- id: validate_inputs
type: io.kestra.plugin.core.flow.If
description: Reject unsafe or meaningless fixture settings before creating a
collection or container.
condition: "{{ inputs.collection_prefix is empty or inputs.max_results < 1 or
inputs.max_results > 10 }}"
then:
- id: reject_invalid_inputs
type: io.kestra.plugin.core.execution.Fail
errorMessage: collection_prefix must be non-empty and max_results must be
between 1 and 10.
- id: provision_demo_services
type: io.kestra.plugin.core.flow.If
description: Start isolated, disposable services only for the deterministic
local demonstration.
condition: "{{ inputs.demo_mode }}"
then:
- id: compose_up
type: io.kestra.plugin.docker.cli.Compose
description: Start pinned Qdrant and PostgreSQL containers under an
execution-scoped Compose project.
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
volumes:
- /var/run/docker.sock:/var/run/docker.sock
composeFile: "{{ render(vars.demo_compose) }}"
composeArgs:
- --project-name
- "tenantshield-{{ execution.id | lower }}"
- up
- -d
- --wait
timeout: PT10M
retry:
type: exponential
interval: PT2S
maxInterval: PT10S
maxAttempts: 3
- id: prepare_audit_ledger
type: io.kestra.plugin.jdbc.postgresql.Queries
description: Create the append-safe PostgreSQL audit table used to retain the
exact verdict and both tenant probe reports.
url: "{{ render(vars.resolved_postgres_url) }}"
username: "{{ render(vars.resolved_postgres_user) }}"
password: "{{ render(vars.resolved_postgres_password) }}"
fetchType: NONE
sql: |
CREATE SCHEMA IF NOT EXISTS tenantshield;
CREATE TABLE IF NOT EXISTS tenantshield.isolation_audits (
execution_id text PRIMARY KEY,
collection_name text NOT NULL,
injected_leak boolean NOT NULL,
alpha_result jsonb NOT NULL,
beta_result jsonb NOT NULL,
leak_count integer NOT NULL CHECK (leak_count >= 0),
verdict text NOT NULL CHECK (verdict IN ('PASS', 'BLOCK')),
created_at timestamptz NOT NULL DEFAULT now()
);
timeout: PT1M
retry:
type: exponential
interval: PT2S
maxInterval: PT10S
maxAttempts: 5
- id: build_fixture_vectors
type: io.kestra.plugin.scripts.python.Script
description: Generate deterministic 32-dimensional hashing-trick embeddings and
four tenant-labelled canary points without calling an external model.
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
containerImage: python:3.12.7-slim
dependencies:
- kestra==0.23.0
outputFiles:
- fixture-points.json
timeout: PT5M
script: |
import hashlib
import json
import math
import re
from kestra import Kestra
DIMENSIONS = 32
documents = [
(101, "alpha", "alpha-atlas", "Project Atlas launch code is ALPHA-ORCHID-731."),
(102, "alpha", "alpha-runbook", "Alpha support escalation uses the violet runbook."),
(201, "beta", "beta-borealis", "Project Borealis launch code is BETA-CEDAR-284."),
(202, "beta", "beta-runbook", "Beta support escalation uses the amber runbook."),
]
def embed(text):
vector = [0.0] * DIMENSIONS
for token in re.findall(r"[a-z0-9]+", text.lower()):
digest = hashlib.blake2b(token.encode(), digest_size=8).digest()
index = int.from_bytes(digest[:4], "big") % DIMENSIONS
vector[index] += 1.0 if digest[4] & 1 else -1.0
norm = math.sqrt(sum(value * value for value in vector)) or 1.0
return [round(value / norm, 8) for value in vector]
points = [
{
"id": point_id,
"vector": embed(text),
"payload": {"tenant_id": tenant, "doc_id": doc_id, "text": text},
}
for point_id, tenant, doc_id, text in documents
]
with open("fixture-points.json", "w", encoding="utf-8") as fixture_file:
json.dump(points, fixture_file, indent=2)
Kestra.outputs({"points_json": json.dumps(points), "point_count": len(points), "dimensions": DIMENSIONS})
- id: create_fixture_collection
type: io.kestra.plugin.qdrant.CreateCollection
description: Create a unique Qdrant collection for this execution using cosine
distance over the deterministic embeddings.
host: "{{ render(vars.resolved_qdrant_host) }}"
port: "{{ render(vars.resolved_qdrant_port) }}"
apiKey: "{{ render(vars.resolved_qdrant_key) }}"
tlsEnabled: "{{ render(vars.resolved_qdrant_tls) }}"
collectionName: "{{ render(vars.collection_name) }}"
vectorSize: 32
distance: COSINE
timeout: PT1M
retry:
type: exponential
interval: PT2S
maxInterval: PT10S
maxAttempts: 5
- id: inject_tenant_documents
type: io.kestra.plugin.qdrant.Upsert
description: Upsert the real alpha and beta vectors with tenant_id payloads into
Qdrant and wait for acknowledgement.
host: "{{ render(vars.resolved_qdrant_host) }}"
port: "{{ render(vars.resolved_qdrant_port) }}"
apiKey: "{{ render(vars.resolved_qdrant_key) }}"
tlsEnabled: "{{ render(vars.resolved_qdrant_tls) }}"
collectionName: "{{ render(vars.collection_name) }}"
batchSize: 4
points: "{{ fromJson(outputs.build_fixture_vectors.vars.points_json) }}"
timeout: PT1M
retry:
type: exponential
interval: PT2S
maxInterval: PT10S
maxAttempts: 5
- id: probe_tenant_boundaries
type: io.kestra.plugin.core.flow.Parallel
description: Query the same Qdrant collection concurrently as alpha and beta,
including an adversarial query for the peer tenant's exact canary.
concurrent: 2
tasks:
- id: probe_alpha
type: io.kestra.plugin.scripts.python.Script
description: Verify every alpha retrieval is payload-filtered to alpha, even
when the query exactly matches beta's secret canary.
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
extraHosts:
- host.docker.internal:host-gateway
containerImage: python:3.12.7-slim
dependencies:
- kestra==0.23.0
- qdrant-client==1.19.1
env:
QDRANT_HOST: "{{ render(vars.resolved_qdrant_host) }}"
QDRANT_PORT: "{{ render(vars.resolved_qdrant_port) }}"
QDRANT_TLS: "{{ render(vars.resolved_qdrant_tls) }}"
QDRANT_API_KEY: "{{ render(vars.resolved_qdrant_key) }}"
COLLECTION_NAME: "{{ render(vars.collection_name) }}"
REQUESTING_TENANT: alpha
OWN_QUERY: Project Atlas launch code is ALPHA-ORCHID-731.
PEER_QUERY: Project Borealis launch code is BETA-CEDAR-284.
DROP_FILTER: "false"
MAX_RESULTS: "{{ inputs.max_results }}"
timeout: PT5M
retry:
type: exponential
interval: PT2S
maxInterval: PT10S
maxAttempts: 3
script: |
import hashlib
import json
import math
import os
import re
from kestra import Kestra
from qdrant_client import QdrantClient, models
def embed(text, dimensions=32):
vector = [0.0] * dimensions
for token in re.findall(r"[a-z0-9]+", text.lower()):
digest = hashlib.blake2b(token.encode(), digest_size=8).digest()
index = int.from_bytes(digest[:4], "big") % dimensions
vector[index] += 1.0 if digest[4] & 1 else -1.0
norm = math.sqrt(sum(value * value for value in vector)) or 1.0
return [round(value / norm, 8) for value in vector]
tenant = os.environ["REQUESTING_TENANT"]
drop_filter = os.environ["DROP_FILTER"].lower() == "true"
query_filter = None if drop_filter else models.Filter(
must=[models.FieldCondition(key="tenant_id", match=models.MatchValue(value=tenant))]
)
client = QdrantClient(
host=os.environ["QDRANT_HOST"],
grpc_port=int(os.environ["QDRANT_PORT"]),
prefer_grpc=True,
https=os.environ["QDRANT_TLS"].lower() == "true",
api_key=os.environ["QDRANT_API_KEY"],
timeout=15,
)
probes = []
leaked_ids = set()
for case, query in (("own_canary", os.environ["OWN_QUERY"]), ("peer_canary", os.environ["PEER_QUERY"])):
points = client.query_points(
collection_name=os.environ["COLLECTION_NAME"],
query=embed(query),
query_filter=query_filter,
limit=int(os.environ["MAX_RESULTS"]),
with_payload=True,
).points
hits = []
for point in points:
payload = point.payload or {}
hit = {"doc_id": payload.get("doc_id"), "tenant_id": payload.get("tenant_id"), "score": round(point.score, 6)}
hits.append(hit)
if payload.get("tenant_id") != tenant:
leaked_ids.add(str(payload.get("doc_id")))
if not hits:
raise RuntimeError(f"{tenant} probe returned no documents for {case}")
probes.append({"case": case, "filter_applied": not drop_filter, "hits": hits})
result = {"tenant": tenant, "status": "LEAK" if leaked_ids else "PASS", "leaked_ids": sorted(leaked_ids), "probes": probes}
Kestra.outputs({"result_json": json.dumps(result), "leak_count": len(leaked_ids), "status": result["status"]})
- id: probe_beta
type: io.kestra.plugin.scripts.python.Script
description: Verify beta isolation; inject_leak intentionally removes this
filter so the negative-control run must be blocked.
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
extraHosts:
- host.docker.internal:host-gateway
containerImage: python:3.12.7-slim
dependencies:
- kestra==0.23.0
- qdrant-client==1.19.1
env:
QDRANT_HOST: "{{ render(vars.resolved_qdrant_host) }}"
QDRANT_PORT: "{{ render(vars.resolved_qdrant_port) }}"
QDRANT_TLS: "{{ render(vars.resolved_qdrant_tls) }}"
QDRANT_API_KEY: "{{ render(vars.resolved_qdrant_key) }}"
COLLECTION_NAME: "{{ render(vars.collection_name) }}"
REQUESTING_TENANT: beta
OWN_QUERY: Project Borealis launch code is BETA-CEDAR-284.
PEER_QUERY: Project Atlas launch code is ALPHA-ORCHID-731.
DROP_FILTER: "{{ inputs.inject_leak }}"
MAX_RESULTS: "{{ inputs.max_results }}"
timeout: PT5M
retry:
type: exponential
interval: PT2S
maxInterval: PT10S
maxAttempts: 3
script: |
import hashlib
import json
import math
import os
import re
from kestra import Kestra
from qdrant_client import QdrantClient, models
def embed(text, dimensions=32):
vector = [0.0] * dimensions
for token in re.findall(r"[a-z0-9]+", text.lower()):
digest = hashlib.blake2b(token.encode(), digest_size=8).digest()
index = int.from_bytes(digest[:4], "big") % dimensions
vector[index] += 1.0 if digest[4] & 1 else -1.0
norm = math.sqrt(sum(value * value for value in vector)) or 1.0
return [round(value / norm, 8) for value in vector]
tenant = os.environ["REQUESTING_TENANT"]
drop_filter = os.environ["DROP_FILTER"].lower() == "true"
query_filter = None if drop_filter else models.Filter(
must=[models.FieldCondition(key="tenant_id", match=models.MatchValue(value=tenant))]
)
client = QdrantClient(
host=os.environ["QDRANT_HOST"],
grpc_port=int(os.environ["QDRANT_PORT"]),
prefer_grpc=True,
https=os.environ["QDRANT_TLS"].lower() == "true",
api_key=os.environ["QDRANT_API_KEY"],
timeout=15,
)
probes = []
leaked_ids = set()
for case, query in (("own_canary", os.environ["OWN_QUERY"]), ("peer_canary", os.environ["PEER_QUERY"])):
points = client.query_points(
collection_name=os.environ["COLLECTION_NAME"],
query=embed(query),
query_filter=query_filter,
limit=int(os.environ["MAX_RESULTS"]),
with_payload=True,
).points
hits = []
for point in points:
payload = point.payload or {}
hit = {"doc_id": payload.get("doc_id"), "tenant_id": payload.get("tenant_id"), "score": round(point.score, 6)}
hits.append(hit)
if payload.get("tenant_id") != tenant:
leaked_ids.add(str(payload.get("doc_id")))
if not hits:
raise RuntimeError(f"{tenant} probe returned no documents for {case}")
probes.append({"case": case, "filter_applied": not drop_filter, "hits": hits})
result = {"tenant": tenant, "status": "LEAK" if leaked_ids else "PASS", "leaked_ids": sorted(leaked_ids), "probes": probes}
Kestra.outputs({"result_json": json.dumps(result), "leak_count": len(leaked_ids), "status": result["status"]})
- id: assemble_security_report
type: io.kestra.plugin.scripts.python.Script
description: Consolidate both independent probe outputs into a machine-readable
verdict and a human-readable evidence artifact.
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
containerImage: python:3.12.7-slim
dependencies:
- kestra==0.23.0
env:
EXECUTION_ID: "{{ execution.id }}"
COLLECTION_NAME: "{{ render(vars.collection_name) }}"
INJECTED_LEAK: "{{ inputs.inject_leak }}"
ALPHA_RESULT: "{{ outputs.probe_alpha.vars.result_json }}"
BETA_RESULT: "{{ outputs.probe_beta.vars.result_json }}"
outputFiles:
- tenantshield-report.md
- tenantshield-results.json
timeout: PT5M
script: |
import json
import os
from kestra import Kestra
alpha = json.loads(os.environ["ALPHA_RESULT"])
beta = json.loads(os.environ["BETA_RESULT"])
leak_count = len(alpha["leaked_ids"]) + len(beta["leaked_ids"])
verdict = "BLOCK" if leak_count else "PASS"
result = {
"execution_id": os.environ["EXECUTION_ID"],
"collection_name": os.environ["COLLECTION_NAME"],
"injected_leak": os.environ["INJECTED_LEAK"].lower() == "true",
"leak_count": leak_count,
"verdict": verdict,
"tenants": [alpha, beta],
}
with open("tenantshield-results.json", "w", encoding="utf-8") as result_file:
json.dump(result, result_file, indent=2)
with open("tenantshield-report.md", "w", encoding="utf-8") as report_file:
report_file.write("# TenantShield isolation report\n\n")
report_file.write(f"- Execution: `{result['execution_id']}`\n")
report_file.write(f"- Collection: `{result['collection_name']}`\n")
report_file.write(f"- Negative control enabled: `{result['injected_leak']}`\n")
report_file.write(f"- Verdict: **{verdict}**\n")
report_file.write(f"- Unauthorized documents: **{leak_count}**\n\n")
for tenant in result["tenants"]:
report_file.write(f"## {tenant['tenant']}, {tenant['status']}\n\n")
report_file.write(f"Leaked document IDs: `{tenant['leaked_ids']}`\n\n")
Kestra.outputs({"result_json": json.dumps(result), "leak_count": leak_count, "verdict": verdict})
- id: record_audit_evidence
type: io.kestra.plugin.jdbc.postgresql.Query
description: Persist the complete alpha and beta reports before evaluating the
deployment gate, including blocked negative-control runs.
url: "{{ render(vars.resolved_postgres_url) }}"
username: "{{ render(vars.resolved_postgres_user) }}"
password: "{{ render(vars.resolved_postgres_password) }}"
fetchType: FETCH_ONE
parameters:
execution_id: "{{ execution.id }}"
collection_name: "{{ render(vars.collection_name) }}"
injected_leak: "{{ inputs.inject_leak }}"
alpha_result: "{{ outputs.probe_alpha.vars.result_json }}"
beta_result: "{{ outputs.probe_beta.vars.result_json }}"
leak_count: "{{ outputs.assemble_security_report.vars.leak_count }}"
verdict: "{{ outputs.assemble_security_report.vars.verdict }}"
sql: |
INSERT INTO tenantshield.isolation_audits (
execution_id, collection_name, injected_leak, alpha_result,
beta_result, leak_count, verdict
) VALUES (
:execution_id, :collection_name, CAST(:injected_leak AS boolean),
CAST(:alpha_result AS jsonb), CAST(:beta_result AS jsonb),
CAST(:leak_count AS integer), :verdict
)
ON CONFLICT (execution_id) DO UPDATE SET
alpha_result = EXCLUDED.alpha_result,
beta_result = EXCLUDED.beta_result,
leak_count = EXCLUDED.leak_count,
verdict = EXCLUDED.verdict
RETURNING execution_id, verdict, leak_count, CAST(created_at AS text) AS created_at;
timeout: PT1M
retry:
type: exponential
interval: PT2S
maxInterval: PT10S
maxAttempts: 3
- id: deployment_gate
type: io.kestra.plugin.core.flow.If
description: Fail closed after evidence is durable whenever either tenant
retrieved an unauthorized document.
condition: "{{ outputs.assemble_security_report.vars.leak_count > 0 }}"
then:
- id: block_deployment
type: io.kestra.plugin.core.execution.Fail
errorMessage: "TenantShield blocked deployment: {{
outputs.assemble_security_report.vars.leak_count }} unauthorized
document(s) were retrieved."
else:
- id: isolation_verified
type: io.kestra.plugin.core.log.Log
message: "TenantShield PASS: alpha and beta retrieved only their own documents;
audit row {{ outputs.record_audit_evidence.row.execution_id }} is
durable."
- id: delete_fixture_collection
type: io.kestra.plugin.qdrant.DeleteCollection
description: Remove the execution-scoped canary collection after a clean test so
production vector stores are not polluted.
host: "{{ render(vars.resolved_qdrant_host) }}"
port: "{{ render(vars.resolved_qdrant_port) }}"
apiKey: "{{ render(vars.resolved_qdrant_key) }}"
tlsEnabled: "{{ render(vars.resolved_qdrant_tls) }}"
collectionName: "{{ render(vars.collection_name) }}"
timeout: PT1M
retry:
type: exponential
interval: PT2S
maxInterval: PT10S
maxAttempts: 3
- id: teardown_demo_services
type: io.kestra.plugin.core.flow.If
description: Remove local containers and volumes after a successful demonstration.
condition: "{{ inputs.demo_mode }}"
then:
- id: compose_down
type: io.kestra.plugin.docker.cli.Compose
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
volumes:
- /var/run/docker.sock:/var/run/docker.sock
composeFile: "{{ render(vars.demo_compose) }}"
composeArgs:
- --project-name
- "tenantshield-{{ execution.id | lower }}"
- down
- -v
- --remove-orphans
timeout: PT2M
errors:
- id: delete_fixture_after_failure
type: io.kestra.plugin.qdrant.DeleteCollection
description: Best-effort Qdrant cleanup for both detected leaks and
infrastructure failures.
host: "{{ render(vars.resolved_qdrant_host) }}"
port: "{{ render(vars.resolved_qdrant_port) }}"
apiKey: "{{ render(vars.resolved_qdrant_key) }}"
tlsEnabled: "{{ render(vars.resolved_qdrant_tls) }}"
collectionName: "{{ render(vars.collection_name) }}"
allowFailure: true
timeout: PT1M
- id: teardown_demo_after_failure
type: io.kestra.plugin.core.flow.If
description: Always remove disposable containers and volumes when the gate
intentionally fails or a dependency is unavailable.
condition: "{{ inputs.demo_mode }}"
allowFailure: true
then:
- id: compose_down_after_failure
type: io.kestra.plugin.docker.cli.Compose
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
volumes:
- /var/run/docker.sock:/var/run/docker.sock
composeFile: "{{ render(vars.demo_compose) }}"
composeArgs:
- --project-name
- "tenantshield-{{ execution.id | lower }}"
- down
- -v
- --remove-orphans
allowFailure: true
timeout: PT2M
- id: alert_security_failure
type: io.kestra.plugin.core.flow.If
description: Notify Slack only when explicitly enabled, keeping the local
fixture runnable without a webhook secret.
condition: "{{ inputs.notify_slack }}"
then:
- id: notify_slack
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: >-
TenantShield blocked or could not complete deployment validation for
execution {{ execution.id }}. Verdict={{
outputs.assemble_security_report.vars.verdict ??
'INFRASTRUCTURE_ERROR' }}, leaks={{
outputs.assemble_security_report.vars.leak_count ?? 'unknown' }}.
Review {{
outputs.assemble_security_report.outputFiles['tenantshield-report.md']
?? 'the execution logs' }} before deploying.
retry:
type: exponential
interval: PT2S
maxInterval: PT10S
maxAttempts: 3
outputs:
- id: verdict
type: STRING
description: PASS when both payload-filtered probes are isolated; BLOCK is
persisted before the deliberate failed execution ends.
value: "{{ outputs.assemble_security_report.vars.verdict ?? 'INCOMPLETE' }}"
- id: leak_count
type: INT
description: Number of distinct unauthorized document IDs observed across both
tenant probes.
value: "{{ outputs.assemble_security_report.vars.leak_count ?? -1 }}"
- id: audit_row
type: JSON
description: PostgreSQL audit row returned after the probe evidence is committed.
value: "{{ outputs.record_audit_evidence.row ?? {} }}"
- id: report_uri
type: STRING
description: Kestra internal-storage URI for the human-readable TenantShield report.
value: "{{
outputs.assemble_security_report.outputFiles['tenantshield-report.md'] ??
'' }}"
TenantShield turns cross-tenant retrieval into a repeatable release test using real Qdrant searches and a PostgreSQL audit decision.
How it works
- In demo mode,
Docker Composestarts pinned Qdrant and PostgreSQL fixtures. - Python creates deterministic hashing-trick embeddings for four canaries; native Qdrant tasks create a unique collection and upsert them.
- A
Paralleltask queries as alpha and beta for both tenants' canaries with payload filters.inject_leak: trueremoves beta's filter as a negative control. - Python writes evidence, PostgreSQL records both probes, and an
Iffails the deployment on a wrongtenant_id. Error handlers clean up and optionally alert Slack.
What you get
- A PASS/BLOCK decision with leaked IDs, hits and scores.
- Markdown/JSON artifacts and a durable
tenantshield.isolation_auditsrow. - Retries, timeouts, fail-closed behavior, and cleanup on either outcome.
Who it's for
AI security, platform, and MLOps teams shipping shared RAG infrastructure.
Why orchestrate this with Kestra
Kestra makes fixtures, parallel probes, secret-backed connections, audit writes, retries, failure gating and Slack escalation one reviewable execution.
Prerequisites
The provided Docker task runners need Docker access; demo mode also needs socket-backed volume mounts. Production mode needs reachable Qdrant gRPC and PostgreSQL endpoints. The demo binds loopback ports 16333/16334 and 15432.
Secrets
QDRANT_API_KEY: production Qdrant authentication key.POSTGRES_USERNAMEandPOSTGRES_PASSWORD: production audit-ledger login.SLACK_WEBHOOK_URL: incoming webhook used only whennotify_slackis true.
Inputs and outputs
demo_mode selects fixtures; inject_leak enables the negative control; endpoint, collection, result-limit and notification inputs tune the run. Outputs expose verdict, leak_count, audit_row, and report_uri.
Quick start
- Give the worker Docker access and enable runner volume mounts, then execute with defaults; expect
PASSand zero leaks. - Re-run with
inject_leak: true; expect PostgreSQL to retainBLOCKand the execution to fail deliberately. - For production, set secrets, use
demo_mode: false, test manually, then enable the schedule.
How to extend
Replace the hashing fixture with your embedding API, add role/region payload dimensions, or call this flow from CI and allow deployment only after a successful execution.
Links
Tasks
Orchestrate with Kestra
Orchestrate Docker with Kestra Orchestrate Postgres with Kestra Orchestrate Slack with KestraCreate a Blueprint
Go to the developer platformMore AI Blueprints
