Configure Logstash 8 security pipelines with threat intelligence enrichment

Install Java and add the Elastic package repository

Logstash 8 requires a supported JDK. We use the bundled JVM shipped with Logstash, but the repository setup differs by distro.

sudo apt update && sudo apt install -y apt-transport-https gnupg curl
curl -fsSL https://artifacts.elastic.co/GPG-KEY-elasticsearch | sudo gpg --dearmor -o /usr/share/keyrings/elastic-keyring.gpg
echo "deb [signed-by=/usr/share/keyrings/elastic-keyring.gpg] https://artifacts.elastic.co/packages/8.x/apt stable main" | sudo tee /etc/apt/sources.list.d/elastic-8.x.list
sudo apt update

[elastic-8.x]
name=Elastic repository for 8.x packages
baseurl=https://artifacts.elastic.co/packages/8.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md

Install Logstash 8

This installs the Logstash service and default systemd unit.

sudo apt install -y logstash

sudo dnf install -y logstash

If you have not yet built the Elasticsearch backend, follow setting up centralized log aggregation with Elasticsearch, Logstash and Kibana first, and use Filebeat for log shipping as the upstream agent on firewalls and hosts that cannot run Logstash directly.

Create dedicated pipeline directories

Security pipelines should be isolated per source type so a bad regex in one does not stop ingestion for the others.

sudo mkdir -p /etc/logstash/conf.d/firewall
sudo mkdir -p /etc/logstash/conf.d/ids
sudo mkdir -p /etc/logstash/conf.d/syslog
sudo chown -R logstash:logstash /etc/logstash/conf.d
sudo chmod -R 750 /etc/logstash/conf.d

Define multiple pipelines in pipelines.yml

Logstash 8 runs several independent pipelines in one process. This isolates firewall, IDS and syslog processing and lets you tune workers per pipeline.

- pipeline.id: firewall
  path.config: "/etc/logstash/conf.d/firewall/*.conf"
  pipeline.workers: 4
  pipeline.batch.size: 250

- pipeline.id: ids
  path.config: "/etc/logstash/conf.d/ids/*.conf"
  pipeline.workers: 2
  pipeline.batch.size: 125

- pipeline.id: syslog
  path.config: "/etc/logstash/conf.d/syslog/*.conf"
  pipeline.workers: 2
  pipeline.batch.size: 125

Build the firewall ingest pipeline

This example parses syslog-formatted firewall logs (Cisco ASA style) arriving over TCP, and extracts source/destination IPs and ports.

input {
  tcp {
    port => 5140
    type => "firewall"
    ssl_enabled => true
    ssl_certificate_authorities => ["/etc/logstash/certs/ca.crt"]
    ssl_certificate => "/etc/logstash/certs/logstash.crt"
    ssl_key => "/etc/logstash/certs/logstash.key"
    ssl_client_authentication => "required"
  }
}

filter {
  if [type] == "firewall" {
    grok {
      match => { "message" => "%%{CISCOTIMESTAMP:timestamp} %%{HOSTNAME:device} %%{GREEDYDATA:asa_message}" }
    }
    grok {
      match => { "asa_message" => "Built %%{WORD:direction} %%{WORD:protocol} connection .* for %%{DATA} :%%{IP:src_ip}/%%{INT:src_port} .* to %%{DATA} :%%{IP:dst_ip}/%%{INT:dst_port}" }
      tag_on_failure => ["_grokparsefailure_firewall"]
    }
    date {
      match => [ "timestamp", "MMM dd HH:mm:ss", "MMM  d HH:mm:ss" ]
      target => "@timestamp"
    }
  }
}

Build the IDS ingest pipeline for Suricata/Snort EVE JSON

Most modern IDS engines emit structured JSON, which simplifies parsing considerably compared to firewall syslog.

input {
  file {
    path => "/var/log/suricata/eve.json"
    codec => "json"
    type => "ids"
    start_position => "beginning"
    sincedb_path => "/var/lib/logstash/sincedb_suricata"
  }
}

