id: mcp-server-supply-chain-audit
namespace: company.security
inputs:
- id: inventory
type: JSON
displayName: MCP server inventory
description: The MCP servers your agents are allowed to run, with the exact
package version you deployed. `registry` is the server name in the
official MCP Registry and is optional; when set, the flow also checks how
far your pin is behind the latest published release.
defaults: |
[
{"name": "filesystem", "ecosystem": "npm", "package": "@modelcontextprotocol/server-filesystem", "version": "0.6.2"},
{"name": "mcp-remote", "ecosystem": "npm", "package": "mcp-remote", "version": "0.1.10"},
{"name": "fetch", "ecosystem": "PyPI", "package": "mcp-server-fetch", "version": "2025.1.14"},
{"name": "python-sdk", "ecosystem": "PyPI", "package": "mcp", "version": "1.8.0"},
{"name": "github", "registry": "io.github.github/github-mcp-server", "ecosystem": "Go", "package": "github.com/github/github-mcp-server", "version": "0.5.0"},
{"name": "remote-filesystem", "registry": "com.pulsemcp/remote-filesystem", "ecosystem": "npm", "package": "remote-filesystem-mcp-server", "version": "0.1.2"}
]
- id: fail_on_severity
type: SELECT
displayName: Fail the execution at this severity
description: Turn the audit into a gate. A CI job that calls the webhook gets a
failed execution when any server carries a vulnerability at or above this
level. NONE only reports.
values: [ NONE, CRITICAL, HIGH, MODERATE ]
defaults: NONE
- id: notify_slack
type: BOOL
displayName: Post findings to Slack
description: Send the findings table to the Slack incoming webhook stored in the
SLACK_WEBHOOK_URL secret.
defaults: true
variables:
inventory_json: "{{ (trigger.body.inventory ?? inputs.inventory) | toJson }}"
registry_names: "{{ fromJson(render(vars.inventory_json)) | jq('[.[] |
select(.registry != null) | .registry]') | first | toJson }}"
triggers:
- id: on_mcp_config_change
type: io.kestra.plugin.core.trigger.Webhook
description: Event-driven entry point. A CI job POSTs the inventory parsed from
`.mcp.json` or your agent config whenever it changes, so a newly added or
re-pinned server is audited before it reaches developers. The key guards
the URL, so keep it in a secret.
key: "{{ secret('MCP_AUDIT_WEBHOOK_KEY') }}"
- id: daily
type: io.kestra.plugin.core.trigger.Schedule
description: Advisories get published against versions you already run, so the
same inventory is re-audited every morning even when nothing changed on
your side.
cron: "0 7 * * *"
tasks:
- id: collect
type: io.kestra.plugin.core.flow.Parallel
description: The vulnerability lookup and the registry lookups are independent,
so they run side by side.
tasks:
- id: vulnerabilities
type: io.kestra.plugin.core.flow.Sequential
description: Find which pinned versions are affected, then fetch the full
advisory for each hit.
tasks:
- id: osv_batch
type: io.kestra.plugin.core.http.Request
description: One OSV.dev batch query for every pinned package. OSV aggregates
GitHub Security Advisories, PyPA and the Go vulnerability
database, and needs no API key.
uri: https://api.osv.dev/v1/querybatch
method: POST
contentType: application/json
body: |
{"queries": [{% for s in fromJson(render(vars.inventory_json)) %}{"package": {"name": "{{ s.package }}", "ecosystem": "{{ s.ecosystem }}"}, "version": "{{ s.version }}"}{% if not loop.last %},{% endif %}{% endfor %}]}
- id: advisories
type: io.kestra.plugin.core.flow.Loop
description: One OSV advisory lookup per distinct vulnerability id, for
severity, CVE alias and the first fixed version.
values: "{{ outputs.osv_batch.body | jq('[.results[].vulns // [] | .[].id] |
unique') | first | toJson }}"
outputs:
- id: advisory
type: STRING
value: "{{ outputs.get_advisory.body }}"
tasks:
- id: get_advisory
type: io.kestra.plugin.core.http.Request
uri: "https://api.osv.dev/v1/vulns/{{ item.value }}"
- id: registry
type: io.kestra.plugin.core.flow.Loop
description: For every server published in the official MCP Registry, fetch its
latest release to measure pin drift and catch deprecated or deleted
servers.
values: "{{ render(vars.registry_names) }}"
outputs:
- id: release
type: STRING
value: >-
{{ {'registry': item.value, 'http_code':
outputs.latest_release.code, 'latest_version':
(outputs.latest_release.code == 200 ?
fromJson(outputs.latest_release.body).server.version : null),
'status': (outputs.latest_release.code == 200 ?
fromJson(outputs.latest_release.body)._meta['io.modelcontextprotocol.registry/official'].status
: 'not found')} | toJson }}
tasks:
- id: latest_release
type: io.kestra.plugin.core.http.Request
uri: "https://registry.modelcontextprotocol.io/v0/servers/{{ item.value |
replace({'/': '%2F'}) }}/versions/latest"
options:
allowFailed: true
- id: build_report
type: io.kestra.plugin.jdbc.duckdb.Query
description: >
DuckDB joins the inventory, the OSV batch answer, the advisories and the
registry answers into one row per server, worst risk first. The severity
ranking and the verdict live in SQL so they are easy to adjust.
inputFiles:
inventory.json: "{{ render(vars.inventory_json) }}"
osv.json: "{{ outputs.osv_batch.body }}"
advisories.json: "{{ (outputs.advisories.outputs ?? []) | toJson }}"
registry.json: "{{ (outputs.registry.outputs ?? []) | toJson }}"
fetchType: FETCH
sql: |
WITH inventory AS (
SELECT row_number() OVER () - 1 AS idx, *
FROM read_json('{{ workingDir }}/inventory.json',
columns = {name: 'VARCHAR', registry: 'VARCHAR', ecosystem: 'VARCHAR', package: 'VARCHAR', version: 'VARCHAR'})
),
results AS (
SELECT unnest(range(json_array_length(json -> '$.results')::BIGINT)) AS idx, json -> '$.results' AS r
FROM read_json_objects('{{ workingDir }}/osv.json')
),
hits AS (
SELECT idx, json_extract_string(v.value, '$.id') AS vuln_id
FROM results, LATERAL json_each(r -> ('$[' || idx || '].vulns')) v
),
advisories AS (
SELECT json_extract_string(adv, '$.id') AS vuln_id,
coalesce(list_filter(json_extract_string(adv, '$.aliases[*]'), x -> x LIKE 'CVE-%')[1], json_extract_string(adv, '$.id')) AS cve,
upper(coalesce(json_extract_string(adv, '$.database_specific.severity'), 'UNKNOWN')) AS severity,
list_filter(json_extract_string(adv, '$.affected[*].ranges[*].events[*].fixed'), x -> x IS NOT NULL)[1] AS fixed_in
FROM (SELECT json_extract_string(json, '$.outputs.advisory')::JSON AS adv
FROM read_json_objects('{{ workingDir }}/advisories.json'))
),
registry AS (
SELECT json_extract_string(rel, '$.registry') AS registry,
json_extract_string(rel, '$.latest_version') AS latest_version,
json_extract_string(rel, '$.status') AS registry_status
FROM (SELECT json_extract_string(json, '$.outputs.release')::JSON AS rel
FROM read_json_objects('{{ workingDir }}/registry.json'))
)
SELECT i.name, i.ecosystem, i.package, i.version AS pinned_version,
g.latest_version, g.registry_status,
count(DISTINCT a.cve) AS vulns,
CASE max(CASE a.severity WHEN 'CRITICAL' THEN 4 WHEN 'HIGH' THEN 3 WHEN 'MODERATE' THEN 2 WHEN 'MEDIUM' THEN 2 WHEN 'LOW' THEN 1 ELSE 0 END)
WHEN 4 THEN 'CRITICAL' WHEN 3 THEN 'HIGH' WHEN 2 THEN 'MODERATE' WHEN 1 THEN 'LOW' ELSE 'NONE' END AS max_severity,
coalesce(string_agg(DISTINCT a.cve, ', '), '') AS advisories,
max_by(a.fixed_in, list_transform(string_split(regexp_replace(a.fixed_in, '[^0-9.]', '', 'g'), '.'), x -> TRY_CAST(x AS INTEGER))) AS upgrade_to,
CASE
WHEN count(a.cve) > 0 THEN 'VULNERABLE'
WHEN g.registry_status IN ('deprecated', 'deleted') THEN 'DEPRECATED'
WHEN g.latest_version IS NOT NULL AND g.latest_version <> i.version THEN 'OUTDATED'
ELSE 'OK'
END AS verdict
FROM inventory i
LEFT JOIN hits h ON h.idx = i.idx
LEFT JOIN advisories a ON a.vuln_id = h.vuln_id
LEFT JOIN registry g ON g.registry = i.registry
GROUP BY i.name, i.ecosystem, i.package, i.version, g.latest_version, g.registry_status
ORDER BY vulns DESC, verdict, i.name
- id: has_findings
type: io.kestra.plugin.core.flow.If
description: Only alert when at least one server is vulnerable, deprecated or
behind the registry.
condition: "{{ outputs.build_report.rows | jq('[.[] | select(.verdict !=
\"OK\")] | length') | first > 0 }}"
then:
- id: log_findings
type: io.kestra.plugin.core.log.Log
message: |
MCP supply-chain audit: {{ outputs.build_report.rows | jq('[.[] | select(.verdict != "OK")] | length') | first }} of {{ outputs.build_report.size }} servers need attention.
{% for r in outputs.build_report.rows %}{% if r.verdict != 'OK' %}- {{ r.name }} ({{ r.package }}@{{ r.pinned_version }}): {{ r.verdict }}{% if r.verdict == 'VULNERABLE' %}, {{ r.vulns }} advisories, max {{ r.max_severity }} [{{ r.advisories }}], upgrade to {{ r.upgrade_to ?? 'latest' }}{% endif %}{% if r.latest_version is defined and r.latest_version != null %}, registry latest {{ r.latest_version }} ({{ r.registry_status }}){% endif %}
{% endif %}{% endfor %}
- id: slack_enabled
type: io.kestra.plugin.core.flow.If
condition: "{{ inputs.notify_slack ?? true }}"
then:
- id: slack_alert
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: |
:rotating_light: *MCP server supply-chain audit* found issues in `{{ flow.namespace }}`
{% for r in outputs.build_report.rows %}{% if r.verdict != 'OK' %}• *{{ r.name }}* `{{ r.package }}@{{ r.pinned_version }}`: {{ r.verdict }}{% if r.verdict == 'VULNERABLE' %} ({{ r.max_severity }}: {{ r.advisories }}, upgrade to {{ r.upgrade_to ?? 'latest' }}){% endif %}
{% endif %}{% endfor %}Kestra execution: {{ execution.id }}
- id: gate
type: io.kestra.plugin.core.flow.If
description: Fail the run when the configured severity threshold is reached, so
a CI caller can block the config change.
condition: >-
{% set rank = {'NONE': 99, 'CRITICAL': 4, 'HIGH': 3, 'MODERATE': 2}
%}{% set threshold = rank[inputs.fail_on_severity ?? 'NONE'] %}{{
outputs.build_report.rows | jq('[.[] |
{"CRITICAL":4,"HIGH":3,"MODERATE":2,"LOW":1,"NONE":0}[.max_severity]]
| max // 0') | first >= threshold }}
then:
- id: block
type: io.kestra.plugin.core.execution.Fail
errorMessage: "At least one MCP server carries a {{ inputs.fail_on_severity }}
or worse vulnerability."
else:
- id: all_clear
type: io.kestra.plugin.core.log.Log
message: "All {{ outputs.build_report.size }} MCP servers are on patched,
current releases."
