AI Security Incident Threat Advisor and MITRE ATT&CK Containment Generator

id: ai-security-incident-threat-advisor
namespace: company.security

description: |
  Ingest and aggregate suspicious authentication and cloud access events using DuckDB,
  evaluate threat vectors with Kestra's AI plugin mapped to MITRE ATT&CK techniques,
  synthesize copy-paste CLI containment commands, and dispatch actionable triage alerts to Slack.

triggers:
  - id: scheduled_threat_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Periodic security telemetry audit. Shipped disabled by default.
    cron: "0 */4 * * *"
    disabled: true

  - id: soc_webhook_ingest
    type: io.kestra.plugin.core.trigger.Webhook
    description: Event-driven webhook triggered by SIEM or cloud audit log shippers.
    key: "{{ secret('SECOPS_WEBHOOK_KEY') }}"

inputs:
  - id: min_anomaly_threshold
    type: INT
    defaults: 3
    description: Minimum number of failed authentication events from an entity to
      trigger AI triage.

  - id: model_name
    type: STRING
    defaults: gpt-4o-mini
    description: LLM used for Tier-1 SOC threat correlation and MITRE ATT&CK mapping.

  - id: security_audit_events
    type: JSON
    defaults:
      - id: evt_001
        timestamp: "2026-10-04T21:00:10Z"
        source_ip: "198.51.100.42"
        target_account: "admin@company.internal"
        action: "LOGIN_FAILED"
        auth_method: "PASSWORD"
      - id: evt_002
        timestamp: "2026-10-04T21:00:14Z"
        source_ip: "198.51.100.42"
        target_account: "root@company.internal"
        action: "LOGIN_FAILED"
        auth_method: "PASSWORD"
      - id: evt_003
        timestamp: "2026-10-04T21:00:19Z"
        source_ip: "198.51.100.42"
        target_account: "deploy@company.internal"
        action: "LOGIN_FAILED"
        auth_method: "PASSWORD"
      - id: evt_004
        timestamp: "2026-10-04T21:00:25Z"
        source_ip: "198.51.100.42"
        target_account: "service_s3@company.internal"
        action: "LOGIN_FAILED"
        auth_method: "API_TOKEN"
      - id: evt_005
        timestamp: "2026-10-04T21:05:00Z"
        source_ip: "192.0.2.15"
        target_account: "analyst@company.internal"
        action: "LOGIN_SUCCESS"
        auth_method: "SSO"
    description: Raw security telemetry events to audit for intrusion patterns.

