CrowdSec: Community-Powered Intrusion Prevention

Modern homelabs face the same threats as enterprise infrastructure: brute force attacks, bot traffic, port scanning, and web exploits. Traditional security tools like Fail2ban provide basic protection but lack the intelligence of community-backed threat detection. Enter CrowdSec—a modern, open-source intrusion prevention system that combines real-time behavioral analysis with crowdsourced threat intelligence from thousands of global users.

This guide will walk you through deploying a production-ready CrowdSec security engine with bouncers, integrating it with your existing homelab services, and exploring AI automation opportunities for intelligent threat response.

Prerequisites

Before starting, ensure you have:

Required:

• Docker and Docker Compose installed
• Basic understanding of TCP/IP networking and firewalls
• Linux familiarity and SSH access to your homelab server
• At least 1 GB free RAM and 100 MB disk space
• Running services to protect (Nginx, SSH, etc.)

Optional:

• Prometheus and Grafana for monitoring
• Cloudflare account for CDN-level protection
• Python knowledge for automation scripts

Time to Complete: 45 minutes Difficulty Level: Intermediate

---

Understanding CrowdSec Architecture

What is CrowdSec?

CrowdSec is an open-source (MIT licensed) intrusion prevention system written in Go that analyzes logs to detect malicious behavior patterns. Unlike Fail2ban’s reactive, regex-based approach, CrowdSec provides:

• Real-time behavioral threat detection using advanced scenarios
• Community-powered blocklists shared across thousands of users worldwide
• Lightweight, modular architecture (100-200 MB RAM)
• Docker-native deployment for containerized environments
• REST API for programmatic control and integration
• Multi-service protection from a single security engine

Core Architecture Components

CrowdSec operates on a modular architecture with clearly separated concerns:

Raw Logs → Parsers → Scenarios → Local API → Bouncers → Enforcement
                         ↓
                   Central API (opt-in)
                         ↓
                 Community Blocklists

Security Engine (Agent): Runs on your servers, analyzes logs in real-time, and detects malicious behavior using parsers and scenarios. Written in Go for lightweight resource usage.

Parsers: YAML-based rules that normalize different log formats into standardized events. Organized in stages (s00-raw, s01-parse, s02-enrich) to extract relevant information for analysis.

Scenarios: Define what constitutes malicious behavior using the “leaky bucket” principle. For example: 10 failed SSH logins in 1 minute triggers an alert. Configurable parameters include capacity (threshold), leak speed (rate), and blackhole (cooldown period).

Bouncers: Remediation components that enforce decisions by blocking IPs, presenting CAPTCHAs, or rate limiting. They query the Local API for decisions but don’t perform detection themselves.

Local API (LAPI): REST interface between Security Engine and Bouncers. Manages authentication tokens and stores decisions in a database. Default: localhost-only for security.

Central API (CAPI): Cloud-based threat intelligence aggregation that collects anonymized signals from all CrowdSec instances, scores threats, and distributes community blocklists.

The Crowdsourced Intelligence Model

CrowdSec’s defining feature is its collaborative security model:

1. Local Detection: Your CrowdSec instance detects an SSH brute force from IP 203.0.113.45
2. Local Blocking: The threat is immediately blocked on your system
3. Central Reporting: The IP is sent to CrowdSec Central API (anonymized, opt-in)
4. Community Aggregation: Central system analyzes reports from 1000+ CrowdSec instances
5. Threat Scoring: IPs flagged by multiple independent sources receive higher confidence scores
6. Blocklist Distribution: High-confidence malicious IPs are added to community blocklists
7. Community Protection: All users receive updated lists, blocking threats before they attack

Privacy safeguards: Only anonymized IP addresses and scenario names are shared—no log data, payloads, or credentials leave your server.

CrowdSec vs Fail2ban

Recommendation: Use CrowdSec for production homelabs with multiple exposed services where you want community-backed protection. Use Fail2ban for edge servers, development environments, or SSH-only protection with minimal setup.

---

Installation and Initial Setup

Docker Deployment

Create a project directory and docker-compose configuration:

# Create project directory
mkdir ~/crowdsec && cd ~/crowdsec

# Create docker-compose.yml
cat > docker-compose.yml <<'EOF'
version: '3.8'

services:
  crowdsec:
    image: crowdsecurity/crowdsec:latest
    container_name: crowdsec
    environment:
      - COLLECTIONS=crowdsecurity/linux crowdsecurity/sshd
      - SCENARIOS=crowdsecurity/base-http-scenarios
    volumes:
      # Mount host logs (read-only)
      - /var/log:/var/log:ro
      # Persistent database
      - crowdsec_db:/var/lib/crowdsec/data
      # Configuration directory
      - crowdsec_config:/etc/crowdsec
    networks:
      - crowdsec_network
    ports:
      - "6060:6060"  # Prometheus metrics
    restart: unless-stopped

volumes:
  crowdsec_db:
  crowdsec_config:

networks:
  crowdsec_network:
    driver: bridge
EOF

# Start the service
docker-compose up -d

# Verify running status
docker-compose logs -f crowdsec

Validation checklist:

• Container running without errors
• No permission denied messages
• Log files being accessed

# Check version
docker-compose exec crowdsec cscli version

# View metrics (should show activity)
docker-compose exec crowdsec cscli metrics

Installing Collections and Scenarios

Collections bundle parsers and scenarios for specific services. Install common protection bundles:

# SSH protection
docker-compose exec crowdsec cscli collections install crowdsecurity/sshd

# Nginx/web server protection
docker-compose exec crowdsec cscli collections install crowdsecurity/nginx

