CrowdSec คืออะไรและทำงานอย่างไร
CrowdSec เป็น open source Intrusion Prevention System (IPS) แบบ collaborative ที่ใช้ crowd-sourced intelligence ในการตรวจจับและบล็อก threats แตกต่างจาก traditional IPS ตรงที่ CrowdSec แชร์ threat intelligence ระหว่างผู้ใช้ทั่วโลก เมื่อ IP ถูกตรวจจับว่า malicious จากผู้ใช้คนหนึ่ง ข้อมูลจะถูกแชร์ไปยังทุกู้คืนใน community
สถาปัตยกรรมของ CrowdSec ประกอบด้วย Agent (Security Engine) ที่วิเคราะห์ logs และตรวจจับ behaviors ผิดปกติ, Bouncer ที่ enforce decisions (block/captcha IPs), Local API ที่จัดการ decisions และสื่อสารกับ components และ Central API (CrowdSec Cloud) ที่แชร์ threat intelligence กับ community
CrowdSec ใช้ behavior-based detection แทน signature-based ทำให้ตรวจจับ zero-day attacks ได้ดีกว่า Scenarios กำหนด patterns ที่ถือว่า malicious เช่น brute force (login failures มากเกิน), port scanning, web application attacks (SQL injection, XSS), DDoS patterns และ credential stuffing
เมื่อเทียบกับ Fail2ban CrowdSec มี performance ดีกว่า (เขียนด้วย Go), มี community threat intelligence, รองรับ multi-server architecture, มี web console สำหรับ management และ ecosystem ของ bouncers สำหรับ services ต่างๆ
ติดตั้ง CrowdSec IPS และตั้งค่าเบื้องต้น
ติดตั้งและ configure CrowdSec
# === ติดตั้ง CrowdSec ===
# Debian/Ubuntu
curl -s https://packagecloud.io/install/repositories/crowdsec/crowdsec/script.deb.sh | sudo bash
sudo apt install crowdsec
# CentOS/RHEL
curl -s https://packagecloud.io/install/repositories/crowdsec/crowdsec/script.rpm.sh | sudo bash
sudo yum install crowdsec
# Docker
# docker run -d --name crowdsec \
# -v /var/log:/var/log:ro \
# -v crowdsec-config:/etc/crowdsec \
# -v crowdsec-data:/var/lib/crowdsec/data \
# -p 8080:8080 \
# crowdsecurity/crowdsec
# === ตรวจสอบการติดตั้ง ===
sudo cscli version
sudo cscli metrics
sudo cscli hub list
# === ติดตั้ง Collections (detection rules) ===
# SSH brute force detection
sudo cscli collections install crowdsecurity/sshd
# Nginx detection
sudo cscli collections install crowdsecurity/nginx
# Apache detection
sudo cscli collections install crowdsecurity/apache2
# Linux base
sudo cscli collections install crowdsecurity/linux
# WordPress protection
sudo cscli collections install crowdsecurity/wordpress
# HTTP attack detection
sudo cscli collections install crowdsecurity/http-cve
# ดู collections ที่ติดตั้ง
sudo cscli collections list
# === ติดตั้ง Bouncer (enforcement) ===
# Firewall bouncer (iptables/nftables)
sudo apt install crowdsec-firewall-bouncer-iptables
# Nginx bouncer
sudo apt install crowdsec-nginx-bouncer
# Register bouncer
sudo cscli bouncers add firewall-bouncer
# เก็บ API key ที่ได้
# === Configuration ===
# /etc/crowdsec/acquis.yaml (log sources)
# filenames:
# - /var/log/nginx/access.log
# - /var/log/nginx/error.log
# labels:
# type: nginx
# ---
# filenames:
# - /var/log/auth.log
# labels:
# type: syslog
# ---
# filenames:
# - /var/log/apache2/access.log
# labels:
# type: apache2
# Restart
sudo systemctl restart crowdsec
sudo systemctl status crowdsec
# === ทดสอบ ===
# ดู decisions (blocked IPs)
sudo cscli decisions list
# ดู alerts
sudo cscli alerts list
# Ban IP manually
sudo cscli decisions add --ip 1.2.3.4 --reason "manual ban" --duration 24h
# Unban IP
sudo cscli decisions delete --ip 1.2.3.4
# ดู metrics
sudo cscli metrics showสร้าง Custom Scenarios และ Parsers
เขียน custom detection rules
# === Custom Parser ===
# /etc/crowdsec/parsers/s01-parse/custom-app.yaml
# name: custom/my-app-parser
# description: "Parse custom application logs"
# filter: "evt.Parsed.program == 'my-app'"
# onsuccess: next_stage
# nodes:
# - grok:
# pattern: '%{IP:source_ip} - %{DATA:user} \[%{HTTPDATE:timestamp}\] "%{WORD:method} %{URIPATHPARAM:uri}" %{NUMBER:status} %{NUMBER:bytes}'
# expression: "evt.Parsed.message"
# - statics:
# - meta: source_ip
# expression: "evt.Parsed.source_ip"
# - meta: http_status
# expression: "evt.Parsed.status"
# - meta: http_path
# expression: "evt.Parsed.uri"
# === Custom Scenario: API Rate Limiting ===
# /etc/crowdsec/scenarios/custom-api-ratelimit.yaml
# type: leaky
# name: custom/api-rate-limit
# description: "Detect API abuse (too many requests)"
# filter: "evt.Meta.source_ip != '' && evt.Meta.http_path startsWith '/api/'"
# groupby: "evt.Meta.source_ip"
# capacity: 100
# leakspeed: "1s"
# blackhole: 5m
# labels:
# type: api_abuse
# remediation: true
# === Custom Scenario: Login Brute Force ===
# /etc/crowdsec/scenarios/custom-login-bruteforce.yaml
# type: leaky
# name: custom/login-bruteforce
# description: "Detect login brute force attempts"
# filter: "evt.Meta.http_path == '/login' && evt.Meta.http_status == '401'"
# groupby: "evt.Meta.source_ip"
# capacity: 5
# leakspeed: "10s"
# blackhole: 10m
# labels:
# type: brute_force
# remediation: true
# === Custom Scenario: Path Traversal ===
# /etc/crowdsec/scenarios/custom-path-traversal.yaml
# type: trigger
# name: custom/path-traversal
# description: "Detect path traversal attempts"
# filter: "evt.Meta.http_path contains '../' || evt.Meta.http_path contains '..\\'"
# groupby: "evt.Meta.source_ip"
# blackhole: 30m
# labels:
# type: path_traversal
# remediation: true
# === ทดสอบ Parser/Scenario ===
# Validate configuration
sudo cscli parsers validate /etc/crowdsec/parsers/s01-parse/custom-app.yaml
sudo cscli scenarios validate /etc/crowdsec/scenarios/custom-api-ratelimit.yaml
# Test with log file
sudo cscli explain --file /var/log/nginx/access.log --type nginx
# Simulate attack
sudo cscli simulation enable custom/api-rate-limit
# Reload after changes
sudo systemctl reload crowdsec
# Monitor in real-time
sudo cscli alerts list --since 5m
sudo journalctl -u crowdsec -fCrowdSec API และ Automation ด้วย Python
ใช้ CrowdSec API สำหรับ automation
#!/usr/bin/env python3
# crowdsec_automation.py — CrowdSec API Automation
import requests
import json
import logging
from datetime import datetime, timedelta
from typing import Dict, List, Optional
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("crowdsec_api")
class CrowdSecAPI:
def __init__(self, api_url="http://localhost:8080", api_key=""):
self.base_url = api_url
self.headers = {
"X-Api-Key": api_key,
"Content-Type": "application/json",
}
def get_decisions(self, ip=None, scope=None, limit=100):
params = {"limit": limit}
if ip:
params["ip"] = ip
if scope:
params["scope"] = scope
resp = requests.get(
f"{self.base_url}/v1/decisions",
headers=self.headers,
params=params,
)
resp.raise_for_status()
return resp.json() or []
def add_decision(self, ip, duration="24h", reason="api_ban", scenario="manual"):
payload = {
"duration": duration,
"reason": reason,
"origin": "cscli",
"scenario": scenario,
"scope": "ip",
"type": "ban",
"value": ip,
}
resp = requests.post(
f"{self.base_url}/v1/decisions",
headers=self.headers,
json=payload,
)
resp.raise_for_status()
logger.info(f"Banned {ip} for {duration}: {reason}")
return resp.json()
def delete_decision(self, ip):
resp = requests.delete(
f"{self.base_url}/v1/decisions",
headers=self.headers,
params={"ip": ip},
)
resp.raise_for_status()
logger.info(f"Unbanned {ip}")
return resp.json()
def get_alerts(self, since=None, limit=50):
params = {"limit": limit}
if since:
params["since"] = since
resp = requests.get(
f"{self.base_url}/v1/alerts",
headers=self.headers,
params=params,
)
resp.raise_for_status()
return resp.json() or []
def get_metrics(self):
resp = requests.get(
f"{self.base_url}/v1/metrics",
headers=self.headers,
)
resp.raise_for_status()
return resp.json()
class CrowdSecMonitor:
def __init__(self, api_client: CrowdSecAPI):
self.api = api_client
def generate_threat_report(self, hours=24):
alerts = self.api.get_alerts(since=f"{hours}h")
decisions = self.api.get_decisions(limit=500)
# Analyze alerts
attack_types = {}
source_ips = {}
for alert in alerts:
scenario = alert.get("scenario", "unknown")
attack_types[scenario] = attack_types.get(scenario, 0) + 1
source = alert.get("source", {})
ip = source.get("ip", "unknown")
source_ips[ip] = source_ips.get(ip, 0) + 1
# Top attackers
top_attackers = sorted(source_ips.items(), key=lambda x: -x[1])[:10]
top_attacks = sorted(attack_types.items(), key=lambda x: -x[1])[:10]
report = {
"period_hours": hours,
"generated_at": datetime.utcnow().isoformat(),
"summary": {
"total_alerts": len(alerts),
"active_bans": len(decisions),
"unique_attackers": len(source_ips),
"attack_types": len(attack_types),
},
"top_attack_types": dict(top_attacks),
"top_attackers": dict(top_attackers),
}
return report
def auto_whitelist(self, trusted_ips):
decisions = self.api.get_decisions(limit=1000)
unblocked = 0
for decision in decisions:
if decision.get("value") in trusted_ips:
self.api.delete_decision(decision["value"])
logger.info(f"Auto-whitelisted trusted IP: {decision['value']}")
unblocked += 1
return unblocked
def threat_intel_export(self, output_file="threat_intel.json"):
decisions = self.api.get_decisions(limit=10000)
intel = {
"format": "crowdsec",
"exported_at": datetime.utcnow().isoformat(),
"indicators": [],
}
for d in decisions:
intel["indicators"].append({
"type": "ip",
"value": d.get("value"),
"threat_type": d.get("scenario", "unknown"),
"confidence": 80,
"first_seen": d.get("created_at"),
"duration": d.get("duration"),
})
with open(output_file, "w") as f:
json.dump(intel, f, indent=2)
logger.info(f"Exported {len(intel['indicators'])} indicators to {output_file}")
return len(intel["indicators"])
# api = CrowdSecAPI(api_key="your-api-key")
# monitor = CrowdSecMonitor(api)
# report = monitor.generate_threat_report(hours=24)
# print(json.dumps(report, indent=2))Multi-Server Deployment และ Console
Deploy CrowdSec across multiple servers
# === Multi-Server Architecture ===
#
# ┌─────────────────────────────────────┐
# │ CrowdSec Console │
# │ (console.crowdsec.net or │
# │ self-hosted) │
# └──────────────┬──────────────────────┘
# │
# ┌──────────┼──────────┐
# │ │ │
# ┌───▼───┐ ┌───▼───┐ ┌───▼───┐
# │Server1│ │Server2│ │Server3│
# │ Agent │ │ Agent │ │ Agent │
# │Bouncer│ │Bouncer│ │Bouncer│
# └───────┘ └───────┘ └───────┘
#
# Shared decisions across all servers
# Central management via Console
# === Ansible Playbook สำหรับ Multi-Server Deployment ===
# deploy_crowdsec.yml
# - hosts: all_servers
# become: true
# vars:
# crowdsec_console_token: "{{ vault_console_token }}"
# collections:
# - crowdsecurity/linux
# - crowdsecurity/sshd
# - crowdsecurity/nginx
#
# tasks:
# - name: Install CrowdSec repository
# shell: curl -s https://packagecloud.io/install/repositories/crowdsec/crowdsec/script.deb.sh | bash
#
# - name: Install CrowdSec
# apt:
# name:
# - crowdsec
# - crowdsec-firewall-bouncer-iptables
# state: present
#
# - name: Install collections
# command: "cscli collections install {{ item }}"
# loop: "{{ collections }}"
#
# - name: Enroll to Console
# command: "cscli console enroll {{ crowdsec_console_token }}"
# when: crowdsec_console_token != ""
#
# - name: Configure acquisition
# template:
# src: acquis.yaml.j2
# dest: /etc/crowdsec/acquis.yaml
# notify: restart crowdsec
#
# - name: Enable and start CrowdSec
# systemd:
# name: crowdsec
# enabled: true
# state: started
#
# handlers:
# - name: restart crowdsec
# systemd:
# name: crowdsec
# state: restarted
# === Docker Compose Multi-Service ===
# docker-compose.yml
# services:
# crowdsec:
# image: crowdsecurity/crowdsec:latest
# restart: always
# ports: ["8080:8080"]
# volumes:
# - /var/log/nginx:/var/log/nginx:ro
# - /var/log/auth.log:/var/log/auth.log:ro
# - crowdsec-config:/etc/crowdsec
# - crowdsec-data:/var/lib/crowdsec/data
# environment:
# COLLECTIONS: "crowdsecurity/nginx crowdsecurity/sshd crowdsecurity/linux"
# ENROLL_KEY: ""
# ENROLL_INSTANCE_NAME: "web-server-1"
#
# bouncer-nginx:
# image: crowdsecurity/nginx-bouncer:latest
# restart: always
# environment:
# CROWDSEC_BOUNCER_API_KEY: ""
# CROWDSEC_AGENT_HOST: "crowdsec:8080"
# network_mode: "host"
#
# volumes:
# crowdsec-config:
# crowdsec-data:
# === Prometheus Monitoring ===
# CrowdSec exposes metrics at /metrics
# prometheus.yml:
# scrape_configs:
# - job_name: crowdsec
# static_configs:
# - targets:
# - server1:6060
# - server2:6060
# - server3:6060
# metrics_path: /metrics
# === Grafana Dashboard Queries ===
# Active decisions: cs_active_decisions
# Alerts per scenario: cs_alerts_total
# Parser lines processed: cs_parser_hits_total
# Bouncer requests: cs_bouncer_requests_total
# === Console Commands ===
# Enroll to CrowdSec Console
sudo cscli console enroll YOUR_ENROLLMENT_KEY
# List enrolled machines
sudo cscli machines list
# Share blocklist to community
sudo cscli console enable sharing
# ตรวจสอบ status
sudo cscli console statusแนวโน้ม IPS Technology สำหรับ 2026
เทคโนโลยี IPS ที่จะมาในอนาคต
# === IPS Technology Trends 2026 ===
#
# 1. AI/ML-Powered Detection
# - Deep learning สำหรับ anomaly detection
# - NLP สำหรับวิเคราะห์ log patterns
# - Reinforcement learning สำหรับ adaptive responses
# - Reduce false positives ด้วย contextual analysis
#
# 2. Zero Trust Integration
# - IPS ทำงานร่วมกับ Zero Trust architecture
# - Per-request authentication and authorization
# - Microsegmentation enforcement
# - Continuous verification แทน perimeter-based
#
# 3. Cloud-Native IPS
# - Kubernetes-native security
# - Service mesh integration (Istio, Linkerd)
# - Serverless function protection
# - Container runtime security
#
# 4. Collaborative Intelligence
# - CrowdSec-style community threat sharing
# - Real-time global threat intelligence
# - Industry-specific threat feeds
# - Privacy-preserving data sharing
#
# 5. Edge Computing Security
# - IPS at edge locations (CDN, IoT gateways)
# - Low-latency detection and response
# - Distributed decision making
# - 5G network security integration
#
# 6. Automated Response (SOAR Integration)
# - Playbook-based automated remediation
# - Self-healing infrastructure
# - Incident response automation
# - Compliance-aware response actions
#
# === Conference Topics for 2026 ===
#
# Track 1: Advanced Threat Detection
# - "Beyond Signatures: ML-based Zero-Day Detection"
# - "Behavioral Analysis at Scale with CrowdSec 3.0"
# - "Graph Neural Networks for Attack Path Detection"
#
# Track 2: Cloud & Container Security
# - "Kubernetes IPS: From Admission to Runtime"
# - "eBPF-based Network Security in Production"
# - "Service Mesh Security Patterns"
#
# Track 3: Collaborative Defense
# - "Building Open Threat Intelligence Networks"
# - "Privacy-Preserving Threat Sharing with MPC"
# - "Community-Driven Security: Lessons from CrowdSec"
#
# Track 4: Compliance & Automation
# - "Automated Compliance with Security-as-Code"
# - "SOAR Playbooks for IPS Response"
# - "Continuous Security Validation in CI/CD"
#
# === Hands-on Labs ===
# Lab 1: Deploy CrowdSec on Kubernetes
# Lab 2: Build Custom ML Detection Scenarios
# Lab 3: Create Threat Intelligence Pipeline
# Lab 4: Automate Incident Response with SOARFAQ คำถามที่พบบ่อย
Q: CrowdSec กับ Fail2ban ต่างกันอย่างไร?
A: CrowdSec เขียนด้วย Go ทำให้ parse logs ได้เร็วกว่า Fail2ban (Python) มาก มี community threat intelligence ที่แชร์ blocklists ทั่วโลก รองรับ multi-server architecture ด้วย Console มี ecosystem ของ bouncers สำหรับ services ต่างๆ Fail2ban เป็น single-server tool ไม่มี threat sharing ต้อง configure ทีละเครื่อง แต่ simple กว่าและมีมานานกว่า สำหรับ production แนะนำ CrowdSec
Q: CrowdSec ฟรีจริงไหม?
A: CrowdSec Engine (agent + bouncer) เป็น open source ฟรี 100% ใช้ได้ไม่จำกัด Community blocklist ฟรี Console มี free tier สำหรับ 2 engines สำหรับองค์กรที่ต้องการ premium blocklists, dedicated support, SLA และ console สำหรับ engines มากกว่า 2 ตัว มี paid plans เริ่มต้นที่ $25/month
Q: CrowdSec มีผลกระทบกับ performance ของ server ไหม?
A: CrowdSec ใช้ resources น้อยมาก ปกติใช้ CPU < 1% และ RAM 50-100MB เขียนด้วย Go ทำให้ efficient สูง Bouncer ทำงานที่ firewall level ไม่มี overhead สำหรับ legitimate traffic มีผลกระทบเฉพาะกับ malicious IPs ที่ถูก ban สำหรับ server ที่มี traffic สูงมาก (>10,000 req/s) ควร monitor resources และปรับ configuration ตามความเหมาะสม
Q: จะป้องกัน false positives อย่างไร?
A: ใช้ simulation mode ก่อน deploy จริง (cscli simulation enable) เพื่อดูว่า scenario จะ block ใครบ้าง whitelist trusted IPs ใน /etc/crowdsec/parsers/s02-enrich/whitelist.yaml ปรับ capacity และ leakspeed ของ scenarios ให้เหมาะกับ traffic patterns ใช้ captcha แทน ban สำหรับ borderline cases ตรวจสอบ decisions เป็นประจำและ unban false positives ทันที