tasks:
  - id: stage_security_telemetry
    type: io.kestra.plugin.jdbc.duckdb.Queries
    description: Ingests raw security audit events into an in-memory DuckDB
      analytical table.
    url: "jdbc:duckdb:"
    fetchType: NONE
    sql: |
      CREATE TABLE audit_events (
        id VARCHAR,
        timestamp VARCHAR,
        source_ip VARCHAR,
        target_account VARCHAR,
        action VARCHAR,
        auth_method VARCHAR
      );

      {% for evt in inputs.security_audit_events %}
      INSERT INTO audit_events VALUES (
        '{{ evt.id }}',
        '{{ evt.timestamp }}',
        '{{ evt.source_ip }}',
        '{{ evt.target_account }}',
        '{{ evt.action }}',
        '{{ evt.auth_method }}'
      );
      {% endfor %}

  - id: aggregate_threat_anomalies
    type: io.kestra.plugin.jdbc.duckdb.Query
    description: Aggregates failed authentication attempts by source IP to isolate
      threat actors.
    url: "jdbc:duckdb:"
    fetchType: FETCH
    sql: |
      SELECT
        source_ip,
        COUNT(*) AS failed_attempts,
        COUNT(DISTINCT target_account) AS targeted_accounts_count,
        string_agg(DISTINCT target_account, ', ') AS targeted_accounts,
        string_agg(DISTINCT auth_method, ', ') AS attempted_methods,
        MIN(timestamp) AS first_seen,
        MAX(timestamp) AS last_seen
      FROM audit_events
      WHERE action = 'LOGIN_FAILED'
      GROUP BY source_ip
      HAVING COUNT(*) >= {{ inputs.min_anomaly_threshold }};

  - id: analyze_threat_with_ai
    type: io.kestra.plugin.ai.completion.ChatCompletion
    description: Analyzes aggregated threat telemetry, maps attack to MITRE ATT&CK,
      and synthesizes containment commands.
    provider:
      type: io.kestra.plugin.ai.provider.OpenAI
      apiKey: "{{ secret('OPENAI_API_KEY') }}"
      modelName: "{{ inputs.model_name }}"
    messages:
      - type: SYSTEM
        content: |
          You are a Senior Tier-1 SOC (Security Operations Center) Incident Analyst.
          Analyze the aggregated intrusion telemetry. Identify the exact MITRE ATT&CK technique,
          determine the severity level, write a concise technical narrative, and generate copy-paste
          CLI containment commands (e.g. AWS VPC Network ACL / iptables / IAM revocation).

          Return valid JSON strictly matching the provided schema.
      - type: USER
        content: |
          Suspicious Telemetry Aggregations:
          {{ outputs.aggregate_threat_anomalies.rows | json }}
    configuration:
      responseFormat:
        type: JSON
        jsonSchema:
          type: object
          properties:
            threat_detected:
              type: boolean
            mitre_technique_id:
              type: string
            mitre_technique_name:
              type: string
            threat_severity:
              type: string
              enum:
                - LOW
                - MEDIUM
                - HIGH
                - CRITICAL
            attack_narrative:
              type: string
            containment_cli_commands:
              type: string
            quarantine_target_ip:
              type: string
          required:
            - threat_detected
            - mitre_technique_id
            - mitre_technique_name
            - threat_severity
            - attack_narrative
            - containment_cli_commands
            - quarantine_target_ip

  - id: consolidate_threat_metrics
    type: io.kestra.plugin.core.output.OutputValues
    description: Exposes structured threat metrics into Kestra execution outputs for
      downstream auditing.
    values:
      anomalous_actors_count: "{{ outputs.aggregate_threat_anomalies.rows ?
        outputs.aggregate_threat_anomalies.rows | length : 0 }}"
      threat_detected: "{{ outputs.analyze_threat_with_ai.jsonOutput.threat_detected }}"
      mitre_technique_id: "{{ outputs.analyze_threat_with_ai.jsonOutput.mitre_technique_id }}"
      mitre_technique_name: "{{ outputs.analyze_threat_with_ai.jsonOutput.mitre_technique_name }}"
      threat_severity: "{{ outputs.analyze_threat_with_ai.jsonOutput.threat_severity }}"
      quarantine_ip: "{{ outputs.analyze_threat_with_ai.jsonOutput.quarantine_target_ip }}"
      containment_commands: "{{
        outputs.analyze_threat_with_ai.jsonOutput.containment_cli_commands }}"

  - id: evaluate_threat_gate
    type: io.kestra.plugin.core.flow.If
    description: Evaluates whether an active intrusion requires containment alerts
      or baseline logging.
    condition: "{{ outputs.analyze_threat_with_ai.jsonOutput.threat_detected == true }}"
    then:
      - id: export_incident_report
        type: io.kestra.plugin.core.storage.Write
        description: Persists an executive SOC incident triage report into Kestra
          internal storage.
        fileName: "soc-incident-triage-report-{{ execution.id }}.md"
        content: |
          # SOC Security Incident Triage & Threat Intelligence Report

          - **Execution ID:** `{{ execution.id }}`
          - **Triage Timestamp:** `{{ now() }}`
          - **Threat Severity:** `{{ outputs.consolidate_threat_metrics.threat_severity }}`
          - **MITRE ATT&CK Technique:** `{{ outputs.consolidate_threat_metrics.mitre_technique_id }} - {{ outputs.consolidate_threat_metrics.mitre_technique_name }}`
          - **Attacking Entity (IP):** `{{ outputs.consolidate_threat_metrics.quarantine_ip }}`

          ## Threat Narrative
          {{ outputs.analyze_threat_with_ai.jsonOutput.attack_narrative }}

          ## Telemetry Evidence
          ```json
          {{ outputs.aggregate_threat_anomalies.rows | json }}
          ```

          ## Recommended Containment Runbook
          Execute the following commands to quarantine the attacker immediately:
          ```bash
          {{ outputs.consolidate_threat_metrics.containment_commands }}
          ```

      - id: notify_soc_slack
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Delivers high-priority triage card to Slack with MITRE technique
          and copy-paste containment script.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "🛡️ *SOC SECURITY INCIDENT ALERT: {{ outputs.consolidate_threat_metrics.threat_severity }}*\n*MITRE Technique:* `{{ outputs.consolidate_threat_metrics.mitre_technique_id }}` - {{ outputs.consolidate_threat_metrics.mitre_technique_name }}\n*Attacker IP:* `{{ outputs.consolidate_threat_metrics.quarantine_ip }}`\n*Narrative:* {{ outputs.analyze_threat_with_ai.jsonOutput.attack_narrative }}\n*Containment Commands:*\n```\n{{ outputs.consolidate_threat_metrics.containment_commands }}\n```\n*Incident Report:* `{{ outputs.export_incident_report.uri }}`"
          }

    else:
      - id: log_baseline_activity
        type: io.kestra.plugin.core.log.Log
        description: Records baseline security telemetry verification.
        message: "Security telemetry audit completed. No anomalies exceeding threshold
          ({{ inputs.min_anomaly_threshold }}). Baseline normal."

errors:
  - id: alert_audit_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alerts on-call SecOps engineers if the telemetry audit flow fails.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "⚠️ *SecOps Audit Flow Failure:* Execution {{ execution.id }} encountered an error processing security telemetry."
      }