# General HTTP attack detection
docker-compose exec crowdsec cscli collections install crowdsecurity/base-http-scenarios

# WordPress-specific protection (if applicable)
docker-compose exec crowdsec cscli collections install crowdsecurity/wordpress

# Update hub to latest versions
docker-compose exec crowdsec cscli hub update

# Verify installation
docker-compose exec crowdsec cscli collections list

Configuring Log Acquisition

CrowdSec needs to know which log files to monitor. Create acquisition configurations in the acquis.d directory:

Nginx Log Monitoring:

docker-compose exec crowdsec sh -c 'cat > /etc/crowdsec/acquis.d/nginx.yaml <<EOF
source: file
filenames:
  - /var/log/nginx/access.log
  - /var/log/nginx/error.log
labels:
  type: nginx
EOF'

SSH Log Monitoring:

docker-compose exec crowdsec sh -c 'cat > /etc/crowdsec/acquis.d/sshd.yaml <<EOF
source: file
filenames:
  - /var/log/auth.log
  - /var/log/secure
labels:
  type: syslog
EOF'

Docker Container Monitoring:

docker-compose exec crowdsec sh -c 'cat > /etc/crowdsec/acquis.d/docker.yaml <<EOF
source: docker
labels:
  type: docker-compose
EOF'

Restart CrowdSec to apply changes:

docker-compose restart crowdsec

Verify log files are being read:

# Check parser activity
docker-compose exec crowdsec cscli metrics

# Monitor real-time logs
docker-compose logs -f crowdsec

---

Understanding Parsers and Scenarios

Parsers: Normalizing Log Data

Parsers convert diverse log formats into standardized events that scenarios can analyze. They’re organized in processing stages:

1. s00-raw: Base format parsing (syslog, JSON)
2. s01-parse: Service-specific parsing (Nginx, SSH, Apache)
3. s02-enrich: Data enrichment, whitelisting, GeoIP
4. s03-postprocess: Additional filtering

Example: Custom Application Parser

Create a parser for custom application authentication logs:

# /etc/crowdsec/parsers/s02-enrich/custom-app-auth.yaml
name: custom/app-auth-parser
description: "Parse custom application auth logs"

# When to apply this parser
filter: "evt.Line.Raw startswith '[AUTH]'"

# Tag what this parser handles
statics:
  - parsed: log_type
    value: app_auth
  - meta: service
    value: custom-app

# Extraction rules using Grok patterns
nodes:
  - grok:
      pattern: "\[AUTH\] %{TIMESTAMP_ISO8601:timestamp} \| %{WORD:action} \| %{IP:source_ip} \| %{USERNAME:user}"
      expression: evt.Line.Raw
      onsuccess: next
      onfail: end

Common Grok patterns:

• %{IP:source_ip} - Extract IP address
• %{WORD:action} - Single word (FAILED, SUCCESS)
• %{INT:count} - Integer (attempt counts)
• %{TIMESTAMP_ISO8601:timestamp} - ISO timestamps
• %{GREEDYDATA:message} - Remaining text

Scenarios: Defining Malicious Behavior

Scenarios define what constitutes an attack using the “leaky bucket” algorithm:

Leaky Bucket Concept:

• Events flow into bucket at varying rates
• Bucket “leaks” at set interval (leakspeed)
• Alert when bucket overflows (capacity exceeded)
• Prevents spam with blackhole (cooldown period)

Scenario Types:

1. Leaky Bucket (Most Common)

Best for sustained attacks like brute force:

# /etc/crowdsec/scenarios/custom/ssh-protection.yaml
name: custom/ssh-brute-force-enhanced
description: "Detect SSH brute force attacks"
type: leaky

filter: "evt.Meta.log_type == 'ssh' && evt.Parsed.action == 'FAILED'"

leakspeed: 30s      # Bucket leaks every 30 seconds
capacity: 5         # Alert after 5 failed attempts
groupby: evt.Meta.source_ip
blackhole: 2h       # 2-hour cooldown after trigger

labels:
  severity: high
  category: authentication
  service: ssh
  remediation: block_ip

2. Counter

Best for volume-based attacks like DDoS:

name: custom/http-dos-detection
type: counter

filter: "evt.Meta.log_type == 'http'"
count: 100          # 100+ requests
window: 1m          # Within 1 minute
groupby: evt.Meta.source_ip

labels:
  severity: medium
  category: dos

3. Custom Scenario: Directory Enumeration

Detect reconnaissance via excessive 404 errors:

# /etc/crowdsec/scenarios/custom/excessive-404.yaml
name: custom/directory-enumeration
description: "Detect directory enumeration via 404 errors"

filter: "evt.Parsed.http_code == 404"
type: leaky

leakspeed: 30s
capacity: 20        # 20 404s in 30 seconds = suspicious
groupby: "evt.Meta.source_ip"
blackhole: 1h

labels:
  service: web
  attack_type: reconnaissance
  severity: medium
  category: enum

Deploy custom scenario:

# Copy scenario to container
docker cp excessive-404.yaml crowdsec:/etc/crowdsec/scenarios/custom/

# Restart to load new scenario
docker-compose restart crowdsec

# Monitor alerts for this scenario
docker-compose exec crowdsec cscli alerts list --filter "scenario=custom/directory-enumeration"

---

Bouncers and Enforcement

Bouncers are the enforcement layer that takes action on CrowdSec decisions. They’re separate from the detection engine and query the Local API for decisions.

Creating Bouncer API Tokens

Each bouncer requires an API token to authenticate with the Local API:

# List existing bouncers
docker-compose exec crowdsec cscli bouncers list

# Create new bouncer token
docker-compose exec crowdsec cscli bouncers add FirewallBouncer

