ELK Stack — Project: Security Event Dashboard (SIEM-lite)

Project Overview

Security Information and Event Management (SIEM) is one of the most powerful use cases for the ELK Stack. In this project you will build a lightweight SIEM that:

• Collects authentication logs (SSH, sudo)
• Collects firewall logs (iptables)
• Collects web access logs with security focus
• Enriches events with GeoIP data for geographic analysis
• Detects brute force attacks, suspicious IPs, and geographic anomalies
• Sends alerts when security thresholds are breached

---

Architecture

┌─────────────────┐  ┌─────────────────┐  ┌─────────────────┐
│   Auth Logs     │  │  Firewall Logs  │  │  Web Access     │
│  (SSH, sudo)    │  │  (iptables)     │  │  Logs (Nginx)   │
└────────┬────────┘  └────────┬────────┘  └────────┬────────┘
         │                    │                    │
         ▼                    ▼                    ▼
┌──────────────────────────────────────────────────────────┐
│                      Filebeat                            │
│               (multiple inputs)                          │
└─────────────────────────┬────────────────────────────────┘
                          ▼
┌──────────────────────────────────────────────────────────┐
│                      Logstash                            │
│  ┌─────────┐  ┌──────────┐  ┌───────────┐  ┌─────────┐ │
│  │  Grok   │─▶│  GeoIP   │─▶│  Threat   │─▶│  Alert  │ │
│  │ Parsing │  │ Enrich   │  │ Detection │  │  Tags   │ │
│  └─────────┘  └──────────┘  └───────────┘  └─────────┘ │
└─────────────────────────┬────────────────────────────────┘
                          ▼
┌──────────────────────────────────────────────────────────┐
│                   Elasticsearch                          │
│  Indices: security-auth-* | security-firewall-*          │
│           security-web-*  | security-alerts-*            │
└─────────────────────────┬────────────────────────────────┘
                          ▼
┌──────────────────────────────────────────────────────────┐
│                      Kibana                              │
│  Security Dashboards | Alerting Rules | GeoIP Maps       │
└──────────────────────────────────────────────────────────┘

---

Project Structure

security-elk/
├── docker-compose.yml
├── filebeat/
│   └── filebeat.yml
├── logstash/
│   ├── pipeline/
│   │   └── security.conf
│   └── enrichment/
│       └── threat-ips.csv
├── log-generators/
│   ├── auth-generator.sh
│   ├── firewall-generator.sh
│   └── web-generator.sh
├── nginx/
│   └── nginx.conf
└── alerts/
    └── watcher-brute-force.json

---

Step 1 — Docker Compose

Create docker-compose.yml:

version: "3.8"

services:
  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.13.0
    container_name: es01
    environment:
      - discovery.type=single-node
      - xpack.security.enabled=false
      - "ES_JAVA_OPTS=-Xms1g -Xmx1g"
    ports:
      - "9200:9200"
    volumes:
      - es-data:/usr/share/elasticsearch/data
    networks:
      - elk
    healthcheck:
      test: ["CMD-SHELL", "curl -f http://localhost:9200/_cluster/health || exit 1"]
      interval: 10s
      timeout: 5s
      retries: 10

  logstash:
    image: docker.elastic.co/logstash/logstash:8.13.0
    container_name: logstash
    volumes:
      - ./logstash/pipeline/security.conf:/usr/share/logstash/pipeline/security.conf:ro
      - ./logstash/enrichment:/usr/share/logstash/enrichment:ro
    ports:
      - "5044:5044"
    environment:
      - "LS_JAVA_OPTS=-Xms512m -Xmx512m"
    networks:
      - elk
    depends_on:
      elasticsearch:
        condition: service_healthy

  kibana:
    image: docker.elastic.co/kibana/kibana:8.13.0
    container_name: kibana
    environment:
      - ELASTICSEARCH_HOSTS=http://es01:9200
    ports:
      - "5601:5601"
    networks:
      - elk
    depends_on:
      elasticsearch:
        condition: service_healthy

  nginx:
    image: nginx:1.25-alpine
    container_name: nginx-web
    volumes:
      - ./nginx/nginx.conf:/etc/nginx/nginx.conf:ro
      - nginx-logs:/var/log/nginx
    ports:
      - "8080:80"
    networks:
      - elk

  filebeat:
    image: docker.elastic.co/beats/filebeat:8.13.0
    container_name: filebeat
    user: root
    volumes:
      - ./filebeat/filebeat.yml:/usr/share/filebeat/filebeat.yml:ro
      - nginx-logs:/var/log/nginx:ro
      - generated-logs:/var/log/generated:ro
    networks:
      - elk
    depends_on:
      - logstash

  auth-generator:
    image: alpine:3.19
    container_name: auth-generator
    volumes:
      - ./log-generators/auth-generator.sh:/generate.sh:ro
      - generated-logs:/var/log/generated
    entrypoint: ["/bin/sh", "/generate.sh"]

  firewall-generator:
    image: alpine:3.19
    container_name: firewall-generator
    volumes:
      - ./log-generators/firewall-generator.sh:/generate.sh:ro
      - generated-logs:/var/log/generated
    entrypoint: ["/bin/sh", "/generate.sh"]