filter {
  if [type] == "ids" and [event_type] == "alert" {
    mutate {
      add_field => {
        "ioc_ip" => "%%{src_ip}"
        "signature" => "%%{[alert][signature]}"
        "severity" => "%%{[alert][severity]}"
      }
    }
  }
}

Build the syslog pipeline for generic infrastructure logs

This captures rsyslog and journald forwarded events from Linux hosts, network devices and application servers.

input {
  udp {
    port => 5514
    type => "syslog"
  }
}

If you are centralizing syslog on the source hosts before shipping to Logstash, see configuring centralized logging with rsyslog and logrotate.

Enrich IOCs with a local MISP indicator export using the translate filter

The translate filter performs fast in-memory dictionary lookups, ideal for large static IOC lists exported periodically from MISP as a CSV.

sudo mkdir -p /etc/logstash/threat-intel
sudo chown logstash:logstash /etc/logstash/threat-intel
sudo chmod 750 /etc/logstash/threat-intel

indicator,category,threat_level
203.0.113.45,c2-server,high
203.0.113.78,scanner,medium
198.51.100.23,malware-distribution,high

filter {
  if [type] == "firewall" and [dst_ip] {
    translate {
      dictionary_path => "/etc/logstash/threat-intel/misp_ip_iocs.csv"
      field => "dst_ip"
      destination => "[threat][misp_category]"
      fallback => "none"
      refresh_interval => 300
    }
  }
}

The refresh_interval reloads the dictionary every 5 minutes so a cron job or scheduled MISP export can update indicators without restarting Logstash.

Query AlienVault OTX in real time with the http filter

For IPs not already covered by your local IOC export, query OTX on demand. Store the API key outside version control.

a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6e7f8a9b0

sudo chown logstash:logstash /etc/logstash/secrets/otx.key
sudo chmod 600 /etc/logstash/secrets/otx.key

filter {
  if [threat][misp_category] == "none" and [dst_ip] {
    http {
      url => "https://otx.alienvault.com/api/v1/indicators/IPv4/%%{[dst_ip]}/general"
      headers => {
        "X-OTX-API-KEY" => "${OTX_API_KEY}"
      }
      target_body => "[otx][response]"
      ssl_verification_mode => "full"
    }
    if [otx][response][pulse_info][count] and [otx][response][pulse_info][count] > 0 {
      mutate {
        add_field => { "[threat][otx_flagged]" => "true" }
      }
    }
  }
}

Load the OTX API key as an environment variable

Reference secrets via environment variables rather than hardcoding them in pipeline configs.

[Service]
EnvironmentFile=/etc/logstash/secrets/logstash.env

OTX_API_KEY=a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6e7f8a9b0
ABUSEIPDB_API_KEY=b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6e7f8a9b0c1

sudo chmod 600 /etc/logstash/secrets/logstash.env
sudo chown logstash:logstash /etc/logstash/secrets/logstash.env
sudo systemctl daemon-reload

For teams already managing secrets centrally, integrate this with HashiCorp Vault secrets management instead of static env files.

Add AbuseIPDB reputation scoring

AbuseIPDB provides a confidence score useful for prioritizing alerts by abuse likelihood.

filter {
  if [threat][otx_flagged] != "true" and [dst_ip] {
    http {
      url => "https://api.abuseipdb.com/api/v2/check?ipAddress=%%{[dst_ip]}&maxAgeInDays=90"
      headers => {
        "Key" => "${ABUSEIPDB_API_KEY}"
        "Accept" => "application/json"
      }
      target_body => "[abuseipdb][response]"
      ssl_verification_mode => "full"
    }
    if [abuseipdb][response][data][abuseConfidenceScore] and [abuseipdb][response][data][abuseConfidenceScore] > 50 {
      mutate {
        add_field => { "[threat][abuseipdb_high_confidence]" => "true" }
      }
    }
  }
}

Add GeoIP and DNS enrichment for context

Geolocation and reverse DNS help analysts triage alerts faster without leaving the dashboard.