# Output will show: API Key: YOUR_API_KEY_HERE

# Save to environment file (never commit to git)
echo "FIREWALL_BOUNCER_KEY=YOUR_API_KEY_HERE" >> .env

Firewall Bouncer: IP-Level Blocking

The firewall bouncer blocks malicious IPs at the network layer using iptables/nftables:

Add to docker-compose.yml:

  firewall-bouncer:
    image: crowdsecurity/firewall-bouncer:latest
    container_name: firewall-bouncer
    environment:
      - CROWDSEC_LAPI_HOST=http://crowdsec:8080
      - CROWDSEC_LAPI_KEY=${FIREWALL_BOUNCER_KEY}
      - BOUNCER_NAME=FirewallBouncer
    depends_on:
      - crowdsec
    networks:
      - crowdsec_network
    cap_add:
      - NET_ADMIN  # Required for iptables access
    restart: unless-stopped

Deploy and verify:

# Recreate with new bouncer
docker-compose up -d

# Check bouncer is running
docker-compose logs -f firewall-bouncer

# Verify iptables rules
docker-compose exec firewall-bouncer iptables -L -n | grep crowdsec

# Check bouncer connectivity
docker-compose exec crowdsec cscli bouncers list

Nginx Bouncer: Application-Layer Protection

Protect Nginx reverse proxy at the HTTP layer:

Docker Compose configuration:

  nginx:
    image: nginx:alpine
    container_name: nginx
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf:ro
      - /var/log/nginx:/var/log/nginx
    networks:
      - crowdsec_network
    depends_on:
      - crowdsec
    restart: unless-stopped

  nginx-bouncer:
    image: crowdsecurity/nginx-bouncer:latest
    container_name: nginx-bouncer
    environment:
      CROWDSEC_LAPI_HOST: "http://crowdsec:8080"
      CROWDSEC_LAPI_KEY: "${NGINX_BOUNCER_KEY}"
    depends_on:
      - crowdsec
    networks:
      - crowdsec_network
    restart: unless-stopped

Traefik Bouncer: Reverse Proxy Middleware

Protect Traefik-based services with CrowdSec middleware:

  traefik:
    image: traefik:latest
    environment:
      - CROWDSEC_ENABLED=true
      - CROWDSEC_LAPI_HOST=http://crowdsec:8080
      - CROWDSEC_LAPI_KEY=${TRAEFIK_BOUNCER_KEY}
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
    networks:
      - crowdsec_network

  myapp:
    image: myapp:latest
    labels:
      # Enable CrowdSec protection for this app
      - "traefik.http.middlewares.crowdsec.plugin.crowdsec.enabled=true"
      - "traefik.http.routers.app.middlewares=crowdsec"
    networks:
      - crowdsec_network

---

Real-World Integration Scenarios

Protecting Nginx Proxy Manager

Nginx Proxy Manager (NPM) is popular in homelabs for reverse proxying. Integrate CrowdSec for comprehensive protection:

version: '3.8'

services:
  npm:
    image: 'jc21/nginx-proxy-manager:latest'
    container_name: npm
    ports:
      - '80:80'
      - '443:443'
      - '81:81'  # Admin interface
    volumes:
      - ./npm-data:/data
      - ./npm-letsencrypt:/etc/letsencrypt
      - /var/log/npm:/var/log/npm
    networks:
      - crowdsec_network
    restart: unless-stopped

  crowdsec:
    image: crowdsecurity/crowdsec:latest
    container_name: crowdsec
    environment:
      - COLLECTIONS=crowdsecurity/nginx
    volumes:
      - /var/log/npm:/var/log/npm:ro  # Share NPM logs
      - /var/log:/var/log:ro
      - crowdsec_db:/var/lib/crowdsec/data
      - crowdsec_config:/etc/crowdsec
    networks:
      - crowdsec_network
    restart: unless-stopped

  firewall-bouncer:
    image: crowdsecurity/firewall-bouncer:latest
    container_name: firewall-bouncer
    environment:
      - CROWDSEC_LAPI_HOST=http://crowdsec:8080
      - CROWDSEC_LAPI_KEY=${FIREWALL_BOUNCER_KEY}
    depends_on:
      - crowdsec
    networks:
      - crowdsec_network
    cap_add:
      - NET_ADMIN
    restart: unless-stopped

volumes:
  crowdsec_db:
  crowdsec_config:

networks:
  crowdsec_network:
    driver: bridge

Configure NPM log acquisition:

docker-compose exec crowdsec sh -c 'cat > /etc/crowdsec/acquis.d/npm.yaml <<EOF
source: file
filenames:
  - /var/log/npm/proxy-host-*.log
  - /var/log/npm/error.log
labels:
  type: nginx
EOF'

docker-compose restart crowdsec

SSH Service Protection

Protect SSH from brute force attacks:

# Install SSH collection
docker-compose exec crowdsec cscli collections install crowdsecurity/sshd

# Verify SSH scenarios are active
docker-compose exec crowdsec cscli scenarios list | grep ssh

# Test SSH protection (intentional failures)
for i in {1..15}; do
  ssh invaliduser@localhost 2>/dev/null
done

# Check alerts
docker-compose exec crowdsec cscli alerts list

# Verify IP was blocked
docker-compose exec crowdsec cscli decisions list

Nextcloud Protection

Protect Nextcloud from brute force and file abuse:

# Install Nextcloud collection
docker-compose exec crowdsec cscli collections install crowdsecurity/nextcloud

# Configure log acquisition
docker-compose exec crowdsec sh -c 'cat > /etc/crowdsec/acquis.d/nextcloud.yaml <<EOF
source: file
filenames:
  - /var/www/nextcloud/data/nextcloud.log