volumes:
  es-data:
  nginx-logs:
  generated-logs:

networks:
  elk:
    driver: bridge

---

Step 2 — Log Generators

Auth log generator

Create log-generators/auth-generator.sh:

#!/bin/sh

LOG_FILE="/var/log/generated/auth.log"
mkdir -p /var/log/generated

USERS=("admin" "root" "deploy" "ubuntu" "ec2-user" "test" "guest" "mysql")
VALID_USERS=("admin" "deploy" "ubuntu")
# Simulated external IPs (some are "suspicious")
IPS=("10.0.1.5" "192.168.1.100" "45.33.32.156" "198.51.100.42" "203.0.113.50" "185.220.101.1" "91.121.87.10" "172.16.0.5")

while true; do
  TIMESTAMP=$(date "+%b %d %H:%M:%S")
  HOST="web-server-01"
  USER=${USERS[$((RANDOM % ${#USERS[@]}))]}
  IP=${IPS[$((RANDOM % ${#IPS[@]}))]}
  PORT=$((RANDOM % 50000 + 1024))

  # 60% successful logins, 40% failures
  if [ $((RANDOM % 10)) -lt 6 ]; then
    echo "$TIMESTAMP $HOST sshd[$$]: Accepted publickey for $USER from $IP port $PORT ssh2" >> "$LOG_FILE"
  else
    echo "$TIMESTAMP $HOST sshd[$$]: Failed password for $USER from $IP port $PORT ssh2" >> "$LOG_FILE"
  fi

  # Simulate brute force: burst of failed logins from one IP
  if [ $((RANDOM % 20)) -eq 0 ]; then
    BRUTE_IP="185.220.101.1"
    for i in $(seq 1 15); do
      echo "$TIMESTAMP $HOST sshd[$$]: Failed password for root from $BRUTE_IP port $((RANDOM % 50000 + 1024)) ssh2" >> "$LOG_FILE"
    done
  fi

  # Sudo events
  if [ $((RANDOM % 5)) -eq 0 ]; then
    echo "$TIMESTAMP $HOST sudo: $USER : TTY=pts/0 ; PWD=/home/$USER ; USER=root ; COMMAND=/bin/systemctl restart nginx" >> "$LOG_FILE"
  fi

  sleep $((RANDOM % 3 + 1))
done

Firewall log generator

Create log-generators/firewall-generator.sh:

#!/bin/sh

LOG_FILE="/var/log/generated/firewall.log"
mkdir -p /var/log/generated

PROTOCOLS=("TCP" "UDP" "ICMP")
SRC_IPS=("10.0.1.5" "192.168.1.100" "45.33.32.156" "198.51.100.42" "203.0.113.50" "185.220.101.1" "91.121.87.10")
DST_PORTS=("22" "80" "443" "3306" "5432" "8080" "6379" "27017" "9200")
ACTIONS=("ACCEPT" "ACCEPT" "ACCEPT" "DROP" "DROP" "REJECT")

while true; do
  TIMESTAMP=$(date "+%b %d %H:%M:%S")
  HOST="firewall-01"
  SRC=${SRC_IPS[$((RANDOM % ${#SRC_IPS[@]}))]}
  DST_PORT=${DST_PORTS[$((RANDOM % ${#DST_PORTS[@]}))]}
  PROTO=${PROTOCOLS[$((RANDOM % ${#PROTOCOLS[@]}))]}
  ACTION=${ACTIONS[$((RANDOM % ${#ACTIONS[@]}))]}
  SRC_PORT=$((RANDOM % 50000 + 1024))

  echo "$TIMESTAMP $HOST kernel: iptables $ACTION: IN=eth0 OUT= SRC=$SRC DST=10.0.0.1 LEN=60 TOS=0x00 PREC=0x00 TTL=64 PROTO=$PROTO SPT=$SRC_PORT DPT=$DST_PORT" >> "$LOG_FILE"

  # Simulate port scan: many different ports from one IP
  if [ $((RANDOM % 25)) -eq 0 ]; then
    SCAN_IP="91.121.87.10"
    for port in 21 22 23 25 53 80 110 143 443 993 995 3306 3389 5432 8080; do
      echo "$TIMESTAMP $HOST kernel: iptables DROP: IN=eth0 OUT= SRC=$SCAN_IP DST=10.0.0.1 LEN=40 TOS=0x00 PREC=0x00 TTL=50 PROTO=TCP SPT=$((RANDOM % 50000 + 1024)) DPT=$port" >> "$LOG_FILE"
    done
  fi

  sleep $((RANDOM % 2 + 1))
done

---

Step 3 — Threat IP List

Create logstash/enrichment/threat-ips.csv:

ip,threat_type,threat_score
185.220.101.1,tor_exit_node,90
91.121.87.10,known_scanner,85
198.51.100.42,botnet_c2,95
203.0.113.50,spam_source,60

---

Step 4 — Filebeat Configuration

Create filebeat/filebeat.yml:

filebeat.inputs:
  - type: log
    enabled: true
    paths:
      - /var/log/generated/auth.log
    fields:
      log_type: auth
    fields_under_root: true

  - type: log
    enabled: true
    paths:
      - /var/log/generated/firewall.log
    fields:
      log_type: firewall
    fields_under_root: true

  - type: log
    enabled: true
    paths:
      - /var/log/nginx/access.log
    fields:
      log_type: web_access
    fields_under_root: true

output.logstash:
  hosts: ["logstash:5044"]

logging.level: info

---

Step 5 — Security-Focused Logstash Pipeline

Create logstash/pipeline/security.conf:

input {
  beats {
    port => 5044
  }
}

filter {

  # ──────────────────────────────
  # AUTH LOG PROCESSING
  # ──────────────────────────────
  if [log_type] == "auth" {
    grok {
      match => {
        "message" => [
          "%{SYSLOGTIMESTAMP:timestamp} %{HOSTNAME:host} sshd\[%{POSINT:pid}\]: %{WORD:auth_result} %{WORD:auth_method} for (invalid user )?%{USERNAME:username} from %{IP:source_ip} port %{POSINT:source_port}",
          "%{SYSLOGTIMESTAMP:timestamp} %{HOSTNAME:host} sudo: %{USERNAME:username} : TTY=%{DATA:tty} ; PWD=%{DATA:pwd} ; USER=%{USERNAME:target_user} ; COMMAND=%{GREEDYDATA:command}"
        ]
      }
      tag_on_failure => ["_auth_parse_failure"]
    }

    if [auth_result] == "Failed" {
      mutate {
        add_tag => ["failed_login"]
        add_field => { "security_event" => "failed_authentication" }
      }
    } else if [auth_result] == "Accepted" {
      mutate {
        add_field => { "security_event" => "successful_authentication" }
      }
    }

    if [command] {
      mutate {
        add_field => { "security_event" => "privilege_escalation" }
        add_tag => ["sudo_event"]
      }
    }

    # GeoIP enrichment on source IP
    if [source_ip] {
      geoip {
        source => "source_ip"
        target => "geoip"
      }
    }

    mutate {
      add_field => { "[@metadata][index]" => "security-auth" }
    }
  }

  # ──────────────────────────────
  # FIREWALL LOG PROCESSING
  # ──────────────────────────────
  else if [log_type] == "firewall" {
    grok {
      match => {
        "message" => "%{SYSLOGTIMESTAMP:timestamp} %{HOSTNAME:host} kernel: iptables %{WORD:fw_action}: IN=%{DATA:in_interface} OUT=%{DATA:out_interface} SRC=%{IP:source_ip} DST=%{IP:dest_ip} LEN=%{NUMBER:packet_length:integer} TOS=%{DATA:tos} PREC=%{DATA:prec} TTL=%{NUMBER:ttl:integer} PROTO=%{WORD:protocol}( SPT=%{NUMBER:source_port:integer})?( DPT=%{NUMBER:dest_port:integer})?"
      }
      tag_on_failure => ["_firewall_parse_failure"]
    }

    if [fw_action] == "DROP" or [fw_action] == "REJECT" {
      mutate {
        add_tag => ["blocked_traffic"]
        add_field => { "security_event" => "blocked_connection" }
      }
    }

    # Tag connections to sensitive ports
    if [dest_port] {
      if [dest_port] in [22, 3306, 5432, 6379, 9200, 27017] {
        mutate {
          add_tag => ["sensitive_port_access"]
          add_field => { "security_event" => "sensitive_port_access" }
        }
      }
    }

    # GeoIP enrichment
    if [source_ip] {
      geoip {
        source => "source_ip"
        target => "geoip"
      }
    }

    mutate {
      add_field => { "[@metadata][index]" => "security-firewall" }
    }
  }

  # ──────────────────────────────
  # WEB ACCESS LOG PROCESSING
  # ──────────────────────────────
  else if [log_type] == "web_access" {
    grok {
      match => {
        "message" => '%{IPORHOST:source_ip} - %{DATA:remote_user} \[%{HTTPDATE:timestamp}\] "%{WORD:http_method} %{URIPATHPARAM:request_uri} HTTP/%{NUMBER:http_version}" %{NUMBER:response_code:integer} %{NUMBER:bytes_sent:integer} "%{DATA:referrer}" "%{DATA:user_agent}"'
      }
      tag_on_failure => ["_web_parse_failure"]
    }

    # Detect suspicious patterns
    if [request_uri] =~ /(\.\.|\/etc\/|\/proc\/|union\s+select|<script|eval\(|base64_decode)/ {
      mutate {
        add_tag => ["suspicious_request", "possible_attack"]
        add_field => { "security_event" => "suspicious_web_request" }
      }
    }

    # Detect scanners via user agent
    if [user_agent] =~ /(sqlmap|nikto|nmap|masscan|dirbuster|gobuster)/ {
      mutate {
        add_tag => ["scanner_detected"]
        add_field => { "security_event" => "scanner_detected" }
      }
    }

    # GeoIP enrichment
    if [source_ip] {
      geoip {
        source => "source_ip"
        target => "geoip"
      }
    }

    mutate {
      add_field => { "[@metadata][index]" => "security-web" }
    }
  }

  # ──────────────────────────────
  # THREAT ENRICHMENT (all types)
  # ──────────────────────────────
  if [source_ip] {
    translate {
      field => "source_ip"
      destination => "[threat][type]"
      dictionary_path => "/usr/share/logstash/enrichment/threat-ips.csv"
      fallback => "unknown"
      refresh_interval => 300
    }

    if [threat][type] != "unknown" {
      mutate {
        add_tag => ["known_threat"]
      }
    }
  }

  # Clean up
  mutate {
    remove_field => ["message"]
  }
}

output {
  # Main indices
  elasticsearch {
    hosts => ["http://es01:9200"]
    index => "%{[@metadata][index]}-%{+YYYY.MM.dd}"
  }

  # Duplicate alerts to a dedicated index
  if "known_threat" in [tags] or "possible_attack" in [tags] or "failed_login" in [tags] {
    elasticsearch {
      hosts => ["http://es01:9200"]
      index => "security-alerts-%{+YYYY.MM.dd}"
    }
  }
}

---

Step 6 — Nginx Configuration

Create nginx/nginx.conf:

worker_processes auto;
error_log /var/log/nginx/error.log warn;
pid       /var/run/nginx.pid;

events {
    worker_connections 1024;
}

http {
    include       /etc/nginx/mime.types;
    default_type  application/octet-stream;

    log_format main '$remote_addr - $remote_user [$time_local] '
                    '"$request" $status $body_bytes_sent '
                    '"$http_referer" "$http_user_agent"';

    access_log /var/log/nginx/access.log main;

    server {
        listen 80;
        server_name localhost;

        location / {
            return 200 "Security Dashboard Backend";
        }
    }
}

---

Step 7 — Launch and Generate Data

# Start the stack
docker-compose up -d

# Wait for services
sleep 40

# Generate web traffic (including some suspicious requests)
for i in $(seq 1 100); do
  curl -s http://localhost:8080/ > /dev/null
  curl -s http://localhost:8080/admin > /dev/null 2>&1
  curl -s "http://localhost:8080/../etc/passwd" > /dev/null 2>&1
  curl -s -A "nikto/2.1.6" http://localhost:8080/ > /dev/null 2>&1
done

echo "Generated web traffic including suspicious requests"

---

Step 8 — Verify Security Data

Check indices

curl -s http://localhost:9200/_cat/indices?v | grep security

Query failed logins

curl -s "http://localhost:9200/security-auth-*/_search?pretty" -H 'Content-Type: application/json' -d'
{
  "size": 5,
  "query": {
    "term": { "security_event": "failed_authentication" }
  }
}'

Query known threats

curl -s "http://localhost:9200/security-alerts-*/_search?pretty" -H 'Content-Type: application/json' -d'
{
  "size": 5,
  "query": {
    "term": { "tags": "known_threat" }
  }
}'

Brute force detection query

curl -s "http://localhost:9200/security-auth-*/_search?pretty" -H 'Content-Type: application/json' -d'
{
  "size": 0,
  "query": {
    "bool": {
      "must": [
        { "term": { "security_event": "failed_authentication" } },
        { "range": { "@timestamp": { "gte": "now-1h" } } }
      ]
    }
  },
  "aggs": {
    "brute_force_ips": {
      "terms": {
        "field": "source_ip.keyword",
        "min_doc_count": 10,
        "order": { "_count": "desc" }
      }
    }
  }
}'

---

Step 9 — Kibana Security Dashboards

Open Kibana at http://localhost:5601.

Data Views

Create data views for:

• security-auth-*
• security-firewall-*
• security-web-*
• security-alerts-*

Dashboard: Security Overview

Panel	Type	Data View	Description
Security Events Over Time	Area chart	security-*	All events by type
Failed Logins	Metric	security-auth-*	Count where security_event = failed_authentication
Blocked Connections	Metric	security-firewall-*	Count where fw_action = DROP
Threat Source Map	Map	security-alerts-*	GeoIP coordinates of known threats
Top Attacking IPs	Horizontal bar	security-alerts-*	Terms on source_ip
Suspicious Web Requests	Data table	security-web-*	Filter tags: possible_attack
Brute Force Detection	Line chart	security-auth-*	Failed logins per IP over time
Sensitive Port Access	Pie chart	security-firewall-*	Terms on dest_port with sensitive_port_access tag

---

Step 10 — Alerting Rules

Brute force alert (Kibana Alerting)

In Kibana go to Stack Management → Rules and create:

• Name: Brute Force Detection
• Check every: 1 minute
• Conditions: Elasticsearch query — count of documents in security-auth-* where security_event = failed_authentication AND grouped by source_ip exceeds 10 in the last 5 minutes
• Action: Log or email notification

Known threat alert

• Name: Known Threat IP Activity
• Check every: 1 minute
• Conditions: Elasticsearch query — any new document in security-alerts-* with tag known_threat
• Action: Log or webhook notification

---

Cleanup

docker-compose down -v

---

What You Learned

• How to build a security-focused log pipeline
• How to parse authentication, firewall, and web logs for security events
• How to enrich logs with GeoIP data and threat intelligence
• How to detect brute force attacks and suspicious activity
• How to build SIEM-style dashboards in Kibana
• How to configure alerting rules for security events

---

Next

→ Practice Exercises & Challenges