Automated AI Security Vulnerability Triage and Alerting

id: automated-security-vulnerability-triage-ai
namespace: company.security

inputs:
  - id: repository_name
    type: STRING
    defaults: "kestra-io/kestra"
    description: "GitHub repository to monitor for security advisories or alerts"

tasks:
  - id: fetch_security_advisories
    type: io.kestra.plugin.core.http.Request
    uri: "https://api.github.com/repos/{{ inputs.repository_name
      }}/security-advisories"
    method: GET
    headers:
      Accept: "application/vnd.github+json"
      User-Agent: "Kestra-Security-Automation"

  - id: analyze_vulnerabilities_ai
    type: io.kestra.plugin.scripts.python.Script
    warningOnStdErr: false
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    containerImage: ghcr.io/kestra-io/pydata:latest
    env:
      OPENAI_API_KEY: "{{ secret('OPENAI_API_KEY') }}"
    script: |
      import os
      import json
      from openai import OpenAI
      import pandas as pd
      from kestra import Kestra

      client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY", ""))

      raw_data = """{{ outputs.fetch_security_advisories.body }}"""
      try:
          advisories = json.loads(raw_data)
      except Exception:
          advisories = []

      # If list is empty or API rate limited, provide sample advisory for validation
      if not isinstance(advisories, list) or len(advisories) == 0:
          advisories = [{
              "ghsa_id": "GHSA-sample-demo",
              "summary": "Potential remote execution vulnerability in dependency parser",
              "severity": "high",
              "vulnerabilities": [{"package": {"name": "sample-parser", "ecosystem": "npm"}}]
          }]

      triaged_results = []
      critical_or_high_count = 0

      for adv in advisories[:5]:
          ghsa_id = adv.get("ghsa_id", "UNKNOWN")
          summary = adv.get("summary", "No summary provided")
          reported_sev = adv.get("severity", "medium").upper()

          prompt = f"""
          Evaluate this security advisory:
          ID: {ghsa_id}
          Summary: {summary}
          Reported Severity: {reported_sev}

          Produce a brief JSON triage report:
          - action_urgency: "P0", "P1", "P2", or "P3"
          - recommended_action: 1-sentence remediation step
          - business_impact: 1-sentence impact description
          """

          try:
              response = client.chat.completions.create(
                  model="gpt-4o-mini",
                  messages=[
                      {"role": "system", "content": "You are a senior security engineer. Provide triage strictly as valid JSON."},
                      {"role": "user", "content": prompt}
                  ],
                  response_format={"type": "json_object"}
              )
              triage = json.loads(response.choices[0].message.content)
          except Exception as e:
              triage = {
                  "action_urgency": "P1" if reported_sev in ["HIGH", "CRITICAL"] else "P2",
                  "recommended_action": "Audit dependency tree and upgrade affected package.",
                  "business_impact": "Potential security exposure in production environments."
              }

          if reported_sev in ["HIGH", "CRITICAL"] or triage.get("action_urgency") in ["P0", "P1"]:
              critical_or_high_count += 1

          triaged_results.append({
              "ghsa_id": ghsa_id,
              "severity": reported_sev,
              "summary": summary,
              "urgency": triage.get("action_urgency", "P2"),
              "recommended_action": triage.get("recommended_action", "Upgrade package"),
              "business_impact": triage.get("business_impact", "Unknown")
          })

      df = pd.DataFrame(triaged_results)
      output_csv = "triaged_security_alerts.csv"
      df.to_csv(output_csv, index=False)

      Kestra.outputs({
          "csv_path": output_csv,
          "alert_count": len(triaged_results),
          "urgent_count": critical_or_high_count,
          "highest_severity": "CRITICAL" if critical_or_high_count > 0 else "LOW"
      })
    outputFiles:
      - "triaged_security_alerts.csv"

  - id: check_urgent_threats
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.analyze_vulnerabilities_ai.vars.urgent_count > 0 }}"
    then:
      - id: notify_security_slack
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "#security-triage",
            "text": ":rotating_light: *Security Advisory Alert - Action Required* :rotating_light:\n*Repository:* {{ inputs.repository_name }}\n*Urgent Vulnerabilities Found:* {{ outputs.analyze_vulnerabilities_ai.vars.urgent_count }}\n*Severity Level:* {{ outputs.analyze_vulnerabilities_ai.vars.highest_severity }}\n*Triage File:* `triaged_security_alerts.csv` generated by Kestra AI Workflow."
          }