labels:
  type: nextcloud
EOF'

docker-compose restart crowdsec

Protected attack types:

• Failed login attempts
• File upload abuse
• API rate limiting violations
• Account enumeration

Home Assistant Integration

Protect Home Assistant from unauthorized access:

Custom scenario for Home Assistant:

# /etc/crowdsec/scenarios/custom/homeassistant-bf.yaml
name: custom/home-assistant-brute-force
description: "Detect Home Assistant failed login attempts"

filter: "evt.Parsed.message contains 'Invalid authentication'"
type: leaky

leakspeed: 1m
capacity: 5
groupby: evt.Meta.source_ip
blackhole: 30m

labels:
  service: home-assistant
  severity: high
  category: authentication

# Configure log acquisition
docker-compose exec crowdsec sh -c 'cat > /etc/crowdsec/acquis.d/homeassistant.yaml <<EOF
source: file
filenames:
  - /root/.homeassistant/home-assistant.log
labels:
  type: home-assistant
EOF'

# Deploy custom scenario
docker cp homeassistant-bf.yaml crowdsec:/etc/crowdsec/scenarios/custom/

docker-compose restart crowdsec

---

Monitoring and Metrics

CrowdSec Console (Official Dashboard)

Connect to the web-based management console:

# 1. Create account at https://app.crowdsec.net
# 2. Generate enrollment key from console settings
# 3. Register your CrowdSec instance:

docker-compose exec crowdsec cscli console enroll YOUR_ENROLLMENT_KEY

# 4. Verify enrollment status
docker-compose exec crowdsec cscli console status

# 5. Access console at https://app.crowdsec.net/engines

Console features:

• Real-time alert visualization
• Threat trends and patterns
• Bouncer connectivity status
• Community blocklist metrics
• Historical threat analysis
• Alert filtering and search

Prometheus and Grafana Integration

Export CrowdSec metrics for time-series analysis:

Enable Prometheus metrics:

docker-compose exec crowdsec sh -c 'cat >> /etc/crowdsec/config.yaml <<EOF
telemetry:
  prometheus:
    enabled: true
    level: full
    listen_addr: 0.0.0.0
    listen_port: 6060
EOF'

docker-compose restart crowdsec

Deploy Prometheus and Grafana:

# Add to docker-compose.yml
  prometheus:
    image: prom/prometheus:latest
    container_name: prometheus
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml:ro
      - prometheus-data:/prometheus
    ports:
      - "9090:9090"
    networks:
      - crowdsec_network
    restart: unless-stopped

  grafana:
    image: grafana/grafana:latest
    container_name: grafana
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin
      - GF_USERS_ALLOW_SIGN_UP=false
    ports:
      - "3000:3000"
    volumes:
      - grafana-data:/var/lib/grafana
    depends_on:
      - prometheus
    networks:
      - crowdsec_network
    restart: unless-stopped

volumes:
  prometheus-data:
  grafana-data:

Create prometheus.yml:

global:
  scrape_interval: 15s

scrape_configs:
  - job_name: 'crowdsec'
    static_configs:
      - targets: ['crowdsec:6060']

Import Grafana dashboards:

1. Access Grafana at http://localhost:3000 (admin/admin)
2. Go to Dashboards → Import
3. Import popular CrowdSec dashboards:
   • ID 21419: CrowdSec Metrics (overview)
   • ID 19011: CrowdSec Insights (detailed analysis)
   • ID 19013: Local API Metrics
4. Select Prometheus as data source

CLI-Based Monitoring

Monitor CrowdSec from the command line:

# Display real-time metrics
docker-compose exec crowdsec cscli metrics

# View recent alerts (last 20)
docker-compose exec crowdsec cscli alerts list --limit 20

# Filter alerts by scenario
docker-compose exec crowdsec cscli alerts list --filter "scenario~ssh"

# View active decisions (blocked IPs)
docker-compose exec crowdsec cscli decisions list

# Query if specific IP is blocked
docker-compose exec crowdsec cscli decisions list -i 203.0.113.45

# Remove decision (whitelist IP)
docker-compose exec crowdsec cscli decisions delete -i 192.168.1.100

# Check Local API status
docker-compose exec crowdsec cscli lapi status

# View bouncer activity
docker-compose exec crowdsec cscli bouncers list

---

AI Automation and Intelligent Threat Response

CrowdSec’s REST API and structured data format enable powerful AI-driven automation. Here are practical implementations for intelligent threat response:

Automated Threat Analysis with Python

Build a Python script that analyzes CrowdSec alerts and escalates responses based on severity:

#!/usr/bin/env python3
"""
CrowdSec Intelligent Threat Analyzer
Analyzes alerts from Local API and escalates responses
"""

import requests
import json
from datetime import datetime, timedelta

CROWDSEC_URL = "http://localhost:8080"
API_KEY = "your-api-key"  # From bouncer token

headers = {
    "X-API-Key": API_KEY,
    "Content-Type": "application/json"
}

def fetch_recent_alerts(minutes=5):
    """Fetch alerts from last N minutes"""
    response = requests.get(
        f"{CROWDSEC_URL}/v1/alerts?limit=100",
        headers=headers
    )

    if response.status_code != 200:
        print(f"Error fetching alerts: {response.status_code}")
        return []

    alerts = response.json()

    # Filter by time
    cutoff = datetime.utcnow() - timedelta(minutes=minutes)
    recent = [
        a for a in alerts
        if datetime.fromisoformat(a['created_at'].replace('Z', '+00:00')) > cutoff
    ]

    return recent