filter {
  if [dst_ip] {
    geoip {
      source => "dst_ip"
      target => "geoip"
      database => "/etc/logstash/geoip/GeoLite2-City.mmdb"
    }
    dns {
      reverse => ["dst_ip"]
      action => "append"
      hit_cache_size => 8000
      hit_cache_ttl => 300
      failed_cache_size => 1000
      failed_cache_ttl => 60
    }
  }
}

Download the free GeoLite2 City database from MaxMind and place it under /etc/logstash/geoip/ with 640 permissions owned by logstash:logstash.

Forward enriched events to Elasticsearch

This output uses a dedicated ILM-managed index and API key authentication instead of a shared superuser account.

output {
  elasticsearch {
    hosts => ["https://10.20.0.11:9200"]
    ssl_enabled => true
    ssl_certificate_authorities => ["/etc/logstash/certs/es-ca.crt"]
    api_key => "${ES_LOGSTASH_API_KEY}"
    index => "security-firewall-%%{+YYYY.MM.dd}"
    ilm_enabled => true
    ilm_rollover_alias => "security-firewall"
    ilm_pattern => "{now/d}-000001"
    ilm_policy => "security-events-policy"
  }
}

Pair this with Elasticsearch ILM configuration to automatically roll over and age out security indices, and review Elasticsearch snapshot and restore policies for long-term retention of security events.

Generate a minimal-privilege Elasticsearch API key

Create a role scoped only to the indices Logstash needs to write, rather than using the built-in superuser.

curl -X POST "https://10.20.0.11:9200/_security/api_key" \
  -H "Content-Type: application/json" \
  -u elastic:StrongElasticPass9! \
  --cacert /etc/logstash/certs/es-ca.crt \
  -d '{
    "name": "logstash-security-writer",
    "role_descriptors": {
      "logstash_writer": {
        "cluster": ["monitor"],
        "indices": [
          {
            "names": ["security-firewall-*", "security-ids-*", "security-syslog-*"],
            "privileges": ["create_index", "write", "manage"]
          }
        ]
      }
    }
  }'

Enable the Logstash monitoring API and X-Pack security

Expose pipeline metrics for capacity planning and detect stalled or slow pipelines before they drop events.

node.name: logstash-siem-01
path.data: /var/lib/logstash
pipeline.workers: 4
queue.type: persisted
queue.max_bytes: 4gb

xpack.monitoring.enabled: true
xpack.monitoring.elasticsearch.hosts: ["https://10.20.0.11:9200"]
xpack.monitoring.elasticsearch.api_key: "${ES_MONITORING_API_KEY}"

api.enabled: true
api.http.host: 127.0.0.1
api.http.port: 9600
api.auth.type: basic
api.auth.basic.username: monitor
api.auth.basic.password: "${LOGSTASH_MONITOR_PASS}"

queue.type: persisted enables a disk-backed queue so Logstash survives a restart without losing in-flight security events.

Restrict the monitoring API to localhost and firewall the pipeline ports

Never expose the monitoring API or ingest ports to the open network. Bind narrowly and firewall explicitly.

sudo ufw allow from 10.20.0.0/24 to any port 5140 proto tcp
sudo ufw allow from 10.20.0.0/24 to any port 5514 proto udp
sudo ufw deny 9600

sudo firewall-cmd --permanent --zone=internal --add-rich-rule='rule family="ipv4" source address="10.20.0.0/24" port protocol="tcp" port="5140" accept'
sudo firewall-cmd --permanent --zone=internal --add-rich-rule='rule family="ipv4" source address="10.20.0.0/24" port protocol="udp" port="5514" accept'
sudo firewall-cmd --reload

For deeper network-layer traffic visibility feeding these pipelines, see Kubernetes network monitoring with Hubble and Cilium if your firewall and IDS run in-cluster.

Enable and start Logstash

Start the service and confirm it picks up all three pipelines.

sudo systemctl daemon-reload
sudo systemctl enable --now logstash
sudo systemctl status logstash