id: cisa-kev-exploited-vulnerability-watch-ai-triage
namespace: company.team
description: |
Watch the CISA Known Exploited Vulnerabilities (KEV) catalog, match newly added CVEs against the
vendors and products in your stack, enrich the matches with FIRST EPSS exploit-probability scores,
and post a prioritized Slack alert where Gemini explains the likely exposure and the next step.
inputs:
- id: stack
type: JSON
displayName: Technology stack
description: List of objects with a vendor, an optional product and where you
run it. Matching is a case-insensitive substring match on the KEV
vendorProject and product fields.
defaults: |
[
{"vendor": "Microsoft", "product": "Exchange Server", "where": "On-premises mail server for staff, reachable from the internet."},
{"vendor": "Fortinet", "product": "FortiOS", "where": "FortiGate VPN appliances at both offices."},
{"vendor": "Atlassian", "product": "Confluence", "where": "Self-hosted Confluence Data Center behind SSO."},
{"vendor": "Apache", "product": "", "where": "Java services using Apache libraries, plus Apache HTTP Server reverse proxies."},
{"vendor": "Ivanti", "product": "", "where": "Ivanti Connect Secure remote access for contractors."},
{"vendor": "ISC", "product": "BIND", "where": "Internal DNS resolvers in the data center."}
]
- id: environment_context
type: STRING
displayName: Environment context
description: One or two sentences about your environment, used by the triage step.
defaults: A 300-person company with a hybrid network. Internet-facing systems
are patched by the platform team; everything else in a monthly patch
window.
- id: lookback_days
type: INT
displayName: Lookback window (days)
description: Only CVEs added to KEV within this many days are considered, so the
first run reports recent additions instead of the whole history.
defaults: 30
- id: model_name
type: STRING
displayName: Gemini model
defaults: gemini-2.5-flash
tasks:
- id: fetch_kev
type: io.kestra.plugin.core.http.Download
description: Download the full KEV catalog (public JSON, no key, about 2 MB).
uri: https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json
headers:
User-Agent: Kestra-KEV-Watch
- id: load_seen
type: io.kestra.plugin.core.kv.Get
description: Load the CVE ids already evaluated by this watch. Missing on the first run.
key: cisa_kev_watch_seen
errorOnMissing: false
- id: match_stack
type: io.kestra.plugin.scripts.python.Script
description: Keep KEV entries that are new since the last run, inside the
lookback window, and match a vendor or product in the stack.
containerImage: python:3.12-slim
inputFiles:
kev.json: "{{ outputs.fetch_kev.uri }}"
seen.json: "{{ (outputs.load_seen.value ?? []) | toJson }}"
stack.json: "{{ inputs.stack | toJson }}"
script: |
import json
from datetime import date, timedelta
with open("kev.json", encoding="utf-8") as f:
catalog = json.load(f)
with open("seen.json") as f:
seen = json.load(f) or []
if isinstance(seen, str):
seen = json.loads(seen)
with open("stack.json") as f:
stack = json.load(f)
if isinstance(stack, str):
stack = json.loads(stack)
seen = set(seen)
cutoff = (date.today() - timedelta(days={{ inputs.lookback_days }})).isoformat()
vulns = catalog.get("vulnerabilities", [])
matches = []
for v in vulns:
if v["cveID"] in seen or v.get("dateAdded", "") < cutoff:
continue
vendor = (v.get("vendorProject") or "").lower()
product = (v.get("product") or "").lower()
for s in stack:
want_vendor = (s.get("vendor") or "").lower().strip()
want_product = (s.get("product") or "").lower().strip()
if want_vendor and want_vendor not in vendor:
continue
if want_product and want_product not in product:
continue
matches.append({
"cve": v["cveID"],
"name": (v.get("vulnerabilityName") or "").strip(),
"vendor": v.get("vendorProject"),
"product": v.get("product"),
"date_added": v.get("dateAdded"),
"due_date": v.get("dueDate"),
"ransomware": v.get("knownRansomwareCampaignUse") == "Known",
"description": (v.get("shortDescription") or "")[:600],
"where": s.get("where", ""),
"link": f"https://nvd.nist.gov/vuln/detail/{v['cveID']}",
})
break
matches.sort(key=lambda m: m["date_added"], reverse=True)
matches = matches[:20]
print(f"KEV catalog {catalog.get('catalogVersion')}: {len(vulns)} entries, "
f"{len(matches)} new matches for the stack (added since {cutoff}).")
print("::" + json.dumps({"outputs": {
"match_count": len(matches),
"matches": matches,
"cve_list": ",".join(m["cve"] for m in matches),
"all_ids": [v["cveID"] for v in vulns],
}}) + "::")
- id: if_matches
type: io.kestra.plugin.core.flow.If
description: Only call EPSS, the LLM and Slack when a new KEV entry affects the stack.
condition: "{{ outputs.match_stack.vars.match_count > 0 }}"
then:
- id: fetch_epss
type: io.kestra.plugin.core.http.Request
description: Exploit prediction scores for the matched CVEs from the public
FIRST EPSS API (no key).
uri: "https://api.first.org/data/v1/epss?cve={{
outputs.match_stack.vars.cve_list }}"
method: GET
- id: triage
type: io.kestra.plugin.ai.completion.ChatCompletion
description: Rate the urgency of every matched CVE for this environment and
suggest one next step, as structured JSON.
provider:
type: io.kestra.plugin.ai.provider.GoogleGemini
apiKey: "{{ secret('GEMINI_API_KEY') }}"
modelName: "{{ inputs.model_name }}"
configuration:
temperature: 0.1
responseFormat:
type: JSON
jsonSchema:
type: object
properties:
items:
type: array
items:
type: object
properties:
cve:
type: string
priority:
type: string
enum:
- P1
- P2
- P3
exposure:
type: string
next_step:
type: string
required:
- cve
- priority
- exposure
- next_step
required:
- items
messages:
- type: SYSTEM
content: |
You are a vulnerability management lead. Every CVE you receive is in the CISA Known Exploited
Vulnerabilities catalog, so it is being exploited in the wild. For each CVE, return its exact id,
a priority (P1 = patch or mitigate today, P2 = this week, P3 = next patch window), an exposure note of
at most 25 words explaining why, based on where we run the product, and one concrete next step of at
most 20 words. Internet-facing systems and known ransomware use raise the priority. Use only the facts
you are given; do not invent version numbers or patch names.
- type: USER
content: |
Environment: {{ inputs.environment_context }}
Matched KEV entries:
{{ outputs.match_stack.vars.matches | toJson }}
EPSS response:
{{ outputs.fetch_epss.body }}
- id: build_alert
type: io.kestra.plugin.scripts.python.Script
description: Join the AI triage to the KEV facts by CVE id, add EPSS, drop
unknown or duplicate ids and sort by priority.
containerImage: python:3.12-slim
inputFiles:
matches.json: "{{ outputs.match_stack.vars.matches | toJson }}"
triage.json: "{{ outputs.triage.jsonOutput | toJson }}"
epss.json: "{{ outputs.fetch_epss.body }}"
script: |
import json
def load(name):
with open(name, encoding="utf-8") as f:
data = json.load(f)
return json.loads(data) if isinstance(data, str) else data
matches = load("matches.json")
triage = load("triage.json") or {}
epss = {row["cve"]: row for row in (load("epss.json") or {}).get("data", [])}
def short(text, limit=280):
text = " ".join(str(text or "").split())
return text if len(text) <= limit else text[:limit].rsplit(" ", 1)[0] + "..."
by_cve = {}
for item in triage.get("items", []):
cve = (item.get("cve") or "").strip().upper()
if cve and cve not in by_cve:
by_cve[cve] = item
rows = []
for m in matches:
t = by_cve.get(m["cve"], {})
e = epss.get(m["cve"], {})
rows.append({**m,
# A KEV entry is exploited by definition, so a missing triage defaults to P2, never to silence.
"priority": t.get("priority") if t.get("priority") in ("P1", "P2", "P3") else "P2",
"exposure": short(t.get("exposure") or "No AI assessment returned for this CVE."),
"next_step": short(t.get("next_step") or "Check the vendor advisory linked from NVD."),
"epss": f"{float(e['epss']) * 100:.1f}%" if e.get("epss") else "n/a",
})
rows.sort(key=lambda r: (r["priority"], r["date_added"]))
dropped = len([c for c in by_cve if c not in {m["cve"] for m in matches}])
print(f"{len(rows)} alert rows, {dropped} unknown CVE ids from the model dropped.")
print("::" + json.dumps({"outputs": {
"rows": rows,
"p1_count": len([r for r in rows if r["priority"] == "P1"]),
}}) + "::")
- id: alert_slack
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Post the KEV facts (from CISA and FIRST, never from the model) with
the AI triage.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: |
{{ outputs.build_alert.vars.p1_count > 0 ? ':rotating_light:' : ':shield:' }} *{{ outputs.match_stack.vars.match_count }} newly exploited CVE(s) affect our stack* (CISA KEV)
{% for r in outputs.build_alert.vars.rows %}
*{{ r.priority }}* <{{ r.link }}|{{ r.cve }}>: {{ r.name }}
> {{ r.vendor }} {{ r.product }} | added {{ r.date_added }} | CISA due {{ r.due_date }} | EPSS {{ r.epss }}{{ r.ransomware ? ' | :skull: known ransomware use' : '' }}
> Exposure: {{ r.exposure }}
> Next step: {{ r.next_step }}
{% endfor %}
- id: remember_seen
type: io.kestra.plugin.core.kv.Set
description: Remember every CVE id in the catalog. Runs after the alert, so a
failed alert is retried on the next run.
key: cisa_kev_watch_seen
kvType: JSON
overwrite: true
value: "{{ outputs.match_stack.vars.all_ids | toJson }}"
errors:
- id: alert_failure
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: A broken watch must not look like "no new exploited CVEs".
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: ":warning: CISA KEV watch failed in execution {{ execution.id }}
({{ flow.namespace }}.{{ flow.id }}). The KEV or EPSS feed may be
unreachable, or GEMINI_API_KEY may be invalid."
triggers:
- id: every_two_hours
type: io.kestra.plugin.core.trigger.Schedule
description: CISA adds entries during US business hours on most weekdays.
Disabled until the secrets are set.
cron: "0 */2 * * *"
disabled: true