def analyze_threat_severity(alert):
    """Calculate threat severity score"""
    score = 0

    # High event count = more severe
    event_count = alert.get('events_count', 0)
    if event_count > 100:
        score += 50
    elif event_count > 50:
        score += 30
    elif event_count > 20:
        score += 10

    # Scenario-based severity
    scenario = alert.get('scenario', '')
    if 'ssh' in scenario.lower():
        score += 20
    if 'sql-injection' in scenario.lower():
        score += 40
    if 'web-scan' in scenario.lower():
        score += 15

    # Duration of attack
    duration = alert.get('stop_at', 0) - alert.get('start_at', 0)
    if duration > 3600:  # Over 1 hour
        score += 25

    return score

def escalate_response(alert, severity_score):
    """Escalate response based on severity"""
    source_ip = alert['source']
    scenario = alert['scenario']

    if severity_score >= 80:
        # CRITICAL: Add to extended blocklist (30 days)
        print(f"[CRITICAL] {source_ip} - {scenario}")
        add_extended_ban(source_ip, duration="720h")  # 30 days
        notify_admin(
            f"CRITICAL ALERT: {scenario} from {source_ip}\n"
            f"Event count: {alert['events_count']}\n"
            f"Action: 30-day ban"
        )

    elif severity_score >= 50:
        # HIGH: Standard block + detailed logging
        print(f"[HIGH] {source_ip} - {scenario}")
        enable_detailed_logging(source_ip)
        notify_admin(
            f"HIGH ALERT: {scenario} from {source_ip}\n"
            f"Event count: {alert['events_count']}"
        )

    elif severity_score >= 30:
        # MEDIUM: Monitor and log
        print(f"[MEDIUM] {source_ip} - {scenario}")
        log_threat_pattern(alert)

    else:
        # LOW: Standard detection
        print(f"[LOW] {source_ip} - {scenario}")

def add_extended_ban(ip, duration="720h"):
    """Add IP to extended blocklist"""
    decision = {
        "ip": ip,
        "duration": duration,
        "type": "ban",
        "reason": "Extended ban - automated escalation"
    }

    response = requests.post(
        f"{CROWDSEC_URL}/v1/decisions",
        headers=headers,
        json=decision
    )

    return response.status_code == 201

def enable_detailed_logging(ip):
    """Enable enhanced logging for specific IP"""
    # Implement your logging enhancement logic
    print(f"Enhanced logging enabled for {ip}")

def log_threat_pattern(alert):
    """Log threat pattern for analysis"""
    with open("/var/log/crowdsec/threat-patterns.log", "a") as f:
        f.write(f"{datetime.utcnow().isoformat()} - {json.dumps(alert)}\n")

def notify_admin(message):
    """Send notification to admin (email, Slack, etc.)"""
    # Implement your notification logic
    print(f"NOTIFICATION: {message}")

def main():
    """Main analysis loop"""
    print("CrowdSec Intelligent Threat Analyzer starting...")

    while True:
        alerts = fetch_recent_alerts(minutes=5)

        for alert in alerts:
            severity = analyze_threat_severity(alert)
            escalate_response(alert, severity)

        # Wait 60 seconds before next check
        import time
        time.sleep(60)

if __name__ == "__main__":
    main()

Predictive Threat Intelligence

Use machine learning to predict attack patterns and proactively harden defenses:

"""
Predictive Threat Intelligence
Analyzes historical patterns to forecast attack waves
"""

import numpy as np
from sklearn.ensemble import IsolationForest
from datetime import datetime, timedelta

def predict_attack_wave(historical_data):
    """
    Predict when next major attack wave will occur

    Args:
        historical_data: List of alert counts per hour (30 days)

    Returns:
        dict: Prediction with time, confidence, and expected intensity
    """

    # Convert to numpy array
    alerts_per_hour = np.array(historical_data)

    # Calculate statistics
    mean_attacks = np.mean(alerts_per_hour)
    std_attacks = np.std(alerts_per_hour)

    # Detect cyclical patterns
    peaks = detect_peaks(alerts_per_hour, threshold=mean_attacks + std_attacks)

    if len(peaks) < 3:
        return {"confidence": 0, "message": "Insufficient data for prediction"}

    # Calculate average interval between peaks
    intervals = np.diff([p['time'] for p in peaks])
    avg_interval = np.mean(intervals)
    interval_std = np.std(intervals)

    # Predict next peak
    last_peak_time = peaks[-1]['time']
    next_peak_time = last_peak_time + avg_interval

    # Calculate confidence based on consistency
    confidence = 1.0 - (interval_std / avg_interval) if avg_interval > 0 else 0
    confidence = max(0, min(1, confidence))

    # Predict intensity
    peak_intensities = [p['intensity'] for p in peaks]
    expected_intensity = np.mean(peak_intensities) + (0.1 * np.std(peak_intensities))

    return {
        'predicted_time': next_peak_time,
        'confidence': confidence,
        'expected_intensity': expected_intensity,
        'recommendation': generate_recommendation(confidence, expected_intensity)
    }

def detect_peaks(data, threshold):
    """Detect peaks in time series data"""
    peaks = []
    for i in range(1, len(data) - 1):
        if data[i] > threshold and data[i] > data[i-1] and data[i] > data[i+1]:
            peaks.append({
                'time': i,
                'intensity': data[i]
            })
    return peaks

def generate_recommendation(confidence, intensity):
    """Generate actionable recommendations"""
    if confidence > 0.8 and intensity > 100:
        return "HIGH: Increase firewall rules, enable rate limiting"
    elif confidence > 0.6 and intensity > 50:
        return "MEDIUM: Monitor closely, prepare incident response"
    else:
        return "LOW: Continue standard monitoring"

