id: pip-audit-dependency-cve-gate
namespace: company.team
description: |
Gate a Python service release on the known-CVE state of its locked
dependencies. A deploy pipeline POSTs the service name to a keyed
webhook, the flow audits the pinned requirements inside the official
pip-audit container, and a Python step grades the JSON report by
severity, honoring a CVE allowlist and a tolerated high-severity
budget. Anything above the budget posts the findings to Slack and fails
the execution so the release stops, while a clean audit is recorded and
the verdict, counts, and advisory ids are published as outputs.
triggers:
- id: on_release_request
type: io.kestra.plugin.core.trigger.Webhook
description: Your deploy pipeline POSTs here with the service name when a
release requests a dependency CVE audit.
key: "{{ secret('PIP_AUDIT_WEBHOOK_KEY') }}"
- id: nightly_audit
type: io.kestra.plugin.core.trigger.Schedule
description: Fallback net, a nightly re-audit of the same locked pins, shipped
disabled until an operator switches it on.
cron: "0 3 * * *"
disabled: true
inputs:
- id: service_name
type: STRING
defaults: checkout-api
description: Name of the Python service being released, used in the Slack
messages and the audit log lines.
- id: max_high_cves
type: INT
defaults: 0
description: Number of HIGH or CRITICAL advisories tolerated before the release
is blocked.
- id: cve_allowlist
type: ARRAY
itemType: STRING
defaults: []
description: CVE ids accepted by security review that must not count toward the
block decision.
tasks:
# --- 1. lockfile audit ---
- id: audit_workspace
type: io.kestra.plugin.core.flow.WorkingDirectory
description: One shared directory so the locked pins, the pip-audit report, and
the grader all see the same files.
tasks:
- id: run_pip_audit
type: io.kestra.plugin.scripts.shell.Commands
description: Write the locked pins and a bundled fixture report, run pip-audit
against the requirements file in its official image, and fall back to
the fixture when no index is reachable so the gate stays deterministic
offline.
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
entryPoint: []
containerImage: pyupio/pip-audit:latest
outputFiles:
- pip-audit-report.json
commands:
- |
set -u
cat > requirements.txt <<'REQ'
requests==2.19.1
urllib3==1.24.1
pyyaml==5.3.1
REQ
cat > fixture-report.json <<'JSON'
[
{
"name": "requests",
"version": "2.19.1",
"vulns": [
{
"id": "CVE-2018-18074",
"fix_versions": ["2.20.0"],
"description": "The Requests package before 2.20.0 leaks the Proxy-Authorization header on redirects.",
"aliases": ["GHSA-x84v-xcm2-53pg"],
"severity": "HIGH"
}
]
},
{
"name": "urllib3",
"version": "1.24.1",
"vulns": [
{
"id": "CVE-2019-11324",
"fix_versions": ["1.24.3"],
"description": "urllib3 before 1.24.3 mishandles certificates for hosts with trailing dots.",
"aliases": ["GHSA-q2q7-5pp4-w6pg"],
"severity": "MEDIUM"
},
{
"id": "CVE-2019-11236",
"fix_versions": ["1.24.3"],
"description": "urllib3 before 1.24.2 allows CRLF injection through a crafted path.",
"aliases": ["GHSA-w27p-4786-9cwp"],
"severity": "MEDIUM"
}
]
},
{
"name": "pyyaml",
"version": "5.3.1",
"vulns": [
{
"id": "CVE-2020-14343",
"fix_versions": ["5.4"],
"description": "PyYAML full_load permits arbitrary code execution through crafted tags.",
"aliases": ["GHSA-8q59-q68h-6hv4"],
"severity": "CRITICAL"
}
]
}
]
JSON
echo "Auditing locked pins with pip-audit"
pip-audit -f json -r requirements.txt --progress-spinner off > pip-audit-report.json || true
if [ ! -s pip-audit-report.json ]; then
cp fixture-report.json pip-audit-report.json
echo "no package index reachable from this worker, emitting the bundled fixture report"
fi
echo "report bytes $(wc -c < pip-audit-report.json)"
- id: parse_report
type: io.kestra.plugin.scripts.python.Script
description: Grade the pip-audit JSON by severity, drop allowlisted ids, and
emit the block decision and counts through the outputs protocol.
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
containerImage: python:3.12-slim
env:
SERVICE_NAME: "{{ inputs.service_name }}"
MAX_HIGH: "{{ inputs.max_high_cves }}"
ALLOWLIST: "{{ inputs.cve_allowlist | join(',') }}"
script: |
import json
import os
report_path = "pip-audit-report.json"
service = os.environ.get("SERVICE_NAME", "unknown-service")
try:
max_high = int(os.environ.get("MAX_HIGH", "0"))
except ValueError:
max_high = 0
allowlist = {a.strip() for a in os.environ.get("ALLOWLIST", "").split(",") if a.strip()}
findings = []
if os.path.exists(report_path) and os.path.getsize(report_path) > 0:
try:
with open(report_path, "r") as handle:
data = json.load(handle)
if isinstance(data, list):
findings = data
except ValueError:
findings = []
rows = []
dependencies_checked = 0
for entry in findings:
if not isinstance(entry, dict):
continue
dependencies_checked += 1
name = str(entry.get("name") or "unknown")
version = str(entry.get("version") or "")
for vuln in entry.get("vulns") or []:
if not isinstance(vuln, dict):
continue
vid = str(vuln.get("id") or "UNKNOWN")
severity = str(vuln.get("severity") or "UNKNOWN").strip().upper()
rows.append({
"id": vid,
"severity": severity,
"package": name + "@" + version,
"fix_versions": [str(v) for v in (vuln.get("fix_versions") or [])],
})
allowlisted = [r for r in rows if r["id"] in allowlist]
counted = [r for r in rows if r["id"] not in allowlist]
def count_severity(sev):
return len([r for r in counted if r["severity"] == sev])
critical_count = count_severity("CRITICAL")
high_count = count_severity("HIGH")
medium_count = count_severity("MEDIUM")
low_count = count_severity("LOW") + count_severity("NEGLIGIBLE") + count_severity("UNKNOWN")
high_and_above = critical_count + high_count
cve_ids = sorted({r["id"] for r in counted})
cve_summary = ", ".join(
r["package"] + ":" + r["id"] + " (" + r["severity"] + ")" for r in counted
) or "none"
blocked = high_and_above > max_high
verdict = "BLOCK" if blocked else "PASS"
payload = {
"service": service,
"dependencies_checked": dependencies_checked,
"cve_count": len(counted),
"critical_count": critical_count,
"high_count": high_count,
"medium_count": medium_count,
"low_count": low_count,
"allowlisted_count": len(allowlisted),
"cve_ids": cve_ids,
"cve_summary": cve_summary,
"blocked": blocked,
"verdict": verdict,
}
print("service=%s cves=%d high_plus=%d allowlisted=%d verdict=%s" % (
service, len(counted), high_and_above, len(allowlisted), verdict))
print("::" + json.dumps({"outputs": payload}) + "::")
# --- 2. verdict gate ---
- id: check_verdict
type: io.kestra.plugin.core.flow.If
description: Block the release only when the allowlist-filtered HIGH and
CRITICAL count sits above the tolerated budget.
condition: "{{ outputs.parse_report.vars.blocked }}"
then:
- id: notify_block
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Post the blocking CVE summary to the release channel so the
pipeline failure is never a mystery.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: |
{
"text": "BLOCKED: dependency CVE gate for {{ inputs.service_name }} - {{ outputs.parse_report.vars.cve_count }} advisory(ies), {{ outputs.parse_report.vars.high_count }} HIGH or CRITICAL after the allowlist. {{ outputs.parse_report.vars.cve_summary }}. Execution {{ execution.id }}."
}
- id: fail_release
type: io.kestra.plugin.core.execution.Fail
description: Fail the execution so the deploy pipeline waiting on this flow
stops before the artifact ships.
errorMessage: "Dependency CVE gate blocked {{ inputs.service_name }} with {{
outputs.parse_report.vars.high_count }} HIGH or CRITICAL
advisory(ies)."
else:
- id: log_pass
type: io.kestra.plugin.core.log.Log
description: Record the clean audit so the execution history doubles as release
evidence.
message: "pip-audit gate passed for {{ inputs.service_name }} with {{
outputs.parse_report.vars.cve_count }} counted advisory(ies) against a
budget of {{ inputs.max_high_cves }} HIGH or CRITICAL."
errors:
- id: notify_failure
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Escalate any failure to the ops channel with the execution url.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: |
{"text": "FAILED: {{ flow.id }} {{ execution.id }}"}
outputs:
- id: verdict
type: STRING
description: BLOCK when the HIGH or CRITICAL budget is exceeded, otherwise PASS.
value: "{{ outputs.parse_report.vars.verdict }}"
- id: cve_count
type: INT
description: Advisories counted after the allowlist was applied, for example 4.
value: "{{ outputs.parse_report.vars.cve_count }}"
- id: high_count
type: INT
description: HIGH or CRITICAL advisories counted after the allowlist was
applied, for example 2.
value: "{{ outputs.parse_report.vars.high_count }}"
- id: cve_ids
type: ARRAY
description: 'Distinct counted advisory ids, for example ["CVE-2020-14343"].'
value: "{{ outputs.parse_report.vars.cve_ids | toJson }}"
- id: dependencies_checked
type: INT
description: Number of locked dependencies that pip-audit examined during this run.
value: "{{ outputs.parse_report.vars.dependencies_checked }}"
