AWS VPC Flow Logs Rejected Traffic Anomaly Sentinel

id: aws-vpc-flow-logs-rejected-traffic-spike-sentinel
namespace: company.ops

inputs:
  - id: aws_region
    type: STRING
    defaults: "us-east-1"
    description: AWS region where the VPC flow logs log group resides.
  - id: log_group_name
    type: STRING
    defaults: "/aws/vpc/flow-logs"
    description: CloudWatch Log Group containing VPC Flow Logs.
  - id: rejected_packets_threshold
    type: INT
    defaults: 1000
    description: Maximum acceptable count of rejected network packets over the
      15-minute window.

tasks:
  - id: query_vpc_rejected_traffic
    type: io.kestra.plugin.aws.cli.AwsCLI
    region: "{{ inputs.aws_region }}"
    accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
    secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
    commands:
      - |
        QUERY_STRING="filter action = 'REJECT' | stats sum(packets) as total_rejected_packets by srcAddr, dstPort | sort total_rejected_packets desc | limit 5"

        QUERY_ID=$(aws logs start-query \
          --log-group-name "{{ inputs.log_group_name }}" \
          --start-time $(date -u -v-15M +%s) \
          --end-time $(date -u +%s) \
          --query-string "$QUERY_STRING" \
          --query queryId --output text)

        sleep 5

        aws logs get-query-results --query-id "$QUERY_ID" --output json > /tmp/vpc_results.json

        TOTAL_REJECTED=$(jq '[.results[][] | select(.field == "total_rejected_packets") | .value | tonumber] | add // 0' /tmp/vpc_results.json)

        echo "::set-output name=total_rejected::$TOTAL_REJECTED::"

  - id: evaluate_rejected_traffic
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.query_vpc_rejected_traffic.vars.total_rejected | int >=
      inputs.rejected_packets_threshold }}"
    then:
      - id: notify_security_team
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK') }}"
        payload: |
          {
            "channel": "#security-alerts",
            "attachments": [
              {
                "color": "danger",
                "title": "🚨 AWS VPC Rejected Traffic Spike Alert",
                "text": "VPC Flow Logs in `{{ inputs.log_group_name }}` recorded *{{ outputs.query_vpc_rejected_traffic.vars.total_rejected }}* rejected packets over the last 15 minutes! Impact: High volume of security group/NACL drops indicates port scanning, lateral traversal attempts, or misconfigured security group egress rules.",
                "footer": "Kestra Security Sentinel • AWS VPC"
              }
            ]
          }

errors:
  - id: alert_pipeline_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {
        "channel": "#security-alerts",
        "attachments": [
          {
            "color": "danger",
            "title": "⚠️ AWS VPC Flow Logs Rejected Traffic Anomaly Sentinel Pipeline Failure",
            "text": "Failed to query CloudWatch Logs Insights for VPC flow logs. Verify log group name and IAM permissions.",
            "footer": "Kestra Workflow Failure Sentinel"
          }
        ]
      }

triggers:
  - id: periodic_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/15 * * * *"
    disabled: true