# Example usage
if __name__ == "__main__":
    # Fetch 30 days of hourly alert counts
    historical_alerts = fetch_historical_alert_counts(days=30)

    prediction = predict_attack_wave(historical_alerts)

    if prediction['confidence'] > 0.7:
        print(f"Attack wave predicted:")
        print(f"  Time: {prediction['predicted_time']}")
        print(f"  Confidence: {prediction['confidence']:.2%}")
        print(f"  Expected intensity: {prediction['expected_intensity']:.0f} alerts/hour")
        print(f"  Recommendation: {prediction['recommendation']}")

        # Take proactive action
        if prediction['confidence'] > 0.85:
            harden_firewall()
            increase_monitoring_frequency()
            alert_security_team(prediction)

Anomaly Detection for Zero-Day Attacks

Detect novel attack patterns that don’t match existing scenarios:

"""
ML-Based Anomaly Detection
Identifies zero-day attacks using Isolation Forest
"""

from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import numpy as np

def extract_features(alert):
    """
    Extract feature vector from alert

    Features:
        - Request rate
        - Response time distribution
        - Character entropy
        - Geographic anomaly score
        - Time-of-day pattern
        - User agent rarity
    """
    return [
        alert.get('request_rate', 0),
        alert.get('avg_response_time', 0),
        calculate_entropy(alert.get('payload', '')),
        geographic_anomaly_score(alert.get('source_ip')),
        time_pattern_score(alert.get('timestamp')),
        user_agent_rarity(alert.get('user_agent', ''))
    ]

def calculate_entropy(text):
    """Calculate Shannon entropy of text"""
    if not text:
        return 0

    prob = [text.count(c) / len(text) for c in set(text)]
    entropy = -sum([p * np.log2(p) for p in prob if p > 0])

    return entropy

def geographic_anomaly_score(ip):
    """Calculate if geographic location is anomalous"""
    # Implement GeoIP lookup and comparison to baseline
    # Return score 0-1 (0=normal, 1=highly anomalous)
    return 0.5  # Placeholder

def time_pattern_score(timestamp):
    """Calculate time-of-day anomaly score"""
    # Most attacks happen at night; score based on deviation
    hour = datetime.fromisoformat(timestamp).hour
    if 0 <= hour <= 6:
        return 0.8  # Nighttime = higher anomaly
    elif 9 <= hour <= 17:
        return 0.2  # Business hours = lower anomaly
    return 0.5

def user_agent_rarity(user_agent):
    """Calculate rarity of user agent string"""
    # Compare against known good user agents
    common_agents = ['Mozilla', 'Chrome', 'Safari', 'Firefox']
    if any(agent in user_agent for agent in common_agents):
        return 0.2  # Common agent = low rarity
    return 0.8  # Unusual agent = high rarity

def train_anomaly_detector(normal_traffic_samples):
    """
    Train Isolation Forest on normal traffic

    Args:
        normal_traffic_samples: List of feature vectors from normal traffic

    Returns:
        Trained IsolationForest model
    """

    # Normalize features
    scaler = StandardScaler()
    X_train = scaler.fit_transform(normal_traffic_samples)

    # Train Isolation Forest
    model = IsolationForest(
        contamination=0.01,  # Expect 1% anomalies
        random_state=42,
        n_estimators=100
    )
    model.fit(X_train)

    return model, scaler

def detect_anomalies(model, scaler, new_alerts):
    """
    Detect anomalies in new alerts

    Returns:
        List of anomalous alerts with scores
    """

    # Extract features
    features = [extract_features(alert) for alert in new_alerts]
    X = scaler.transform(features)

    # Predict anomalies (-1 = anomaly, 1 = normal)
    predictions = model.predict(X)
    scores = model.decision_function(X)

    anomalies = []
    for i, (pred, score) in enumerate(zip(predictions, scores)):
        if pred == -1:
            anomalies.append({
                'alert': new_alerts[i],
                'anomaly_score': score,
                'confidence': calculate_anomaly_confidence(score)
            })

    return anomalies

def calculate_anomaly_confidence(score):
    """Convert anomaly score to confidence percentage"""
    # Scores typically range from -0.5 to 0.5
    # More negative = more anomalous
    confidence = max(0, min(1, (-score + 0.5)))
    return confidence

# Example usage
if __name__ == "__main__":
    # Train on 7 days of normal traffic
    normal_traffic = fetch_normal_traffic_samples(days=7)
    model, scaler = train_anomaly_detector(normal_traffic)

    # Analyze new alerts
    new_alerts = fetch_recent_alerts(minutes=15)
    anomalies = detect_anomalies(model, scaler, new_alerts)

    for anomaly in anomalies:
        if anomaly['confidence'] &gt; 0.7:
            print(f"ZERO-DAY ALERT: {anomaly['alert']['source']}")
            print(f"  Confidence: {anomaly['confidence']:.2%}")
            print(f"  Anomaly score: {anomaly['anomaly_score']:.3f}")

            # Create custom alert
            create_custom_alert(
                scenario='custom/ml-anomaly-detection',
                source=anomaly['alert']['source'],
                severity='high',
                details=f"ML-detected anomaly (confidence: {anomaly['confidence']:.2%})"
            )

Bash Automation for Webhook-Like Polling

Implement event-driven automation with Bash:

#!/bin/bash
# CrowdSec Event Poller
# Polls for new alerts and triggers automated responses

CROWDSEC_URL="http://localhost:8080"
API_KEY="your-api-key"
POLL_INTERVAL=60  # seconds

# Notification function
notify() {
    local severity=$1
    local message=$2

    # Send to syslog
    logger -t crowdsec-automation "[$severity] $message"

    # Optional: Send email, Slack, Discord, etc.
    # echo "$message" | mail -s "CrowdSec Alert [$severity]" [email protected]
}

# Fetch recent alerts
fetch_alerts() {
    local since=$1
    curl -s -H "X-API-Key: $API_KEY"         "$CROWDSEC_URL/v1/alerts?has_active_decision=true"         | jq -r --arg since "$since" '.[] | select(.created_at > $since)'
}

# Process alert and take action
process_alert() {
    local alert=$1
    local ip=$(echo "$alert" | jq -r '.source')
    local scenario=$(echo "$alert" | jq -r '.scenario')
    local event_count=$(echo "$alert" | jq -r '.events_count')

    # Scenario-specific actions
    if [[ "$scenario" == *"ssh"* ]]; then
        # SSH attack: Add firewall rule
        notify "HIGH" "SSH attack from $ip ($event_count events)"

        # Add iptables rule if not in Docker
        # iptables -A INPUT -s $ip -j DROP

        # Log to security log
        echo "$(date -u +%Y-%m-%dT%H:%M:%SZ) | SSH_ATTACK | $ip | $event_count"             >> /var/log/crowdsec/security-events.log

    elif [[ "$scenario" == *"sql-injection"* ]]; then
        # SQL injection: Critical alert
        notify "CRITICAL" "SQL injection attempt from $ip"

        # Check for data exfiltration
        check_database_logs "$ip"

        # Create incident report
        create_incident_report "$ip" "$scenario" "$event_count"

    elif [[ "$scenario" == *"web-scan"* ]]; then
        # Web scanning: Monitor closely
        notify "MEDIUM" "Web scan detected from $ip"

        # Enable verbose logging for this IP
        # (implementation depends on your web server)

    elif [[ "$scenario" == *"dos"* ]]; then
        # DoS attack: Rate limiting
        notify "HIGH" "DoS attack from $ip"

        # Enable rate limiting
        enable_rate_limiting "$ip"
    fi
}

# Helper: Check database logs for suspicious activity
check_database_logs() {
    local ip=$1
    # Implementation depends on your database setup
    notify "INFO" "Checking database logs for $ip"
}

# Helper: Create incident report
create_incident_report() {
    local ip=$1
    local scenario=$2
    local event_count=$3

    local report_file="/var/log/crowdsec/incident-$(date +%Y%m%d-%H%M%S).txt"

    cat > "$report_file" <<EOF
SECURITY INCIDENT REPORT
========================
Date: $(date -u)
Source IP: $ip
Scenario: $scenario
Event Count: $event_count

CrowdSec Decision:
$(curl -s -H "X-API-Key: $API_KEY" "$CROWDSEC_URL/v1/decisions?ip=$ip" | jq)

Recommendations:
- Review logs for data exfiltration
- Check for compromised accounts
- Verify database integrity
- Report to security team
EOF

    notify "INFO" "Incident report created: $report_file"
}

# Helper: Enable rate limiting
enable_rate_limiting() {
    local ip=$1
    notify "INFO" "Enabling rate limiting for $ip"
    # Implementation depends on your setup (nginx, haproxy, etc.)
}

# Main loop
main() {
    echo "CrowdSec Event Poller starting..."
    local last_check=$(date -u -d '1 minute ago' +%Y-%m-%dT%H:%M:%SZ)

    while true; do
        local current_time=$(date -u +%Y-%m-%dT%H:%M:%SZ)

        # Fetch alerts since last check
        local alerts=$(fetch_alerts "$last_check")

        # Process each alert
        if [[ -n "$alerts" ]]; then
            echo "$alerts" | jq -c '.' | while read -r alert; do
                process_alert "$alert"
            done
        fi

        last_check=$current_time
        sleep $POLL_INTERVAL
    done
}

# Run main loop
main

AI Opportunities Summary

CrowdSec creates numerous opportunities for AI enhancement:

1. Intelligent Threat Scoring: Replace binary ban decisions with ML-based confidence scores
2. Adaptive Response Actions: Escalate responses based on attack severity and context
3. Predictive Threat Intelligence: Forecast attack waves using time-series analysis
4. Anomaly Detection: Identify zero-day attacks that don’t match existing scenarios
5. Natural Language Summaries: Generate plain-English incident reports with recommendations
6. Automated Incident Response: Autonomous workflows for common attack patterns
7. Personalized Security Profiles: ML-driven optimization of thresholds based on your environment

---

Troubleshooting Common Issues

CrowdSec Not Detecting Logs

Symptoms: No alerts generated, metrics show zero events

Diagnostic steps:

# 1. Verify log file access
docker-compose exec crowdsec ls -la /var/log/nginx/
docker-compose exec crowdsec tail /var/log/nginx/access.log

# 2. Check acquis configuration
docker-compose exec crowdsec cat /etc/crowdsec/acquis.yaml

# 3. Verify parsers are loaded
docker-compose exec crowdsec cscli parsers list | grep nginx

# 4. Check for errors
docker-compose logs crowdsec | grep -i error

# 5. Enable debug logging
docker-compose exec crowdsec sh -c 'cat >> /etc/crowdsec/config.yaml <<EOF
log_level: debug
EOF'
docker-compose restart crowdsec

Common solutions:

• File paths don’t match mounted volumes → Fix docker-compose volumes
• Missing collection → Install with cscli collections install
• Wrong log format → Install parser for specific service
• Permission denied → Mount log directory with :ro flag

Bouncers Not Blocking Traffic

Symptoms: CrowdSec detects alerts but traffic isn’t blocked

Diagnostic steps:

# 1. Verify bouncer is running
docker-compose logs firewall-bouncer

# 2. Check bouncer token validity
docker-compose exec crowdsec cscli bouncers list

# 3. Confirm Local API reachability
docker-compose exec firewall-bouncer curl http://crowdsec:8080/health

# 4. Verify network connectivity
docker network inspect crowdsec_network

# 5. Check iptables rules (if using firewall bouncer)
docker-compose exec firewall-bouncer iptables -L -n | grep crowdsec

Common solutions:

• Invalid API key → Regenerate with cscli bouncers renew
• Network disconnection → Check docker-compose networks
• Missing NET_ADMIN capability → Add to docker-compose
• Bouncer not querying LAPI → Check environment variables

High CPU/Memory Usage

Optimization strategies:

# Add resource limits to docker-compose.yml
services:
  crowdsec:
    deploy:
      resources:
        limits:
          cpus: '1'
          memory: 512M
        reservations:
          cpus: '0.5'
          memory: 256M

# Reduce log retention
docker-compose exec crowdsec sh -c 'cat >> /etc/crowdsec/config.yaml <<EOF
db_config:
  flush_every: 10s
  max_records: 10000
EOF'

# Use log sampling for high-volume logs
docker-compose exec crowdsec sh -c 'cat > /etc/crowdsec/acquis.d/nginx.yaml <<EOF
source: file
filename: /var/log/nginx/access.log
sample_rate: 10  # Process every 10th line
labels:
  type: nginx
EOF'

docker-compose restart crowdsec

False Positives

Whitelist legitimate traffic:

# Temporarily remove decision
docker-compose exec crowdsec cscli decisions delete -i 192.168.1.100

# Permanently whitelist with postoverflow
docker-compose exec crowdsec sh -c 'cat > /etc/crowdsec/postoverflows/whitelist.yaml <<EOF
name: postoverflow/whitelist-trusted
filter: "evt.Parsed.source_ip in ['192.168.1.0/24', '10.0.0.0/8']"
EOF'

docker-compose restart crowdsec

---

Best Practices and Recommendations

Security

1. Store API keys securely: Use .env files (never commit to git) or Docker secrets
2. Rotate tokens periodically: Renew bouncer tokens quarterly
3. Enable SSL/TLS for remote LAPI: Use certificates for production deployments
4. Monitor bouncer activity: Regularly check cscli bouncers list for anomalies
5. Implement whitelisting: Protect legitimate IPs (office, VPN, admin locations)

Performance

1. Use log sampling for high-volume logs (>10k requests/hour)
2. Limit collection scope: Only install collections for services you’re running
3. Set resource limits in Docker Compose
4. Regular database maintenance: Flush old records periodically
5. Monitor metrics: Track parser performance and scenario efficiency

Operational

1. Test scenarios in simulation mode before enabling enforcement
2. Review alerts regularly: Check for false positives and adjust thresholds
3. Keep hub updated: Run cscli hub update weekly
4. Backup configurations: Version control your custom parsers and scenarios
5. Document customizations: Maintain notes on why custom scenarios were created

Integration

1. Layer defenses: Use CrowdSec with Cloudflare for multi-level protection
2. Centralize logs: Send CrowdSec alerts to SIEM for correlation
3. Automate responses: Use Python/Bash scripts for intelligent escalation
4. Monitor from multiple angles: Console + Grafana + CLI
5. Community participation: Enroll in CAPI to benefit from and contribute to threat intelligence

---

Next Steps and Related Guides

Advanced Topics

Multi-Instance Deployments: Scale CrowdSec across multiple servers with centralized LAPI

Custom Scenario Development: Build sophisticated multi-stage detection scenarios

Cloudflare Integration: Sync CrowdSec decisions to Cloudflare firewall rules for CDN-level protection

SIEM Integration: Export alerts to Splunk, ELK, or Wazuh for centralized analysis

Machine Learning Enhancement: Build predictive models for threat forecasting

Related Homelab Security Tools

• UFW: Host-level firewall management
• Fail2ban: Lightweight alternative for resource-constrained systems
• Traefik: Modern reverse proxy with native Docker integration
• Nginx Proxy Manager: User-friendly reverse proxy management
• Suricata: Network-level IDS/IPS
• Wazuh: SIEM for centralized security monitoring

Community Resources

• Official Documentation: https://docs.crowdsec.net/
• Community Discourse: https://discourse.crowdsec.net/
• Hub Repository: https://hub.crowdsec.net/ (scenarios, parsers, bouncers)
• GitHub: https://github.com/crowdsecurity/crowdsec
• Blog: https://www.crowdsec.net/blog

---

Conclusion

CrowdSec represents a paradigm shift in homelab security—from reactive pattern matching to proactive, community-backed threat intelligence. Its modular architecture allows you to start simple (basic SSH protection) and scale to sophisticated multi-bouncer deployments protecting your entire infrastructure.

The integration of REST APIs and structured data format creates unlimited opportunities for AI-driven automation: predictive threat intelligence, adaptive responses, anomaly detection, and intelligent incident reporting. As machine learning models mature, CrowdSec can evolve from a detection tool to an intelligent, autonomous security layer.

By deploying CrowdSec, you’re not just protecting your homelab—you’re joining a global community of security-conscious users who share threat intelligence and collectively strengthen defenses against evolving threats.

Start your CrowdSec journey today: Deploy the basic stack, monitor for a week, then gradually add custom scenarios and automation as you learn the patterns unique to your infrastructure.

Happy homelabbing, and stay secure!