Unify xdp-blocker and xdp-ddos into single xdp-defense project

Chain two XDP programs via libxdp dispatcher on the same interface:
xdp_blocker (priority 10) handles CIDR/country/whitelist blocking,
xdp_ddos (priority 20) handles rate limiting, EWMA analysis, and AI
anomaly detection. Whitelist maps are shared via BPF map pinning so
whitelisted IPs bypass both blocklist checks and DDoS rate limiting.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
kaffa
2026-02-07 08:39:21 +09:00
commit 1bcaddce25
12 changed files with 3523 additions and 0 deletions

8
.gitignore vendored Normal file
View File

@@ -0,0 +1,8 @@
*.o
*.ko
*.cmd
.tmp_versions/
modules.order
Module.symvers
__pycache__/
*.pyc

102
Makefile Normal file
View File

@@ -0,0 +1,102 @@
# XDP Defense - Unified XDP Blocker + DDoS Defense
# Build, install, and manage the integrated XDP defense system
PROJ_DIR := /opt/xdp-defense
BPF_DIR := $(PROJ_DIR)/bpf
LIB_DIR := $(PROJ_DIR)/lib
BIN_DIR := $(PROJ_DIR)/bin
CFG_DIR := $(PROJ_DIR)/config
INSTALL_BIN := /usr/local/bin
SYSTEMD_DIR := /etc/systemd/system
ETC_DIR := /etc/xdp-defense
DATA_DIR := /var/lib/xdp-defense
BLOCKER_CFG := /etc/xdp-blocker
CLANG ?= clang
CLANG_FLAGS := -O2 -g -Wall -target bpf \
-I/usr/include -I/usr/include/bpf -I/usr/include/xdp
BPF_OBJECTS := $(BPF_DIR)/xdp_blocker.o $(BPF_DIR)/xdp_ddos.o
.PHONY: all build install uninstall enable disable clean check-deps status
all: build
build: $(BPF_OBJECTS)
$(BPF_DIR)/xdp_blocker.o: $(BPF_DIR)/xdp_blocker.c
$(CLANG) $(CLANG_FLAGS) -c $< -o $@
$(BPF_DIR)/xdp_ddos.o: $(BPF_DIR)/xdp_ddos.c
$(CLANG) $(CLANG_FLAGS) -c $< -o $@
install: build
@echo "Installing XDP Defense..."
# Directories
install -d $(ETC_DIR)
install -d $(DATA_DIR)
install -d $(BLOCKER_CFG)/countries
install -d $(BLOCKER_CFG)/whitelist
# CLI
install -m 755 $(BIN_DIR)/xdp-defense $(INSTALL_BIN)/xdp-defense
ln -sf $(INSTALL_BIN)/xdp-defense $(INSTALL_BIN)/xdp-block
# Startup script
install -m 755 $(BIN_DIR)/xdp-startup.sh $(INSTALL_BIN)/xdp-startup
# Python libraries
install -m 644 $(LIB_DIR)/xdp_common.py $(INSTALL_BIN)/xdp_common.py
install -m 755 $(LIB_DIR)/xdp_country.py $(INSTALL_BIN)/xdp-country
install -m 755 $(LIB_DIR)/xdp_whitelist.py $(INSTALL_BIN)/xdp-whitelist
install -m 755 $(LIB_DIR)/xdp_defense_daemon.py $(INSTALL_BIN)/xdp-defense-daemon
# Config (don't overwrite existing)
test -f $(ETC_DIR)/config.yaml || install -m 644 $(CFG_DIR)/config.yaml $(ETC_DIR)/config.yaml
# Systemd service
install -m 644 $(CFG_DIR)/xdp-defense.service $(SYSTEMD_DIR)/xdp-defense.service
systemctl daemon-reload
@echo ""
@echo "Installed successfully."
@echo " CLI: $(INSTALL_BIN)/xdp-defense"
@echo " Compat: $(INSTALL_BIN)/xdp-block -> xdp-defense"
@echo " Config: $(ETC_DIR)/config.yaml"
@echo ""
@echo "Run 'make enable' to enable on boot."
uninstall:
@echo "Uninstalling XDP Defense..."
systemctl stop xdp-defense 2>/dev/null || true
systemctl disable xdp-defense 2>/dev/null || true
rm -f $(INSTALL_BIN)/xdp-defense
rm -f $(INSTALL_BIN)/xdp-block
rm -f $(INSTALL_BIN)/xdp-startup
rm -f $(INSTALL_BIN)/xdp_common.py
rm -f $(INSTALL_BIN)/xdp-country
rm -f $(INSTALL_BIN)/xdp-whitelist
rm -f $(INSTALL_BIN)/xdp-defense-daemon
rm -f $(SYSTEMD_DIR)/xdp-defense.service
systemctl daemon-reload
@echo "Uninstalled. Config preserved in $(ETC_DIR) and $(BLOCKER_CFG)"
enable:
systemctl enable xdp-defense
@echo "XDP Defense will start on boot"
disable:
systemctl disable xdp-defense
@echo "XDP Defense will not start on boot"
status:
@systemctl status xdp-defense 2>/dev/null || echo "Service not installed"
clean:
rm -f $(BPF_DIR)/*.o
check-deps:
@echo "Checking dependencies..."
@which clang >/dev/null 2>&1 || (echo "ERROR: clang not found" && exit 1)
@which bpftool >/dev/null 2>&1 || (echo "ERROR: bpftool not found" && exit 1)
@which ip >/dev/null 2>&1 || (echo "ERROR: iproute2 not found" && exit 1)
@which python3 >/dev/null 2>&1 || (echo "ERROR: python3 not found" && exit 1)
@which xdp-loader >/dev/null 2>&1 || (echo "ERROR: xdp-loader not found (libxdp)" && exit 1)
@python3 -c "import yaml" 2>/dev/null || (echo "ERROR: python3-yaml not found" && exit 1)
@test -f /usr/include/xdp/xdp_helpers.h || (echo "ERROR: xdp_helpers.h not found" && exit 1)
@echo "All dependencies satisfied"

971
bin/xdp-defense Executable file
View File

@@ -0,0 +1,971 @@
#!/bin/bash
# XDP Defense - Unified CLI for XDP Blocker + DDoS Defense
# Combines CIDR/country/whitelist blocking with rate limiting + AI detection
# Uses libxdp dispatcher for chaining two XDP programs
set -e
PROJ_DIR="/opt/xdp-defense"
BPF_DIR="$PROJ_DIR/bpf"
LIB_DIR="$PROJ_DIR/lib"
PIN_PATH="/sys/fs/bpf/xdp-defense"
CONFIG_FILE="/etc/xdp-defense/config.yaml"
DATA_DIR="/var/lib/xdp-defense"
PID_FILE="$DATA_DIR/daemon.pid"
BLOCKLIST_FILE="/etc/xdp-blocker/blocklist.txt"
COUNTRY_DIR="/etc/xdp-blocker/countries"
GEOIP_DB="/usr/share/GeoIP/GeoLite2-Country.mmdb"
CITY_DB="/usr/share/GeoIP/GeoLite2-City.mmdb"
ASN_DB="/usr/share/GeoIP/GeoLite2-ASN.mmdb"
COMMON_PY="xdp_common"
# Ensure Python can find xdp_common.py (installed to /usr/local/bin)
export PYTHONPATH="/usr/local/bin:${PYTHONPATH:-}"
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
CYAN='\033[0;36m'
BOLD='\033[1m'
NC='\033[0m'
log_ok() { echo -e "${GREEN}[OK]${NC} $1"; logger -t xdp-defense "OK: $1" 2>/dev/null || true; }
log_err() { echo -e "${RED}[ERROR]${NC} $1" >&2; logger -t xdp-defense -p user.err "ERROR: $1" 2>/dev/null || true; }
log_info() { echo -e "${CYAN}[INFO]${NC} $1"; logger -t xdp-defense "INFO: $1" 2>/dev/null || true; }
get_iface() {
if [ -f "$CONFIG_FILE" ]; then
local iface
iface=$(python3 -c "
import yaml
with open('$CONFIG_FILE') as f:
print(yaml.safe_load(f).get('general',{}).get('interface','eth0'))
" 2>/dev/null)
echo "${iface:-eth0}"
else
echo "eth0"
fi
}
get_map_id() {
bpftool map show -j 2>/dev/null | python3 -c "
import sys, json
try:
maps = json.load(sys.stdin)
for m in maps:
if m.get('name') == '$1':
print(m['id'])
break
except:
pass
" 2>/dev/null
}
validate_cidr() {
local input=$1
local ip prefix
if [[ "$input" == *"/"* ]]; then
ip="${input%/*}"
prefix="${input#*/}"
else
ip="$input"
prefix=32
fi
if ! [[ "$prefix" =~ ^[0-9]+$ ]] || [ "$prefix" -lt 0 ] || [ "$prefix" -gt 32 ]; then
log_err "Invalid prefix: /$prefix (must be 0-32)"
return 1
fi
IFS='.' read -r a b c d <<< "$ip"
if [ -z "$a" ] || [ -z "$b" ] || [ -z "$c" ] || [ -z "$d" ]; then
log_err "Invalid IP format: $ip"
return 1
fi
for octet in "$a" "$b" "$c" "$d"; do
if ! [[ "$octet" =~ ^[0-9]+$ ]] || [ "$octet" -lt 0 ] || [ "$octet" -gt 255 ]; then
log_err "Invalid IP octet: $octet (must be 0-255)"
return 1
fi
done
return 0
}
# ==================== Load / Unload (xdp-loader) ====================
cmd_load() {
local iface=${1:-$(get_iface)}
[ ! -f "$BPF_DIR/xdp_blocker.o" ] || [ ! -f "$BPF_DIR/xdp_ddos.o" ] && cmd_build
# Unload any existing XDP programs
xdp-loader unload "$iface" --all 2>/dev/null || true
# Create pin path
mkdir -p "$PIN_PATH"
log_info "Loading XDP blocker (priority 10) on $iface..."
xdp-loader load "$iface" "$BPF_DIR/xdp_blocker.o" \
--pin-path "$PIN_PATH"
log_info "Loading XDP DDoS (priority 20) on $iface..."
xdp-loader load "$iface" "$BPF_DIR/xdp_ddos.o" \
--pin-path "$PIN_PATH"
# Enable blocker filtering
cmd_blocker_enable quiet
# Apply rate config
cmd_ddos_config_apply quiet
log_ok "XDP Defense loaded on $iface (2 programs chained)"
# Restore blocklist
[ -f "$BLOCKLIST_FILE" ] && cmd_blocker_restore || true
# Restore blocked countries
if [ -d "$COUNTRY_DIR" ]; then
for cc_file in "$COUNTRY_DIR"/*.txt; do
[ -f "$cc_file" ] || continue
local cc
cc=$(basename "$cc_file" .txt)
log_info "Restoring country: $cc"
python3 "$LIB_DIR/xdp_country.py" add "$cc" >/dev/null 2>&1 || true
done
fi
# Restore whitelists
if [ -d "/etc/xdp-blocker/whitelist" ]; then
for wl_file in /etc/xdp-blocker/whitelist/*.txt; do
[ -f "$wl_file" ] || continue
local name
name=$(basename "$wl_file" .txt)
log_info "Restoring whitelist: $name"
python3 "$LIB_DIR/xdp_whitelist.py" add "$name" >/dev/null 2>&1 || true
done
fi
}
cmd_unload() {
local iface=${1:-$(get_iface)}
xdp-loader unload "$iface" --all 2>/dev/null || true
log_ok "XDP Defense unloaded from $iface"
}
cmd_build() {
echo "Building XDP programs..."
make -C "$PROJ_DIR" build
log_ok "Built: xdp_blocker.o + xdp_ddos.o"
}
cmd_status() {
echo -e "${BOLD}=== XDP Defense Status ===${NC}"
echo ""
# Check xdp-loader status
local iface
iface=$(get_iface)
local loader_status
loader_status=$(xdp-loader status "$iface" 2>/dev/null || true)
if echo "$loader_status" | grep -q "xdp_blocker\|xdp_ddos"; then
echo -e "XDP Programs: ${GREEN}LOADED${NC}"
echo "$loader_status" | grep -E "prog_id|prio" | head -10 | while read -r line; do
echo " $line"
done
else
# Fallback: check bpftool
local blocker_prog
blocker_prog=$(bpftool prog show 2>/dev/null | grep "xdp_blocker")
local ddos_prog
ddos_prog=$(bpftool prog show 2>/dev/null | grep "xdp_ddos")
if [ -n "$blocker_prog" ] && [ -n "$ddos_prog" ]; then
echo -e "XDP Programs: ${GREEN}LOADED (both)${NC}"
elif [ -n "$blocker_prog" ]; then
echo -e "XDP Programs: ${YELLOW}PARTIAL (blocker only)${NC}"
elif [ -n "$ddos_prog" ]; then
echo -e "XDP Programs: ${YELLOW}PARTIAL (ddos only)${NC}"
else
echo -e "XDP Programs: ${RED}NOT LOADED${NC}"
fi
fi
echo -e "Interface: ${GREEN}${iface}${NC}"
# Blocker status
local cfg_id
cfg_id=$(get_map_id config)
if [ -n "$cfg_id" ]; then
local val
val=$(bpftool map lookup id "$cfg_id" key 0 0 0 0 2>/dev/null | grep -oP '"value":\s*1' | head -1)
[ -n "$val" ] && echo -e "Blocker: ${GREEN}ENABLED${NC}" || echo -e "Blocker: ${YELLOW}DISABLED${NC}"
fi
# Map counts
local bl_id wl_id
bl_id=$(get_map_id blocklist_v4)
wl_id=$(get_map_id whitelist_v4)
if [ -n "$wl_id" ]; then
local wl_count
wl_count=$(bpftool map dump id "$wl_id" -j 2>/dev/null | python3 -c "import sys,json; print(len(json.load(sys.stdin)))" 2>/dev/null || echo 0)
echo "Whitelist: $wl_count entries"
fi
if [ -n "$bl_id" ]; then
local bl_count
bl_count=$(bpftool map dump id "$bl_id" -j 2>/dev/null | python3 -c "import sys,json; print(len(json.load(sys.stdin)))" 2>/dev/null || echo 0)
echo "Blocklist: $bl_count entries"
fi
# Blocked countries
if [ -d "$COUNTRY_DIR" ] && [ "$(ls -A "$COUNTRY_DIR" 2>/dev/null)" ]; then
echo ""
echo "Countries:"
for cc_file in "$COUNTRY_DIR"/*.txt; do
[ -f "$cc_file" ] || continue
local cc
cc=$(basename "$cc_file" .txt | tr 'a-z' 'A-Z')
local count
count=$(wc -l < "$cc_file")
echo " $cc: $count CIDRs"
done
fi
# Daemon
echo ""
if [ -f "$PID_FILE" ] && kill -0 "$(cat "$PID_FILE")" 2>/dev/null; then
echo -e "Daemon: ${GREEN}RUNNING${NC} (PID $(cat "$PID_FILE"))"
else
echo -e "Daemon: ${RED}STOPPED${NC}"
fi
# Rate config
python3 -c "
from ${COMMON_PY} import read_rate_config
cfg = read_rate_config()
if cfg:
print(f'Rate limit: {cfg[\"pps_threshold\"]} pps (window: {cfg[\"window_ns\"] // 1_000_000_000}s)')
else:
print('Rate limit: (not configured)')
" 2>/dev/null || true
# Blocked IPs
python3 -c "
from ${COMMON_PY} import dump_blocked_ips
v4 = dump_blocked_ips('blocked_ips_v4')
v6 = dump_blocked_ips('blocked_ips_v6')
print(f'Blocked IPs: {len(v4) + len(v6)}')
" 2>/dev/null || true
# AI status
echo ""
cmd_ai_status 2>/dev/null || true
}
# ==================== Blocker Commands ====================
cmd_blocker_add() {
local cidr=$1
[ -z "$cidr" ] && { log_err "Usage: xdp-defense blocker add <ip/cidr>"; exit 1; }
if [[ "$cidr" == *":"* ]]; then
local map_id
map_id=$(get_map_id blocklist_v6)
[ -z "$map_id" ] && { log_err "IPv6 map not found. Is XDP loaded?"; exit 1; }
local key_hex
key_hex=$(python3 -c "from ${COMMON_PY} import cidr_to_key_v6; print(cidr_to_key_v6('$cidr'))" 2>/dev/null)
[ -z "$key_hex" ] && { log_err "Invalid IPv6 CIDR: $cidr"; exit 1; }
bpftool map update id "$map_id" key hex $key_hex value hex 01 00 00 00 00 00 00 00 2>/dev/null
[ "$2" != "quiet" ] && log_ok "Added (v6): $cidr" || true
mkdir -p "$(dirname "$BLOCKLIST_FILE")"
grep -qxF "$cidr" "$BLOCKLIST_FILE" 2>/dev/null || echo "$cidr" >> "$BLOCKLIST_FILE"
return
fi
validate_cidr "$cidr" || exit 1
local ip prefix
if [[ "$cidr" == *"/"* ]]; then
ip="${cidr%/*}"
prefix="${cidr#*/}"
else
ip="$cidr"
prefix=32
fi
local map_id
map_id=$(get_map_id blocklist_v4)
[ -z "$map_id" ] && { log_err "Map not found. Is XDP loaded?"; exit 1; }
IFS='.' read -r a b c d <<< "$ip"
local key_hex
key_hex=$(printf '%02x 00 00 00 %02x %02x %02x %02x' "$prefix" "$a" "$b" "$c" "$d")
bpftool map update id "$map_id" key hex $key_hex value hex 01 00 00 00 00 00 00 00 2>/dev/null
[ "$2" != "quiet" ] && log_ok "Added: $cidr" || true
mkdir -p "$(dirname "$BLOCKLIST_FILE")"
grep -qxF "$cidr" "$BLOCKLIST_FILE" 2>/dev/null || echo "$cidr" >> "$BLOCKLIST_FILE"
}
cmd_blocker_del() {
local cidr=$1
[ -z "$cidr" ] && { log_err "Usage: xdp-defense blocker del <ip/cidr>"; exit 1; }
if [[ "$cidr" == *":"* ]]; then
local map_id
map_id=$(get_map_id blocklist_v6)
[ -z "$map_id" ] && { log_err "IPv6 map not found"; exit 1; }
local key_hex
key_hex=$(python3 -c "from ${COMMON_PY} import cidr_to_key_v6; print(cidr_to_key_v6('$cidr'))" 2>/dev/null)
[ -z "$key_hex" ] && { log_err "Invalid IPv6 CIDR: $cidr"; exit 1; }
bpftool map delete id "$map_id" key hex $key_hex 2>/dev/null && log_ok "Removed (v6): $cidr"
local tmpfile="${BLOCKLIST_FILE}.tmp.$$"
grep -vxF "$cidr" "$BLOCKLIST_FILE" > "$tmpfile" 2>/dev/null && mv "$tmpfile" "$BLOCKLIST_FILE" || rm -f "$tmpfile"
return
fi
validate_cidr "$cidr" || exit 1
local ip prefix
if [[ "$cidr" == *"/"* ]]; then
ip="${cidr%/*}"
prefix="${cidr#*/}"
else
ip="$cidr"
prefix=32
fi
local map_id
map_id=$(get_map_id blocklist_v4)
[ -z "$map_id" ] && { log_err "Map not found"; exit 1; }
IFS='.' read -r a b c d <<< "$ip"
local key_hex
key_hex=$(printf '%02x 00 00 00 %02x %02x %02x %02x' "$prefix" "$a" "$b" "$c" "$d")
bpftool map delete id "$map_id" key hex $key_hex 2>/dev/null && log_ok "Removed: $cidr"
local tmpfile="${BLOCKLIST_FILE}.tmp.$$"
grep -vxF "$cidr" "$BLOCKLIST_FILE" > "$tmpfile" 2>/dev/null && mv "$tmpfile" "$BLOCKLIST_FILE" || rm -f "$tmpfile"
}
cmd_blocker_list() {
local map_id
map_id=$(get_map_id blocklist_v4)
[ -z "$map_id" ] && { log_err "Map not found"; exit 1; }
echo "=== Blocked CIDRs ==="
bpftool map dump id "$map_id" -j 2>/dev/null | \
python3 -c "
import sys, json, socket, struct
data = json.load(sys.stdin)
for entry in data:
fmt = entry.get('formatted', entry)
key = fmt['key']
prefix = key['prefixlen']
addr = key['addr']
ip = socket.inet_ntoa(struct.pack('<I', addr))
print(f' {ip}/{prefix}')
" 2>/dev/null || echo " (empty)"
}
cmd_blocker_stats() {
local map_id
map_id=$(get_map_id stats_map)
[ -z "$map_id" ] && { log_err "Stats map not found"; exit 1; }
echo "=== Blocker Statistics ==="
bpftool map dump id "$map_id" -j 2>/dev/null | \
python3 -c "
import sys, json
data = json.load(sys.stdin)
labels = ['Passed', 'Dropped', 'Whitelist', 'Errors']
for entry in data:
fmt = entry.get('formatted', entry)
key = fmt['key'] if isinstance(fmt['key'], int) else int(fmt['key'][0], 16)
values = fmt.get('values', [])
total_pkts = sum(v['value']['packets'] for v in values)
total_bytes = sum(v['value']['bytes'] for v in values)
if key < len(labels):
print(f'{labels[key]:9}: {total_pkts:>10} pkts, {total_bytes:>12} bytes')
"
}
cmd_blocker_enable() {
local map_id
map_id=$(get_map_id config)
[ -n "$map_id" ] && bpftool map update id "$map_id" key 0 0 0 0 value 1 0 0 0
[ "$1" != "quiet" ] && log_ok "Blocker filtering enabled" || true
}
cmd_blocker_disable() {
local map_id
map_id=$(get_map_id config)
[ -n "$map_id" ] && bpftool map update id "$map_id" key 0 0 0 0 value 0 0 0 0
log_ok "Blocker filtering disabled"
}
cmd_blocker_restore() {
[ ! -f "$BLOCKLIST_FILE" ] && return
local count=0
while read -r cidr; do
[[ -z "$cidr" || "$cidr" == "#"* ]] && continue
cmd_blocker_add "$cidr" quiet 2>/dev/null && ((count++)) || true
done < "$BLOCKLIST_FILE"
log_ok "Restored $count blocklist entries"
}
# ==================== Country Commands ====================
cmd_country_add() {
local cc=$(echo "$1" | tr 'A-Z' 'a-z')
[ -z "$cc" ] && { log_err "Usage: xdp-defense country add <country_code>"; exit 1; }
[[ "$cc" =~ ^[a-z]{2}$ ]] || { log_err "Invalid country code: $cc (must be 2 letters)"; exit 1; }
python3 "$LIB_DIR/xdp_country.py" add "$cc"
}
cmd_country_del() {
local cc=$(echo "$1" | tr 'A-Z' 'a-z')
[ -z "$cc" ] && { log_err "Usage: xdp-defense country del <country_code>"; exit 1; }
python3 "$LIB_DIR/xdp_country.py" del "$cc"
}
cmd_country_list() {
python3 "$LIB_DIR/xdp_country.py" list
}
# ==================== Whitelist Commands ====================
cmd_whitelist_add() {
local name=$1
[ -z "$name" ] && { log_err "Usage: xdp-defense whitelist add <preset>"; exit 1; }
python3 "$LIB_DIR/xdp_whitelist.py" add "$name"
}
cmd_whitelist_del() {
local name=$1
[ -z "$name" ] && { log_err "Usage: xdp-defense whitelist del <name>"; exit 1; }
python3 "$LIB_DIR/xdp_whitelist.py" del "$name"
}
cmd_whitelist_list() {
python3 "$LIB_DIR/xdp_whitelist.py" list
}
# ==================== DDoS Commands ====================
cmd_ddos_stats() {
echo -e "${BOLD}=== DDoS Statistics ===${NC}"
python3 -c "
from ${COMMON_PY} import read_percpu_stats
stats = read_percpu_stats('global_stats', 5)
labels = ['Passed', 'Dropped (blocked)', 'Dropped (rate)', 'Total', 'Errors']
for i, label in enumerate(labels):
val = stats.get(i, 0)
print(f' {label:20}: {val:>12}')
" 2>/dev/null || log_err "Cannot read stats. Is XDP loaded?"
}
cmd_ddos_top() {
local n=${1:-10}
echo -e "${BOLD}=== Top $n IPs by Packet Count ===${NC}"
python3 -c "
from ${COMMON_PY} import dump_rate_counters
entries = dump_rate_counters('rate_counter_v4', $n)
if not entries:
print(' (empty)')
else:
print(f' {\"IP\":>18} {\"Packets\":>10} {\"Bytes\":>12}')
print(f' {\"-\"*18} {\"-\"*10} {\"-\"*12}')
for ip, pkts, bts, _ in entries:
print(f' {ip:>18} {pkts:>10} {bts:>12}')
entries6 = dump_rate_counters('rate_counter_v6', $n)
if entries6:
print()
print(' IPv6:')
for ip, pkts, bts, _ in entries6:
print(f' {ip:>42} {pkts:>10} {bts:>12}')
" 2>/dev/null || log_err "Cannot read rate counters. Is XDP loaded?"
}
cmd_ddos_blocked() {
echo -e "${BOLD}=== Blocked IPs ===${NC}"
python3 -c "
from ${COMMON_PY} import dump_blocked_ips
with open('/proc/uptime') as f:
now_ns = int(float(f.read().split()[0]) * 1_000_000_000)
entries = dump_blocked_ips('blocked_ips_v4')
entries6 = dump_blocked_ips('blocked_ips_v6')
all_entries = entries + entries6
if not all_entries:
print(' (none)')
else:
print(f' {\"IP\":>18} {\"Drops\":>10} {\"Expires\":>20}')
print(f' {\"-\"*18} {\"-\"*10} {\"-\"*20}')
for ip, expire, blocked, drops in all_entries:
if expire == 0:
exp_str = 'permanent'
else:
remain = max(0, (expire - now_ns) // 1_000_000_000)
if remain > 3600:
exp_str = f'{remain // 3600}h {(remain % 3600) // 60}m'
elif remain > 60:
exp_str = f'{remain // 60}m {remain % 60}s'
else:
exp_str = f'{remain}s'
print(f' {ip:>18} {drops:>10} {exp_str:>20}')
" 2>/dev/null || log_err "Cannot read blocked IPs. Is XDP loaded?"
}
cmd_ddos_block() {
local ip=$1
local duration=${2:-0}
[ -z "$ip" ] && { log_err "Usage: xdp-defense ddos block <ip> [duration_sec]"; exit 1; }
python3 -c "from ${COMMON_PY} import block_ip; block_ip('$ip', $duration)" 2>/dev/null || \
{ log_err "Failed to block $ip"; exit 1; }
if [ "$duration" -gt 0 ] 2>/dev/null; then
log_ok "Blocked $ip for ${duration}s"
else
log_ok "Blocked $ip (permanent)"
fi
}
cmd_ddos_unblock() {
local ip=$1
[ -z "$ip" ] && { log_err "Usage: xdp-defense ddos unblock <ip>"; exit 1; }
python3 -c "from ${COMMON_PY} import unblock_ip; unblock_ip('$ip')" 2>/dev/null || \
{ log_err "Failed to unblock $ip"; exit 1; }
log_ok "Unblocked $ip"
}
cmd_ddos_config() {
local subcmd=${1:-show}
case "$subcmd" in
show)
echo -e "${BOLD}=== Rate Configuration ===${NC}"
echo -e "\n${CYAN}Active (BPF map):${NC}"
python3 -c "
from ${COMMON_PY} import read_rate_config
cfg = read_rate_config()
if cfg:
pps = cfg['pps_threshold']
bps = cfg['bps_threshold']
win = cfg['window_ns']
print(f' PPS threshold: {pps}')
print(f' BPS threshold: {bps if bps > 0 else \"disabled\"}')
print(f' Window: {win // 1_000_000_000}s ({win}ns)')
else:
print(' (not loaded)')
" 2>/dev/null
if [ -f "$CONFIG_FILE" ]; then
echo -e "\n${CYAN}Config file ($CONFIG_FILE):${NC}"
python3 -c "
import yaml
with open('$CONFIG_FILE') as f:
cfg = yaml.safe_load(f)
rl = cfg.get('rate_limits', {})
print(f' Default PPS: {rl.get(\"default_pps\", \"N/A\")}')
print(f' Default BPS: {rl.get(\"default_bps\", \"N/A\")}')
print(f' Window: {rl.get(\"window_sec\", \"N/A\")}s')
profiles = rl.get('profiles', {})
if profiles:
print()
print(' Time profiles:')
for name, p in profiles.items():
hours = p.get('hours', '')
pps = p.get('pps', 'N/A')
print(f' {name}: pps={pps}, hours={hours}')
" 2>/dev/null
fi
;;
set)
local key=$2
local value=$3
[ -z "$key" ] || [ -z "$value" ] && { log_err "Usage: xdp-defense ddos config set <pps|bps|window> <value>"; exit 1; }
python3 -c "
from ${COMMON_PY} import read_rate_config, write_rate_config
cfg = read_rate_config()
if not cfg:
cfg = {'pps_threshold': 1000, 'bps_threshold': 0, 'window_ns': 1000000000}
key = '$key'
value = int('$value')
if key == 'pps':
cfg['pps_threshold'] = value
elif key == 'bps':
cfg['bps_threshold'] = value
elif key == 'window':
cfg['window_ns'] = value * 1000000000
else:
print(f'Unknown key: {key}')
raise SystemExit(1)
write_rate_config(cfg['pps_threshold'], cfg['bps_threshold'], cfg['window_ns'])
" 2>/dev/null || { log_err "Failed to set config"; exit 1; }
log_ok "Set $key=$value"
;;
*)
log_err "Unknown config command: $subcmd"
;;
esac
}
cmd_ddos_config_apply() {
local quiet=$1
[ ! -f "$CONFIG_FILE" ] && return
python3 -c "
import yaml
from ${COMMON_PY} import write_rate_config
with open('$CONFIG_FILE') as f:
cfg = yaml.safe_load(f)
rl = cfg.get('rate_limits', {})
pps = rl.get('default_pps', 1000)
bps = rl.get('default_bps', 0)
win = rl.get('window_sec', 1)
write_rate_config(pps, bps, win * 1000000000)
" 2>/dev/null || return 0
[ "$quiet" != "quiet" ] && log_ok "Config applied from $CONFIG_FILE" || true
}
# ==================== Daemon Commands ====================
cmd_daemon_start() {
if [ -f "$PID_FILE" ] && kill -0 "$(cat "$PID_FILE")" 2>/dev/null; then
log_err "Daemon already running (PID $(cat "$PID_FILE"))"
return 1
fi
log_info "Starting defense daemon..."
python3 "$(which xdp-defense-daemon 2>/dev/null || echo "$LIB_DIR/xdp_defense_daemon.py")" &
log_ok "Daemon started"
}
cmd_daemon_start_foreground() {
# For systemd ExecStart
exec python3 "$(which xdp-defense-daemon 2>/dev/null || echo "$LIB_DIR/xdp_defense_daemon.py")"
}
cmd_daemon_stop() {
if [ ! -f "$PID_FILE" ]; then
log_err "Daemon not running (no PID file)"
return 1
fi
local pid
pid=$(cat "$PID_FILE")
if kill -0 "$pid" 2>/dev/null; then
kill -TERM "$pid"
local i=0
while [ $i -lt 50 ] && kill -0 "$pid" 2>/dev/null; do
sleep 0.1
i=$((i + 1))
done
if kill -0 "$pid" 2>/dev/null; then
kill -9 "$pid" 2>/dev/null || true
fi
log_ok "Daemon stopped"
else
log_info "Daemon not running (stale PID file)"
fi
rm -f "$PID_FILE"
}
cmd_daemon_restart() {
cmd_daemon_stop 2>/dev/null || true
sleep 1
cmd_daemon_start
}
# ==================== AI Commands ====================
cmd_ai_status() {
local model_file="$DATA_DIR/ai_model.pkl"
if [ -f "$model_file" ]; then
local age=$(( ($(date +%s) - $(stat -c %Y "$model_file")) / 3600 ))
echo -e "AI Model: ${GREEN}TRAINED${NC} (${age}h ago)"
else
echo -e "AI Model: ${YELLOW}NOT TRAINED${NC} (in learning mode)"
fi
if [ -f "$CONFIG_FILE" ]; then
python3 -c "
import yaml
with open('$CONFIG_FILE') as f:
cfg = yaml.safe_load(f)
ai = cfg.get('ai', {})
enabled = ai.get('enabled', False)
if enabled:
print(f'AI Detection: enabled ({ai.get(\"model_type\", \"IsolationForest\")})')
else:
print('AI Detection: disabled')
" 2>/dev/null
fi
}
cmd_ai_retrain() {
log_info "Triggering AI model retrain..."
if [ -f "$PID_FILE" ] && kill -0 "$(cat "$PID_FILE")" 2>/dev/null; then
kill -USR1 "$(cat "$PID_FILE")"
log_ok "Retrain signal sent to daemon"
else
log_err "Daemon not running. Start daemon first."
exit 1
fi
}
# ==================== GeoIP ====================
cmd_geoip() {
local ip=$1
[ -z "$ip" ] && { log_err "Usage: xdp-defense geoip <ip>"; exit 1; }
echo "=== GeoIP Lookup: $ip ==="
if [ -f "$CITY_DB" ]; then
local country=$(mmdblookup --file "$CITY_DB" --ip "$ip" country names en 2>/dev/null | grep -oP '"[^"]+"' | tr -d '"')
local iso=$(mmdblookup --file "$CITY_DB" --ip "$ip" country iso_code 2>/dev/null | grep -oP '"[^"]+"' | tr -d '"')
echo "Country: $country ($iso)"
local city=$(mmdblookup --file "$CITY_DB" --ip "$ip" city names en 2>/dev/null | grep -oP '"[^"]+"' | tr -d '"')
local region=$(mmdblookup --file "$CITY_DB" --ip "$ip" subdivisions 0 names en 2>/dev/null | grep -oP '"[^"]+"' | tr -d '"')
local postal=$(mmdblookup --file "$CITY_DB" --ip "$ip" postal code 2>/dev/null | grep -oP '"[^"]+"' | tr -d '"')
local lat=$(mmdblookup --file "$CITY_DB" --ip "$ip" location latitude 2>/dev/null | grep -oP '[0-9.]+')
local lon=$(mmdblookup --file "$CITY_DB" --ip "$ip" location longitude 2>/dev/null | grep -oP '[-0-9.]+')
local tz=$(mmdblookup --file "$CITY_DB" --ip "$ip" location time_zone 2>/dev/null | grep -oP '"[^"]+"' | tr -d '"')
[ -n "$city" ] && echo "City: $city"
[ -n "$region" ] && echo "Region: $region"
[ -n "$postal" ] && echo "Postal: $postal"
[ -n "$lat" ] && [ -n "$lon" ] && echo "Location: $lat, $lon"
[ -n "$tz" ] && echo "Timezone: $tz"
elif [ -f "$GEOIP_DB" ]; then
local country=$(mmdblookup --file "$GEOIP_DB" --ip "$ip" country names en 2>/dev/null | grep -oP '"[^"]+"' | tr -d '"')
local iso=$(mmdblookup --file "$GEOIP_DB" --ip "$ip" country iso_code 2>/dev/null | grep -oP '"[^"]+"' | tr -d '"')
echo "Country: $country ($iso)"
else
echo "Country: (GeoIP DB not found)"
fi
if [ -f "$ASN_DB" ]; then
local asn=$(mmdblookup --file "$ASN_DB" --ip "$ip" autonomous_system_number 2>/dev/null | grep -oP '\d+')
local org=$(mmdblookup --file "$ASN_DB" --ip "$ip" autonomous_system_organization 2>/dev/null | grep -oP '"[^"]+"' | tr -d '"')
echo "ASN: AS$asn"
echo "Org: $org"
fi
# Check if blocked (CIDR blocklist)
local map_id
map_id=$(get_map_id blocklist_v4)
if [ -n "$map_id" ]; then
IFS='.' read -r a b c d <<< "$ip"
local key_hex
key_hex=$(printf '20 00 00 00 %02x %02x %02x %02x' "$a" "$b" "$c" "$d")
if bpftool map lookup id "$map_id" key hex $key_hex 2>/dev/null | grep -q "value"; then
echo -e "Blocker: ${RED}BLOCKED${NC}"
else
echo -e "Blocker: ${GREEN}ALLOWED${NC}"
fi
fi
# Check if DDoS blocked
local ddos_map_id
ddos_map_id=$(get_map_id blocked_ips_v4)
if [ -n "$ddos_map_id" ]; then
local ddos_key_hex
ddos_key_hex=$(python3 -c "from ${COMMON_PY} import ip_to_hex_key; print(ip_to_hex_key('$ip'))" 2>/dev/null)
if [ -n "$ddos_key_hex" ] && bpftool map lookup id "$ddos_map_id" key hex $ddos_key_hex 2>/dev/null | grep -q "value"; then
echo -e "DDoS: ${RED}BLOCKED${NC}"
else
echo -e "DDoS: ${GREEN}ALLOWED${NC}"
fi
fi
}
# ==================== Stop All ====================
cmd_stop_all() {
local iface=${1:-$(get_iface)}
cmd_daemon_stop 2>/dev/null || true
cmd_unload "$iface" 2>/dev/null || true
log_ok "All stopped"
}
# ==================== Help ====================
cmd_help() {
cat << 'EOF'
XDP Defense - Unified XDP Blocker + DDoS Defense
High-performance packet filtering with CIDR/country blocking and adaptive rate limiting
Usage: xdp-defense <command> [subcommand] [args]
System:
load [iface] Load both XDP programs via xdp-loader (default: config interface)
unload [iface] Unload all XDP programs
status Show overall status
build Rebuild XDP programs
stop-all [iface] Stop daemon and unload XDP
Blocker (CIDR/IP):
blocker add <ip/cidr> Add to blocklist
blocker del <ip/cidr> Remove from blocklist
blocker list List blocked CIDRs
blocker stats Show blocker statistics
blocker enable/disable Toggle blocker filtering
Country:
country add <cc> Block a country (e.g., br, cn, ru)
country del <cc> Unblock a country
country list List blocked countries
Whitelist:
whitelist add <preset> Add preset (cloudflare, aws, google, github)
whitelist del <name> Remove from whitelist
whitelist list List whitelisted services
DDoS:
ddos stats Show DDoS statistics
ddos top [N] Show top N IPs by packet count
ddos blocked Show blocked IPs with expiry
ddos block <ip> [sec] Block IP (optional duration, 0=permanent)
ddos unblock <ip> Unblock IP
ddos config [show|set <k> <v>] Manage rate config
AI:
ai status Show AI model status
ai retrain Trigger AI model retrain
Daemon:
daemon start Start defense daemon (background)
daemon stop Stop defense daemon
daemon restart Restart defense daemon
GeoIP:
geoip <ip> Look up IP geolocation and block status
Examples:
xdp-defense load
xdp-defense blocker add 138.121.244.0/22
xdp-defense country add br
xdp-defense whitelist add cloudflare
xdp-defense ddos block 192.168.1.100 3600
xdp-defense daemon start
xdp-defense status
EOF
}
# ==================== Backward Compatibility ====================
# If invoked as xdp-block, map to blocker subcommands
SCRIPT_NAME=$(basename "$0")
if [ "$SCRIPT_NAME" = "xdp-block" ]; then
case "${1:-help}" in
add) cmd_blocker_add "$2" "$3" ;;
del) cmd_blocker_del "$2" ;;
list) cmd_blocker_list ;;
stats) cmd_blocker_stats ;;
enable) cmd_blocker_enable ;;
disable) cmd_blocker_disable ;;
status) cmd_status ;;
load) cmd_load "$2" ;;
unload) cmd_unload "$2" ;;
build) cmd_build ;;
country-add) cmd_country_add "$2" ;;
country-del) cmd_country_del "$2" ;;
country-list) cmd_country_list ;;
geoip) cmd_geoip "$2" ;;
*) cmd_help ;;
esac
exit 0
fi
# ==================== Main ====================
mkdir -p "$DATA_DIR" 2>/dev/null || true
mkdir -p "$(dirname "$BLOCKLIST_FILE")" 2>/dev/null || true
case "${1:-help}" in
load) cmd_load "$2" ;;
unload) cmd_unload "$2" ;;
build) cmd_build ;;
status) cmd_status ;;
stop-all) cmd_stop_all "$2" ;;
blocker)
case "${2:-help}" in
add) cmd_blocker_add "$3" "$4" ;;
del) cmd_blocker_del "$3" ;;
list) cmd_blocker_list ;;
stats) cmd_blocker_stats ;;
enable) cmd_blocker_enable ;;
disable) cmd_blocker_disable ;;
*) echo "Usage: xdp-defense blocker {add|del|list|stats|enable|disable}" ;;
esac
;;
country)
case "${2:-help}" in
add) cmd_country_add "$3" ;;
del) cmd_country_del "$3" ;;
list) cmd_country_list ;;
*) echo "Usage: xdp-defense country {add|del|list}" ;;
esac
;;
whitelist)
case "${2:-help}" in
add) cmd_whitelist_add "$3" ;;
del) cmd_whitelist_del "$3" ;;
list) cmd_whitelist_list ;;
*) echo "Usage: xdp-defense whitelist {add|del|list}" ;;
esac
;;
ddos)
case "${2:-help}" in
stats) cmd_ddos_stats ;;
top) cmd_ddos_top "$3" ;;
blocked) cmd_ddos_blocked ;;
block) cmd_ddos_block "$3" "$4" ;;
unblock) cmd_ddos_unblock "$3" ;;
config) cmd_ddos_config "$3" "$4" "$5" ;;
*) echo "Usage: xdp-defense ddos {stats|top|blocked|block|unblock|config}" ;;
esac
;;
ai)
case "${2:-status}" in
status) cmd_ai_status ;;
retrain) cmd_ai_retrain ;;
*) cmd_ai_status ;;
esac
;;
daemon)
case "${2:-help}" in
start) cmd_daemon_start ;;
start-foreground) cmd_daemon_start_foreground ;;
stop) cmd_daemon_stop ;;
restart) cmd_daemon_restart ;;
*) echo "Usage: xdp-defense daemon {start|stop|restart}" ;;
esac
;;
geoip) cmd_geoip "$2" ;;
*) cmd_help ;;
esac

12
bin/xdp-startup.sh Executable file
View File

@@ -0,0 +1,12 @@
#!/bin/bash
# XDP Defense - Boot Restore Script
# Restores XDP programs and data on system startup
# Called by systemd ExecStartPre or manually
set -e
DEFENSE_CMD="/usr/local/bin/xdp-defense"
# Load XDP programs (blocker + ddos) via xdp-loader
"$DEFENSE_CMD" load
echo "XDP Defense boot restore complete"

206
bpf/xdp_blocker.c Normal file
View File

@@ -0,0 +1,206 @@
// SPDX-License-Identifier: GPL-2.0
// XDP CIDR Blocker - High-performance packet filtering with LPM trie
// Supports whitelist (allowlist) and blocklist (denylist)
// Part of xdp-defense: chained via libxdp dispatcher (priority 10)
#include <linux/bpf.h>
#include <linux/if_ether.h>
#include <linux/ip.h>
#include <linux/ipv6.h>
#include <linux/in.h>
#include <bpf/bpf_helpers.h>
#include <bpf/bpf_endian.h>
#include <xdp/xdp_helpers.h>
// LPM trie key for IPv4 CIDR matching
struct ipv4_lpm_key {
__u32 prefixlen;
__u32 addr;
};
// LPM trie key for IPv6 CIDR matching
struct ipv6_lpm_key {
__u32 prefixlen;
__u8 addr[16];
};
// VLAN header (802.1Q)
struct vlan_hdr {
__be16 h_vlan_TCI;
__be16 h_vlan_encapsulated_proto;
};
// Statistics structure
struct stats {
__u64 packets;
__u64 bytes;
};
// IPv4 WHITELIST - checked first, allows traffic even if in blocklist
// Pinned for sharing with xdp_ddos program
struct {
__uint(type, BPF_MAP_TYPE_LPM_TRIE);
__type(key, struct ipv4_lpm_key);
__type(value, __u64);
__uint(max_entries, 4096);
__uint(map_flags, BPF_F_NO_PREALLOC);
__uint(pinning, LIBBPF_PIN_BY_NAME);
} whitelist_v4 SEC(".maps");
// IPv4 BLOCKLIST - blocks traffic unless in whitelist
struct {
__uint(type, BPF_MAP_TYPE_LPM_TRIE);
__type(key, struct ipv4_lpm_key);
__type(value, __u64);
__uint(max_entries, 262144);
__uint(map_flags, BPF_F_NO_PREALLOC);
} blocklist_v4 SEC(".maps");
// IPv6 whitelist - pinned for sharing with xdp_ddos program
struct {
__uint(type, BPF_MAP_TYPE_LPM_TRIE);
__type(key, struct ipv6_lpm_key);
__type(value, __u64);
__uint(max_entries, 4096);
__uint(map_flags, BPF_F_NO_PREALLOC);
__uint(pinning, LIBBPF_PIN_BY_NAME);
} whitelist_v6 SEC(".maps");
// IPv6 blocklist
struct {
__uint(type, BPF_MAP_TYPE_LPM_TRIE);
__type(key, struct ipv6_lpm_key);
__type(value, __u64);
__uint(max_entries, 262144);
__uint(map_flags, BPF_F_NO_PREALLOC);
} blocklist_v6 SEC(".maps");
// Per-CPU statistics: 0=passed, 1=dropped, 2=whitelisted, 3=errors
struct {
__uint(type, BPF_MAP_TYPE_PERCPU_ARRAY);
__type(key, __u32);
__type(value, struct stats);
__uint(max_entries, 4);
} stats_map SEC(".maps");
// Configuration map
struct {
__uint(type, BPF_MAP_TYPE_ARRAY);
__type(key, __u32);
__type(value, __u32);
__uint(max_entries, 1); // 0=enabled/disabled
} config SEC(".maps");
static __always_inline void update_stats(__u32 idx, __u64 bytes) {
struct stats *s = bpf_map_lookup_elem(&stats_map, &idx);
if (s) {
s->packets++;
s->bytes += bytes;
}
}
SEC("xdp")
int xdp_blocker(struct xdp_md *ctx) {
void *data_end = (void *)(long)ctx->data_end;
void *data = (void *)(long)ctx->data;
// Check if enabled
__u32 cfg_key = 0;
__u32 *enabled = bpf_map_lookup_elem(&config, &cfg_key);
if (enabled && *enabled == 0) {
return XDP_PASS;
}
struct ethhdr *eth = data;
if ((void *)(eth + 1) > data_end) {
update_stats(3, 0);
return XDP_PASS;
}
__u16 eth_proto = bpf_ntohs(eth->h_proto);
__u64 pkt_len = data_end - data;
void *l3_hdr = (void *)(eth + 1);
// Handle VLAN tags (802.1Q and QinQ)
if (eth_proto == ETH_P_8021Q || eth_proto == ETH_P_8021AD) {
struct vlan_hdr *vhdr = l3_hdr;
if ((void *)(vhdr + 1) > data_end) {
update_stats(3, pkt_len);
return XDP_PASS;
}
eth_proto = bpf_ntohs(vhdr->h_vlan_encapsulated_proto);
l3_hdr = (void *)(vhdr + 1);
// Handle QinQ (double VLAN)
if (eth_proto == ETH_P_8021Q || eth_proto == ETH_P_8021AD) {
vhdr = l3_hdr;
if ((void *)(vhdr + 1) > data_end) {
update_stats(3, pkt_len);
return XDP_PASS;
}
eth_proto = bpf_ntohs(vhdr->h_vlan_encapsulated_proto);
l3_hdr = (void *)(vhdr + 1);
}
}
// Handle IPv4
if (eth_proto == ETH_P_IP) {
struct iphdr *iph = l3_hdr;
if ((void *)(iph + 1) > data_end) {
update_stats(3, pkt_len);
return XDP_PASS;
}
struct ipv4_lpm_key key = {
.prefixlen = 32,
.addr = iph->saddr,
};
// Check WHITELIST first - if whitelisted, always allow
if (bpf_map_lookup_elem(&whitelist_v4, &key)) {
update_stats(2, pkt_len); // whitelisted
return XDP_PASS;
}
// Check BLOCKLIST - if blocked and not whitelisted, drop
if (bpf_map_lookup_elem(&blocklist_v4, &key)) {
update_stats(1, pkt_len); // dropped
return XDP_DROP;
}
}
// Handle IPv6
else if (eth_proto == ETH_P_IPV6) {
struct ipv6hdr *ip6h = l3_hdr;
if ((void *)(ip6h + 1) > data_end) {
update_stats(3, pkt_len);
return XDP_PASS;
}
struct ipv6_lpm_key key = {
.prefixlen = 128,
};
__builtin_memcpy(key.addr, &ip6h->saddr, 16);
// Check WHITELIST first
if (bpf_map_lookup_elem(&whitelist_v6, &key)) {
update_stats(2, pkt_len);
return XDP_PASS;
}
// Check BLOCKLIST
if (bpf_map_lookup_elem(&blocklist_v6, &key)) {
update_stats(1, pkt_len);
return XDP_DROP;
}
}
update_stats(0, pkt_len);
return XDP_PASS;
}
char _license[] SEC("license") = "GPL";
// libxdp dispatcher configuration: priority 10, chain on XDP_PASS
struct {
__uint(priority, 10);
__uint(XDP_PASS, 1);
} XDP_RUN_CONFIG(xdp_blocker);

442
bpf/xdp_ddos.c Normal file
View File

@@ -0,0 +1,442 @@
// SPDX-License-Identifier: GPL-2.0
// XDP DDoS Defense - Adaptive rate limiting with traffic feature collection
// Per-IP rate counters, automatic blocking with expiry, AI feature aggregation
// Part of xdp-defense: chained via libxdp dispatcher (priority 20)
#include <linux/bpf.h>
#include <linux/if_ether.h>
#include <linux/ip.h>
#include <linux/ipv6.h>
#include <linux/tcp.h>
#include <linux/udp.h>
#include <linux/icmp.h>
#include <linux/in.h>
#include <bpf/bpf_helpers.h>
#include <bpf/bpf_endian.h>
#include <xdp/xdp_helpers.h>
// VLAN header (802.1Q)
struct vlan_hdr {
__be16 h_vlan_TCI;
__be16 h_vlan_encapsulated_proto;
};
// LPM trie keys for shared whitelist maps
struct ipv4_lpm_key {
__u32 prefixlen;
__u32 addr;
};
struct ipv6_lpm_key {
__u32 prefixlen;
__u8 addr[16];
};
// Rate counter entry per IP
struct rate_entry {
__u64 packets;
__u64 bytes;
__u64 last_seen; // ktime_ns
};
// Rate configuration (set by userspace daemon)
struct rate_cfg {
__u64 pps_threshold; // packets per second threshold
__u64 bps_threshold; // bytes per second threshold (0 = disabled)
__u64 window_ns; // time window in nanoseconds (default 1s)
};
// Blocked IP entry with expiry
struct block_entry {
__u64 expire_ns; // ktime_ns when block expires (0 = permanent)
__u64 blocked_at; // ktime_ns when blocked
__u64 drop_count; // packets dropped while blocked
};
// Traffic features for AI analysis (per-CPU, aggregated by daemon)
struct traffic_features {
__u64 total_packets;
__u64 total_bytes;
__u64 tcp_syn_count;
__u64 tcp_other_count;
__u64 udp_count;
__u64 icmp_count;
__u64 other_proto_count;
__u64 unique_ips_approx; // approximate via counter
__u64 small_pkt_count; // packets < 100 bytes
__u64 large_pkt_count; // packets > 1400 bytes
};
// ==================== BPF Maps ====================
// Shared whitelist maps (pinned by xdp_blocker, reused here)
struct {
__uint(type, BPF_MAP_TYPE_LPM_TRIE);
__type(key, struct ipv4_lpm_key);
__type(value, __u64);
__uint(max_entries, 4096);
__uint(map_flags, BPF_F_NO_PREALLOC);
__uint(pinning, LIBBPF_PIN_BY_NAME);
} whitelist_v4 SEC(".maps");
struct {
__uint(type, BPF_MAP_TYPE_LPM_TRIE);
__type(key, struct ipv6_lpm_key);
__type(value, __u64);
__uint(max_entries, 4096);
__uint(map_flags, BPF_F_NO_PREALLOC);
__uint(pinning, LIBBPF_PIN_BY_NAME);
} whitelist_v6 SEC(".maps");
// Per-IPv4 rate counters
struct {
__uint(type, BPF_MAP_TYPE_LRU_HASH);
__type(key, __u32); // IPv4 address
__type(value, struct rate_entry);
__uint(max_entries, 65536);
} rate_counter_v4 SEC(".maps");
// Per-IPv6 rate counters
struct {
__uint(type, BPF_MAP_TYPE_LRU_HASH);
__type(key, struct in6_addr); // IPv6 address
__type(value, struct rate_entry);
__uint(max_entries, 32768);
} rate_counter_v6 SEC(".maps");
// Rate configuration (index 0)
struct {
__uint(type, BPF_MAP_TYPE_ARRAY);
__type(key, __u32);
__type(value, struct rate_cfg);
__uint(max_entries, 1);
} rate_config SEC(".maps");
// Blocked IPv4 addresses
struct {
__uint(type, BPF_MAP_TYPE_LRU_HASH);
__type(key, __u32);
__type(value, struct block_entry);
__uint(max_entries, 16384);
} blocked_ips_v4 SEC(".maps");
// Blocked IPv6 addresses
struct {
__uint(type, BPF_MAP_TYPE_LRU_HASH);
__type(key, struct in6_addr);
__type(value, struct block_entry);
__uint(max_entries, 8192);
} blocked_ips_v6 SEC(".maps");
// Global statistics: 0=passed, 1=dropped_blocked, 2=dropped_rate, 3=total, 4=errors
struct {
__uint(type, BPF_MAP_TYPE_PERCPU_ARRAY);
__type(key, __u32);
__type(value, __u64);
__uint(max_entries, 5);
} global_stats SEC(".maps");
// Traffic features for AI (index 0)
struct {
__uint(type, BPF_MAP_TYPE_PERCPU_ARRAY);
__type(key, __u32);
__type(value, struct traffic_features);
__uint(max_entries, 1);
} traffic_feature SEC(".maps");
// ==================== Helpers ====================
static __always_inline void inc_stat(__u32 idx) {
__u64 *val = bpf_map_lookup_elem(&global_stats, &idx);
if (val)
(*val)++;
}
static __always_inline void update_features(__u64 pkt_len, __u8 proto, __u8 tcp_flags) {
__u32 key = 0;
struct traffic_features *f = bpf_map_lookup_elem(&traffic_feature, &key);
if (!f)
return;
f->total_packets++;
f->total_bytes += pkt_len;
f->unique_ips_approx++;
if (pkt_len < 100)
f->small_pkt_count++;
else if (pkt_len > 1400)
f->large_pkt_count++;
switch (proto) {
case IPPROTO_TCP:
if (tcp_flags & 0x02) // SYN
f->tcp_syn_count++;
else
f->tcp_other_count++;
break;
case IPPROTO_UDP:
f->udp_count++;
break;
case IPPROTO_ICMP:
case IPPROTO_ICMPV6:
f->icmp_count++;
break;
default:
f->other_proto_count++;
break;
}
}
// Check if an IPv4 IP is blocked (with expiry check)
static __always_inline int check_blocked_v4(__u32 ip, __u64 now) {
struct block_entry *b = bpf_map_lookup_elem(&blocked_ips_v4, &ip);
if (!b)
return 0;
// Check expiry (0 = permanent)
if (b->expire_ns != 0 && now > b->expire_ns) {
bpf_map_delete_elem(&blocked_ips_v4, &ip);
return 0;
}
b->drop_count++;
return 1;
}
// Check if an IPv6 IP is blocked (with expiry check)
static __always_inline int check_blocked_v6(struct in6_addr *ip, __u64 now) {
struct block_entry *b = bpf_map_lookup_elem(&blocked_ips_v6, ip);
if (!b)
return 0;
if (b->expire_ns != 0 && now > b->expire_ns) {
bpf_map_delete_elem(&blocked_ips_v6, ip);
return 0;
}
b->drop_count++;
return 1;
}
// Rate check for IPv4: returns 1 if rate exceeded
static __always_inline int rate_check_v4(__u32 ip, __u64 now, __u64 pkt_len) {
__u32 cfg_key = 0;
struct rate_cfg *cfg = bpf_map_lookup_elem(&rate_config, &cfg_key);
if (!cfg || cfg->pps_threshold == 0)
return 0;
__u64 window = cfg->window_ns;
if (window == 0)
window = 1000000000ULL; // default 1 second
struct rate_entry *entry = bpf_map_lookup_elem(&rate_counter_v4, &ip);
if (entry) {
__u64 elapsed = now - entry->last_seen;
if (elapsed < window) {
entry->packets++;
entry->bytes += pkt_len;
if (entry->packets > cfg->pps_threshold)
return 1;
if (cfg->bps_threshold > 0 && entry->bytes > cfg->bps_threshold)
return 1;
} else {
// Reset window
entry->packets = 1;
entry->bytes = pkt_len;
entry->last_seen = now;
}
} else {
struct rate_entry new_entry = {
.packets = 1,
.bytes = pkt_len,
.last_seen = now,
};
bpf_map_update_elem(&rate_counter_v4, &ip, &new_entry, BPF_ANY);
}
return 0;
}
// Rate check for IPv6: returns 1 if rate exceeded
static __always_inline int rate_check_v6(struct in6_addr *ip, __u64 now, __u64 pkt_len) {
__u32 cfg_key = 0;
struct rate_cfg *cfg = bpf_map_lookup_elem(&rate_config, &cfg_key);
if (!cfg || cfg->pps_threshold == 0)
return 0;
__u64 window = cfg->window_ns;
if (window == 0)
window = 1000000000ULL;
struct rate_entry *entry = bpf_map_lookup_elem(&rate_counter_v6, ip);
if (entry) {
__u64 elapsed = now - entry->last_seen;
if (elapsed < window) {
entry->packets++;
entry->bytes += pkt_len;
if (entry->packets > cfg->pps_threshold)
return 1;
if (cfg->bps_threshold > 0 && entry->bytes > cfg->bps_threshold)
return 1;
} else {
entry->packets = 1;
entry->bytes = pkt_len;
entry->last_seen = now;
}
} else {
struct rate_entry new_entry = {
.packets = 1,
.bytes = pkt_len,
.last_seen = now,
};
bpf_map_update_elem(&rate_counter_v6, ip, &new_entry, BPF_ANY);
}
return 0;
}
// ==================== Main XDP Program ====================
SEC("xdp")
int xdp_ddos(struct xdp_md *ctx) {
void *data_end = (void *)(long)ctx->data_end;
void *data = (void *)(long)ctx->data;
__u64 now = bpf_ktime_get_ns();
struct ethhdr *eth = data;
if ((void *)(eth + 1) > data_end) {
inc_stat(4);
return XDP_PASS;
}
__u16 eth_proto = bpf_ntohs(eth->h_proto);
__u64 pkt_len = data_end - data;
void *l3_hdr = (void *)(eth + 1);
// Handle VLAN tags (802.1Q and QinQ)
if (eth_proto == ETH_P_8021Q || eth_proto == ETH_P_8021AD) {
struct vlan_hdr *vhdr = l3_hdr;
if ((void *)(vhdr + 1) > data_end) {
inc_stat(4);
return XDP_PASS;
}
eth_proto = bpf_ntohs(vhdr->h_vlan_encapsulated_proto);
l3_hdr = (void *)(vhdr + 1);
// Handle QinQ (double VLAN)
if (eth_proto == ETH_P_8021Q || eth_proto == ETH_P_8021AD) {
vhdr = l3_hdr;
if ((void *)(vhdr + 1) > data_end) {
inc_stat(4);
return XDP_PASS;
}
eth_proto = bpf_ntohs(vhdr->h_vlan_encapsulated_proto);
l3_hdr = (void *)(vhdr + 1);
}
}
// Increment total counter
inc_stat(3);
// Handle IPv4
if (eth_proto == ETH_P_IP) {
struct iphdr *iph = l3_hdr;
if ((void *)(iph + 1) > data_end) {
inc_stat(4);
return XDP_PASS;
}
__u32 saddr = iph->saddr;
__u8 proto = iph->protocol;
__u8 tcp_flags = 0;
// Extract TCP flags if applicable
if (proto == IPPROTO_TCP) {
struct tcphdr *tcph = l3_hdr + (iph->ihl * 4);
if ((void *)(tcph + 1) <= data_end) {
tcp_flags = ((__u8 *)tcph)[13];
}
}
// Update traffic features (always, even for whitelisted)
update_features(pkt_len, proto, tcp_flags);
// Whitelist check - bypass rate limiting but collect stats
struct ipv4_lpm_key wl_key = {.prefixlen = 32, .addr = saddr};
if (bpf_map_lookup_elem(&whitelist_v4, &wl_key)) {
inc_stat(0); // passed
return XDP_PASS;
}
// Check blocked list
if (check_blocked_v4(saddr, now)) {
inc_stat(1);
return XDP_DROP;
}
// Rate check
if (rate_check_v4(saddr, now, pkt_len)) {
inc_stat(2);
return XDP_DROP;
}
inc_stat(0);
return XDP_PASS;
}
// Handle IPv6
else if (eth_proto == ETH_P_IPV6) {
struct ipv6hdr *ip6h = l3_hdr;
if ((void *)(ip6h + 1) > data_end) {
inc_stat(4);
return XDP_PASS;
}
struct in6_addr saddr = ip6h->saddr;
__u8 proto = ip6h->nexthdr;
__u8 tcp_flags = 0;
if (proto == IPPROTO_TCP) {
struct tcphdr *tcph = (void *)(ip6h + 1);
if ((void *)(tcph + 1) <= data_end) {
tcp_flags = ((__u8 *)tcph)[13];
}
}
// Update traffic features (always, even for whitelisted)
update_features(pkt_len, proto, tcp_flags);
// Whitelist check - bypass rate limiting but collect stats
struct ipv6_lpm_key wl_key = {.prefixlen = 128};
__builtin_memcpy(wl_key.addr, &saddr, 16);
if (bpf_map_lookup_elem(&whitelist_v6, &wl_key)) {
inc_stat(0); // passed
return XDP_PASS;
}
// Check blocked list
if (check_blocked_v6(&saddr, now)) {
inc_stat(1);
return XDP_DROP;
}
// Rate check
if (rate_check_v6(&saddr, now, pkt_len)) {
inc_stat(2);
return XDP_DROP;
}
inc_stat(0);
return XDP_PASS;
}
// Non-IP traffic: pass through
inc_stat(0);
return XDP_PASS;
}
char _license[] SEC("license") = "GPL";
// libxdp dispatcher configuration: priority 20, chain on XDP_PASS
struct {
__uint(priority, 20);
__uint(XDP_PASS, 1);
} XDP_RUN_CONFIG(xdp_ddos);

70
config/config.yaml Normal file
View File

@@ -0,0 +1,70 @@
# XDP Defense - Unified Configuration
# Combines CIDR/country/whitelist blocking with DDoS rate limiting + AI detection
general:
interface: eth0
log_level: info # debug, info, warning, error
pid_file: /var/lib/xdp-defense/daemon.pid
data_dir: /var/lib/xdp-defense
bpf_dir: /opt/xdp-defense/bpf
pin_path: /sys/fs/bpf/xdp-defense
blocker:
enabled: true
config_dir: /etc/xdp-blocker # existing data path (blocklist, countries, whitelist)
rate_limits:
default_pps: 1000 # packets per second threshold
default_bps: 0 # bytes per second (0 = disabled)
window_sec: 1 # time window in seconds
# Time-based profiles (daemon switches automatically)
profiles:
business_hours:
hours: "09:00-18:00"
weekdays: "mon-fri"
pps: 2000 # higher during business hours
bps: 0
night:
hours: "00:00-06:00"
pps: 500 # stricter at night
bps: 0
escalation:
# Violations before escalation
rate_limit_after: 1 # violations before eBPF rate limiting kicks in
temp_block_after: 5 # violations before temporary block
perm_block_after: 999999 # effectively disabled
# Temporary block duration (seconds)
temp_block_duration: 300 # 5 minutes
# Violation memory window (seconds) - violations older than this are forgotten
violation_window: 600 # 10 minutes
# Cooldown: after unblocking, track more aggressively
cooldown_multiplier: 0.5 # multiply thresholds by this after recent block
ewma:
alpha: 0.3 # EWMA smoothing factor (0-1, higher = more reactive)
poll_interval: 1 # seconds between rate counter polls
threshold_multiplier: 3.0 # alert when EWMA > multiplier * baseline
ai:
enabled: true
model_type: IsolationForest
contamination: auto # let sklearn decide boundary
n_estimators: 100 # number of trees
# Learning phase
learning_duration: 86400 # 24 hours baseline collection
min_samples: 1000 # minimum samples before training
# Inference
poll_interval: 5 # seconds between feature reads
anomaly_threshold: -0.16 # sklearn decision_function threshold
# Retraining
retrain_interval: 604800 # 7 days in seconds
model_file: /var/lib/xdp-defense/ai_model.pkl
training_data_file: /var/lib/xdp-defense/training_data.csv

View File

@@ -0,0 +1,25 @@
[Unit]
Description=XDP Defense - Unified CIDR Blocker + DDoS Defense
After=network-online.target
Wants=network-online.target
Documentation=man:xdp-defense(8)
[Service]
Type=simple
ExecStartPre=/usr/local/bin/xdp-defense load
ExecStart=/usr/local/bin/xdp-defense daemon start-foreground
ExecStop=/usr/local/bin/xdp-defense stop-all
ExecReload=/bin/kill -HUP $MAINPID
Restart=on-failure
RestartSec=5
# Security hardening
ProtectSystem=strict
ReadWritePaths=/var/lib/xdp-defense /etc/xdp-defense /etc/xdp-blocker /sys/fs/bpf
ProtectHome=true
NoNewPrivileges=false
CapabilityBoundingSet=CAP_NET_ADMIN CAP_BPF CAP_SYS_ADMIN CAP_PERFMON
AmbientCapabilities=CAP_NET_ADMIN CAP_BPF CAP_SYS_ADMIN CAP_PERFMON
[Install]
WantedBy=multi-user.target

542
lib/xdp_common.py Normal file
View File

@@ -0,0 +1,542 @@
#!/usr/bin/env python3
"""
XDP Defense - Common Utilities
Merged from xdp-blocker/xdp_common.py and xdp-ddos/xdp_ddos_common.py
Provides: map management, CIDR handling, IP encoding, rate config, block/unblock, stats
"""
import subprocess
import json
import os
import tempfile
import struct
import socket
import ipaddress
import logging
import logging.handlers
# ==================== Logging ====================
_syslog_handler = None
try:
_syslog_handler = logging.handlers.SysLogHandler(address='/dev/log')
_syslog_handler.setFormatter(logging.Formatter('xdp-defense: %(message)s'))
except Exception:
pass
audit_log = logging.getLogger('xdp-defense')
audit_log.setLevel(logging.INFO)
if _syslog_handler:
audit_log.addHandler(_syslog_handler)
# ==================== BPF Map Helpers ====================
def get_map_id(map_name):
"""Get BPF map ID by name."""
result = subprocess.run(
["bpftool", "map", "show", "-j"],
capture_output=True, text=True
)
if result.returncode != 0:
return None
try:
maps = json.loads(result.stdout)
for m in maps:
if m.get("name") == map_name:
return m.get("id")
except Exception:
pass
return None
# ==================== CIDR / IPv4 Helpers (from blocker) ====================
def validate_cidr(cidr):
"""Validate IPv4 CIDR notation. Returns (ip, prefix, parts) or raises ValueError."""
if '/' in cidr:
ip, prefix_str = cidr.split('/', 1)
prefix = int(prefix_str)
else:
ip = cidr
prefix = 32
if prefix < 0 or prefix > 32:
raise ValueError(f"Invalid prefix: /{prefix}")
parts = [int(x) for x in ip.split('.')]
if len(parts) != 4 or any(p < 0 or p > 255 for p in parts):
raise ValueError(f"Invalid IP: {ip}")
return ip, prefix, parts
def cidr_to_key(cidr):
"""Convert CIDR to LPM trie key hex string."""
_, prefix, parts = validate_cidr(cidr)
key_hex = f"{prefix:02x} 00 00 00 {parts[0]:02x} {parts[1]:02x} {parts[2]:02x} {parts[3]:02x}"
return key_hex
def is_ipv6(cidr):
"""Check if a CIDR string is IPv6."""
return ':' in cidr
def validate_cidr_v6(cidr):
"""Validate IPv6 CIDR notation. Returns (network, prefix) or raises ValueError."""
try:
net = ipaddress.IPv6Network(cidr, strict=False)
return net, net.prefixlen
except (ipaddress.AddressValueError, ValueError) as e:
raise ValueError(f"Invalid IPv6 CIDR: {cidr}") from e
def cidr_to_key_v6(cidr):
"""Convert IPv6 CIDR to LPM trie key hex string.
Key format: prefixlen (4 bytes LE) + addr (16 bytes)
"""
net, prefix = validate_cidr_v6(cidr)
addr_bytes = net.network_address.packed
prefix_hex = f"{prefix:02x} 00 00 00"
addr_hex = ' '.join(f"{b:02x}" for b in addr_bytes)
return f"{prefix_hex} {addr_hex}"
def classify_cidrs(cidrs):
"""Split a list of CIDRs into v4 and v6 lists."""
v4 = []
v6 = []
for c in cidrs:
if is_ipv6(c):
v6.append(c)
else:
v4.append(c)
return v4, v6
def batch_map_operation(map_id, cidrs, operation="update", value_hex="01 00 00 00 00 00 00 00", batch_size=1000, ipv6=False):
"""Execute batch map operations with progress reporting.
Args:
map_id: BPF map ID
cidrs: list of CIDR strings
operation: "update" or "delete"
value_hex: hex value for update operations
batch_size: number of operations per batch
Returns:
Number of processed entries
"""
total = len(cidrs)
processed = 0
for i in range(0, total, batch_size):
batch = cidrs[i:i + batch_size]
batch_file = None
try:
with tempfile.NamedTemporaryFile(mode='w', suffix='.batch', delete=False) as f:
batch_file = f.name
for cidr in batch:
try:
key_hex = cidr_to_key_v6(cidr) if ipv6 else cidr_to_key(cidr)
if operation == "update":
f.write(f"map update id {map_id} key hex {key_hex} value hex {value_hex}\n")
else:
f.write(f"map delete id {map_id} key hex {key_hex}\n")
except (ValueError, Exception):
continue
subprocess.run(
["bpftool", "batch", "file", batch_file],
capture_output=True, text=True
)
finally:
if batch_file and os.path.exists(batch_file):
os.unlink(batch_file)
processed += len(batch)
pct = processed * 100 // total if total > 0 else 100
print(f"\r Progress: {processed}/{total} ({pct}%)", end="", flush=True)
print()
proto = "v6" if ipv6 else "v4"
audit_log.info(f"batch {operation} {proto}: {processed} entries on map {map_id}")
return processed
# ==================== IP Encoding Helpers (from ddos) ====================
def ip_to_hex_key(ip_str):
"""Convert IP address to hex key for LRU_HASH maps.
IPv4: 4 bytes (network byte order)
IPv6: 16 bytes (network byte order)
Returns space-separated hex string.
"""
try:
addr = ipaddress.ip_address(ip_str)
except ValueError:
raise ValueError(f"Invalid IP address: {ip_str}")
packed = addr.packed
return ' '.join(f"{b:02x}" for b in packed)
def hex_key_to_ip(hex_bytes, version=4):
"""Convert hex byte list back to IP string.
hex_bytes: list of hex strings like ['0a', '00', '00', '01']
"""
raw = bytes(int(h, 16) for h in hex_bytes)
if version == 6:
return str(ipaddress.IPv6Address(raw))
return str(ipaddress.IPv4Address(raw))
# ==================== DDoS Stats / Features (from ddos) ====================
def read_percpu_stats(map_name="global_stats", num_entries=5):
"""Read PERCPU_ARRAY map and return summed values per key.
Returns dict: {key_index: summed_value}
"""
map_id = get_map_id(map_name)
if map_id is None:
return {}
result = subprocess.run(
["bpftool", "map", "dump", "id", str(map_id), "-j"],
capture_output=True, text=True
)
if result.returncode != 0:
return {}
try:
data = json.loads(result.stdout)
except Exception:
return {}
stats = {}
for entry in data:
fmt = entry.get("formatted", entry)
key = fmt.get("key")
if isinstance(key, list):
key = int(key[0], 16) if isinstance(key[0], str) else key[0]
elif isinstance(key, str):
key = int(key, 16)
values = fmt.get("values", [])
total = 0
for v in values:
val = v.get("value", v.get("val", 0))
if isinstance(val, str):
val = int(val, 0)
elif isinstance(val, list):
val = int.from_bytes(
bytes(int(x, 16) for x in val), byteorder='little'
)
total += val
stats[key] = total
return stats
def read_percpu_features():
"""Read traffic_features PERCPU_ARRAY and return aggregated struct.
Returns dict with field names and summed values.
"""
map_id = get_map_id("traffic_feature")
if map_id is None:
return {}
result = subprocess.run(
["bpftool", "map", "dump", "id", str(map_id), "-j"],
capture_output=True, text=True
)
if result.returncode != 0:
return {}
try:
data = json.loads(result.stdout)
except Exception:
return {}
field_names = [
"total_packets", "total_bytes", "tcp_syn_count", "tcp_other_count",
"udp_count", "icmp_count", "other_proto_count", "unique_ips_approx",
"small_pkt_count", "large_pkt_count"
]
num_fields = len(field_names)
aggregated = {f: 0 for f in field_names}
for entry in data:
fmt = entry.get("formatted", entry)
values = fmt.get("values", [])
for cpu_val in values:
val = cpu_val.get("value", [])
if isinstance(val, list) and len(val) >= num_fields * 8:
raw = bytes(int(x, 16) for x in val[:num_fields * 8])
for i, name in enumerate(field_names):
v = struct.unpack_from('<Q', raw, i * 8)[0]
aggregated[name] += v
elif isinstance(val, dict):
for name in field_names:
aggregated[name] += val.get(name, 0)
return aggregated
# ==================== Rate Config (from ddos) ====================
def read_rate_config():
"""Read current rate_config from BPF map.
Returns dict: {pps_threshold, bps_threshold, window_ns}
"""
map_id = get_map_id("rate_config")
if map_id is None:
return None
result = subprocess.run(
["bpftool", "map", "lookup", "id", str(map_id), "key", "0", "0", "0", "0", "-j"],
capture_output=True, text=True
)
if result.returncode != 0:
return None
try:
data = json.loads(result.stdout)
fmt = data.get("formatted", data)
val = fmt.get("value", {})
if isinstance(val, dict):
return {
"pps_threshold": val.get("pps_threshold", 0),
"bps_threshold": val.get("bps_threshold", 0),
"window_ns": val.get("window_ns", 0),
}
elif isinstance(val, list):
raw = bytes(int(x, 16) for x in val[:24])
pps = struct.unpack_from('<Q', raw, 0)[0]
bps = struct.unpack_from('<Q', raw, 8)[0]
win = struct.unpack_from('<Q', raw, 16)[0]
return {"pps_threshold": pps, "bps_threshold": bps, "window_ns": win}
except Exception:
pass
return None
def write_rate_config(pps_threshold, bps_threshold=0, window_ns=1000000000):
"""Write rate_config to BPF map."""
map_id = get_map_id("rate_config")
if map_id is None:
raise RuntimeError("rate_config map not found")
raw = struct.pack('<QQQ', pps_threshold, bps_threshold, window_ns)
val_hex = ' '.join(f"{b:02x}" for b in raw)
result = subprocess.run(
["bpftool", "map", "update", "id", str(map_id),
"key", "0", "0", "0", "0", "value", "hex"] + val_hex.split(),
capture_output=True, text=True
)
if result.returncode != 0:
raise RuntimeError(f"Failed to write rate_config: {result.stderr}")
audit_log.info(f"rate_config updated: pps={pps_threshold} bps={bps_threshold} window={window_ns}")
# ==================== Block / Unblock (from ddos) ====================
def block_ip(ip_str, duration_sec=0):
"""Add IP to blocked_ips map.
duration_sec: 0 = permanent, >0 = temporary block
"""
addr = ipaddress.ip_address(ip_str)
is_v6 = isinstance(addr, ipaddress.IPv6Address)
map_name = "blocked_ips_v6" if is_v6 else "blocked_ips_v4"
map_id = get_map_id(map_name)
if map_id is None:
raise RuntimeError(f"{map_name} map not found")
key_hex = ip_to_hex_key(ip_str)
if duration_sec > 0:
with open('/proc/uptime', 'r') as f:
uptime_sec = float(f.read().split()[0])
now_ns = int(uptime_sec * 1_000_000_000)
expire_ns = now_ns + (duration_sec * 1_000_000_000)
else:
expire_ns = 0
now_ns_val = 0
raw = struct.pack('<QQQ', expire_ns, now_ns_val, 0)
val_hex = ' '.join(f"{b:02x}" for b in raw)
result = subprocess.run(
["bpftool", "map", "update", "id", str(map_id),
"key", "hex"] + key_hex.split() + ["value", "hex"] + val_hex.split(),
capture_output=True, text=True
)
if result.returncode != 0:
raise RuntimeError(f"Failed to block {ip_str}: {result.stderr}")
duration_str = f"{duration_sec}s" if duration_sec > 0 else "permanent"
audit_log.info(f"blocked {ip_str} ({duration_str})")
def unblock_ip(ip_str):
"""Remove IP from blocked_ips map."""
addr = ipaddress.ip_address(ip_str)
is_v6 = isinstance(addr, ipaddress.IPv6Address)
map_name = "blocked_ips_v6" if is_v6 else "blocked_ips_v4"
map_id = get_map_id(map_name)
if map_id is None:
raise RuntimeError(f"{map_name} map not found")
key_hex = ip_to_hex_key(ip_str)
result = subprocess.run(
["bpftool", "map", "delete", "id", str(map_id),
"key", "hex"] + key_hex.split(),
capture_output=True, text=True
)
if result.returncode != 0:
raise RuntimeError(f"Failed to unblock {ip_str}: {result.stderr}")
audit_log.info(f"unblocked {ip_str}")
# ==================== Dump Helpers (from ddos) ====================
def dump_rate_counters(map_name="rate_counter_v4", top_n=10):
"""Dump top-N IPs by packet count from rate counter map.
Returns list of (ip_str, packets, bytes, last_seen_ns).
"""
map_id = get_map_id(map_name)
if map_id is None:
return []
is_v6 = "v6" in map_name
result = subprocess.run(
["bpftool", "map", "dump", "id", str(map_id), "-j"],
capture_output=True, text=True
)
if result.returncode != 0:
return []
try:
data = json.loads(result.stdout)
except Exception:
return []
entries = []
for entry in data:
fmt = entry.get("formatted", entry)
key = fmt.get("key", [])
val = fmt.get("value", {})
try:
if isinstance(key, list):
ip_str = hex_key_to_ip(key, version=6 if is_v6 else 4)
elif isinstance(key, dict):
if is_v6:
addr8 = key.get('in6_u', {}).get('u6_addr8', [])
if addr8:
raw = bytes(addr8)
ip_str = str(ipaddress.IPv6Address(raw))
else:
continue
else:
ip_str = socket.inet_ntoa(struct.pack('<I', key)) if isinstance(key, int) else str(key)
elif isinstance(key, int):
ip_str = socket.inet_ntoa(struct.pack('<I', key))
else:
continue
except Exception:
continue
try:
if isinstance(val, dict):
pkts = val.get("packets", 0)
bts = val.get("bytes", 0)
last = val.get("last_seen", 0)
elif isinstance(val, list) and len(val) >= 24:
raw = bytes(int(x, 16) for x in val[:24])
pkts = struct.unpack_from('<Q', raw, 0)[0]
bts = struct.unpack_from('<Q', raw, 8)[0]
last = struct.unpack_from('<Q', raw, 16)[0]
else:
continue
except Exception:
continue
entries.append((ip_str, pkts, bts, last))
entries.sort(key=lambda x: x[1], reverse=True)
return entries[:top_n]
def dump_blocked_ips(map_name="blocked_ips_v4"):
"""Dump all blocked IPs with their block info.
Returns list of (ip_str, expire_ns, blocked_at, drop_count).
"""
map_id = get_map_id(map_name)
if map_id is None:
return []
is_v6 = "v6" in map_name
result = subprocess.run(
["bpftool", "map", "dump", "id", str(map_id), "-j"],
capture_output=True, text=True
)
if result.returncode != 0:
return []
try:
data = json.loads(result.stdout)
except Exception:
return []
entries = []
for entry in data:
fmt = entry.get("formatted", entry)
key = fmt.get("key", [])
val = fmt.get("value", {})
try:
if isinstance(key, list):
ip_str = hex_key_to_ip(key, version=6 if is_v6 else 4)
elif isinstance(key, dict) and is_v6:
addr8 = key.get('in6_u', {}).get('u6_addr8', [])
if addr8:
ip_str = str(ipaddress.IPv6Address(bytes(addr8)))
else:
continue
elif isinstance(key, int):
ip_str = socket.inet_ntoa(struct.pack('<I', key))
else:
continue
except Exception:
continue
try:
if isinstance(val, dict):
expire = val.get("expire_ns", 0)
blocked = val.get("blocked_at", 0)
drops = val.get("drop_count", 0)
elif isinstance(val, list) and len(val) >= 24:
raw = bytes(int(x, 16) for x in val[:24])
expire = struct.unpack_from('<Q', raw, 0)[0]
blocked = struct.unpack_from('<Q', raw, 8)[0]
drops = struct.unpack_from('<Q', raw, 16)[0]
else:
continue
except Exception:
continue
entries.append((ip_str, expire, blocked, drops))
return entries

175
lib/xdp_country.py Executable file
View File

@@ -0,0 +1,175 @@
#!/usr/bin/env python3
"""
XDP Country Blocker - Fast batch IP blocking
Uses bpftool batch for high-speed map updates
"""
import sys
import time
import urllib.request
from pathlib import Path
from xdp_common import get_map_id, batch_map_operation, classify_cidrs
COUNTRY_DIR = Path("/etc/xdp-blocker/countries")
IPDENY_V4_URL = "https://www.ipdeny.com/ipblocks/data/countries/{}.zone"
IPDENY_V6_URL = "https://www.ipdeny.com/ipblocks/data/ipv6/ipv6-country-blocks/{}.zone"
def download_country(cc):
"""Download country IPv4 + IPv6 IP lists"""
COUNTRY_DIR.mkdir(parents=True, exist_ok=True)
cc_file = COUNTRY_DIR / f"{cc.lower()}.txt"
cidrs = []
# Download IPv4
print(f"[INFO] Downloading {cc.upper()} IPv4 ranges...")
try:
urllib.request.urlretrieve(IPDENY_V4_URL.format(cc.lower()), cc_file)
with open(cc_file) as f:
cidrs.extend(line.strip() for line in f if line.strip())
print(f" IPv4: {len(cidrs)} CIDRs")
except Exception as e:
print(f" [WARN] IPv4 download failed: {e}")
# Download IPv6
v6_count = 0
try:
v6_tmp = COUNTRY_DIR / f"{cc.lower()}_v6.tmp"
urllib.request.urlretrieve(IPDENY_V6_URL.format(cc.lower()), v6_tmp)
with open(v6_tmp) as f:
v6_cidrs = [line.strip() for line in f if line.strip()]
cidrs.extend(v6_cidrs)
v6_count = len(v6_cidrs)
v6_tmp.unlink(missing_ok=True)
print(f" IPv6: {v6_count} CIDRs")
except Exception:
pass
if not cidrs:
print(f"[ERROR] No IP ranges found for {cc.upper()}")
return None
with open(cc_file, 'w') as f:
f.write('\n'.join(cidrs) + '\n')
return cc_file
def add_country(cc):
"""Add country IPs to XDP blocklist using batch (IPv4 + IPv6)"""
cc = cc.lower()
cc_file = COUNTRY_DIR / f"{cc}.txt"
if not cc_file.exists():
cc_file = download_country(cc)
if not cc_file:
return False
with open(cc_file) as f:
cidrs = [line.strip() for line in f if line.strip()]
v4_cidrs, v6_cidrs = classify_cidrs(cidrs)
if v4_cidrs:
map_id = get_map_id("blocklist_v4")
if not map_id:
print("[ERROR] blocklist_v4 map not found. Is XDP loaded?")
return False
print(f"[INFO] Adding {len(v4_cidrs)} IPv4 CIDRs for {cc.upper()}...")
batch_map_operation(map_id, v4_cidrs, operation="update")
if v6_cidrs:
map_id_v6 = get_map_id("blocklist_v6")
if map_id_v6:
print(f"[INFO] Adding {len(v6_cidrs)} IPv6 CIDRs for {cc.upper()}...")
batch_map_operation(map_id_v6, v6_cidrs, operation="update", ipv6=True)
else:
print("[WARN] blocklist_v6 map not found, skipping IPv6")
print(f"[OK] Added {cc.upper()}: {len(v4_cidrs)} v4 + {len(v6_cidrs)} v6 CIDRs")
return True
def del_country(cc):
"""Remove country IPs from XDP blocklist"""
cc = cc.lower()
cc_file = COUNTRY_DIR / f"{cc}.txt"
if not cc_file.exists():
print(f"[ERROR] Country {cc.upper()} is not blocked")
return False
with open(cc_file) as f:
cidrs = [line.strip() for line in f if line.strip()]
v4_cidrs, v6_cidrs = classify_cidrs(cidrs)
if v4_cidrs:
map_id = get_map_id("blocklist_v4")
if map_id:
print(f"[INFO] Removing {len(v4_cidrs)} IPv4 CIDRs for {cc.upper()}...")
batch_map_operation(map_id, v4_cidrs, operation="delete")
if v6_cidrs:
map_id_v6 = get_map_id("blocklist_v6")
if map_id_v6:
print(f"[INFO] Removing {len(v6_cidrs)} IPv6 CIDRs for {cc.upper()}...")
batch_map_operation(map_id_v6, v6_cidrs, operation="delete", ipv6=True)
cc_file.unlink()
print(f"[OK] Removed {cc.upper()}: {len(v4_cidrs)} v4 + {len(v6_cidrs)} v6 CIDRs")
return True
def list_countries():
"""List blocked countries"""
print("=== Blocked Countries ===")
if not COUNTRY_DIR.exists():
print(" (none)")
return
files = list(COUNTRY_DIR.glob("*.txt"))
if not files:
print(" (none)")
return
for cc_file in sorted(files):
cc = cc_file.stem.upper()
count = sum(1 for _ in open(cc_file))
mtime = cc_file.stat().st_mtime
age = int((time.time() - mtime) / 86400)
print(f" {cc}: {count} CIDRs (updated {age}d ago)")
def show_help():
print("""XDP Country Blocker - Fast batch IP blocking
Usage: xdp-country <command> [args]
Commands:
add <cc> Block a country (e.g., br, cn, ru, kp)
del <cc> Unblock a country
list List blocked countries
Examples:
xdp-country add br # Block Brazil (~13K CIDRs in seconds)
xdp-country add cn # Block China
xdp-country del br # Unblock Brazil
xdp-country list
""")
def main():
if len(sys.argv) < 2:
show_help()
return
cmd = sys.argv[1]
if cmd == "add" and len(sys.argv) >= 3:
add_country(sys.argv[2])
elif cmd == "del" and len(sys.argv) >= 3:
del_country(sys.argv[2])
elif cmd == "list":
list_countries()
else:
show_help()
if __name__ == "__main__":
main()

714
lib/xdp_defense_daemon.py Executable file
View File

@@ -0,0 +1,714 @@
#!/usr/bin/env python3
"""
XDP Defense Daemon
Userspace component: EWMA-based rate analysis, AI anomaly detection,
time-profile switching, and automatic escalation.
4 worker threads + main thread (signal handling):
- EWMA Thread: polls rate counters, calculates EWMA, detects violations
- AI Thread: reads traffic features, runs Isolation Forest inference
- Profile Thread: checks time-of-day, switches rate_config profiles
- Cleanup Thread: removes expired entries from blocked_ips maps
"""
import os
import sys
import time
import signal
import threading
import logging
import logging.handlers
import csv
import pickle
from collections import defaultdict
from datetime import datetime
import yaml
# ==================== Logging ====================
log = logging.getLogger('xdp-defense-daemon')
log.setLevel(logging.INFO)
_console = logging.StreamHandler()
_console.setFormatter(logging.Formatter('%(asctime)s [%(levelname)s] %(message)s'))
log.addHandler(_console)
try:
_syslog = logging.handlers.SysLogHandler(address='/dev/log')
_syslog.setFormatter(logging.Formatter('xdp-defense-daemon: %(message)s'))
log.addHandler(_syslog)
except Exception:
pass
# ==================== Configuration ====================
DEFAULT_CONFIG = {
'general': {
'interface': 'eth0',
'log_level': 'info',
'pid_file': '/var/lib/xdp-defense/daemon.pid',
'data_dir': '/var/lib/xdp-defense',
},
'rate_limits': {
'default_pps': 1000,
'default_bps': 0,
'window_sec': 1,
'profiles': {},
},
'escalation': {
'rate_limit_after': 1,
'temp_block_after': 5,
'perm_block_after': 20,
'temp_block_duration': 300,
'violation_window': 600,
'cooldown_multiplier': 0.5,
},
'ewma': {
'alpha': 0.3,
'poll_interval': 1,
'threshold_multiplier': 3.0,
},
'ai': {
'enabled': True,
'model_type': 'IsolationForest',
'contamination': 0.05,
'n_estimators': 100,
'learning_duration': 259200,
'min_samples': 1000,
'poll_interval': 5,
'anomaly_threshold': -0.3,
'retrain_interval': 604800,
'min_packets_for_sample': 20,
'model_file': '/var/lib/xdp-defense/ai_model.pkl',
'training_data_file': '/var/lib/xdp-defense/training_data.csv',
},
}
CONFIG_PATH = '/etc/xdp-defense/config.yaml'
def load_config(path=CONFIG_PATH):
"""Load config with defaults."""
cfg = DEFAULT_CONFIG.copy()
try:
with open(path) as f:
user = yaml.safe_load(f) or {}
for section in cfg:
if section in user and isinstance(user[section], dict):
cfg[section].update(user[section])
except FileNotFoundError:
log.warning("Config not found at %s, using defaults", path)
except Exception as e:
log.error("Failed to load config: %s", e)
return cfg
# ==================== ViolationTracker ====================
class ViolationTracker:
"""Track per-IP violation counts and manage escalation."""
def __init__(self, escalation_cfg):
self.cfg = escalation_cfg
self.violations = defaultdict(list)
self.lock = threading.Lock()
def record_violation(self, ip):
"""Record a violation and return escalation level.
Returns: 'rate_limit', 'temp_block', 'perm_block', or None
"""
now = time.time()
window = self.cfg.get('violation_window', 600)
with self.lock:
self.violations[ip] = [t for t in self.violations[ip] if now - t < window]
self.violations[ip].append(now)
count = len(self.violations[ip])
perm_after = self.cfg.get('perm_block_after', 20)
temp_after = self.cfg.get('temp_block_after', 5)
if count >= perm_after:
return 'perm_block'
elif count >= temp_after:
return 'temp_block'
return 'rate_limit'
def clear(self, ip):
with self.lock:
self.violations.pop(ip, None)
def cleanup_expired(self):
"""Remove entries with no recent violations."""
now = time.time()
window = self.cfg.get('violation_window', 600)
with self.lock:
expired = [ip for ip, times in self.violations.items()
if all(now - t >= window for t in times)]
for ip in expired:
del self.violations[ip]
# ==================== EWMAAnalyzer ====================
class EWMAAnalyzer:
"""Per-IP EWMA calculation for rate anomaly detection."""
def __init__(self, alpha=0.3, threshold_multiplier=3.0):
self.alpha = alpha
self.threshold_multiplier = threshold_multiplier
self.ewma = {}
self.baseline = {}
self.lock = threading.Lock()
def update(self, ip, current_pps):
"""Update EWMA for an IP. Returns True if anomalous."""
with self.lock:
if ip not in self.ewma:
self.ewma[ip] = current_pps
self.baseline[ip] = current_pps
return False
self.ewma[ip] = self.alpha * current_pps + (1 - self.alpha) * self.ewma[ip]
self.baseline[ip] = 0.01 * current_pps + 0.99 * self.baseline[ip]
base = max(self.baseline[ip], 1)
if self.ewma[ip] > base * self.threshold_multiplier:
return True
return False
def get_stats(self, ip):
with self.lock:
return {
'ewma': self.ewma.get(ip, 0),
'baseline': self.baseline.get(ip, 0),
}
def cleanup_stale(self, active_ips):
"""Remove tracking for IPs no longer in rate counters."""
with self.lock:
stale = set(self.ewma.keys()) - set(active_ips)
for ip in stale:
self.ewma.pop(ip, None)
self.baseline.pop(ip, None)
# ==================== AIDetector ====================
class AIDetector:
"""Isolation Forest based anomaly detection on traffic features."""
def __init__(self, ai_cfg):
self.cfg = ai_cfg
self.model = None
self.scaler = None
self.started_at = time.time()
self.training_data = []
self.is_learning = True
self._retrain_requested = False
@property
def enabled(self):
return self.cfg.get('enabled', False)
def request_retrain(self):
self._retrain_requested = True
def collect_sample(self, features):
"""Collect a feature sample during learning phase."""
if not self.enabled:
return
self.training_data.append(features)
learning_dur = self.cfg.get('learning_duration', 259200)
min_samples = self.cfg.get('min_samples', 1000)
elapsed = time.time() - self.started_at
if (elapsed >= learning_dur and len(self.training_data) >= min_samples) or self._retrain_requested:
self._train()
self._retrain_requested = False
def _train(self):
"""Train the Isolation Forest model."""
try:
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import numpy as np
except ImportError:
log.error("scikit-learn not installed. AI detection disabled.")
self.cfg['enabled'] = False
return
if len(self.training_data) < 10:
log.warning("Not enough training data (%d samples)", len(self.training_data))
return
log.info("Training AI model with %d samples...", len(self.training_data))
try:
X = np.array(self.training_data)
self.scaler = StandardScaler()
X_scaled = self.scaler.fit_transform(X)
self.model = IsolationForest(
n_estimators=self.cfg.get('n_estimators', 100),
contamination=self.cfg.get('contamination', 'auto'),
random_state=42,
)
self.model.fit(X_scaled)
self.is_learning = False
model_file = self.cfg.get('model_file', '/var/lib/xdp-defense/ai_model.pkl')
with open(model_file, 'wb') as f:
pickle.dump({'model': self.model, 'scaler': self.scaler}, f)
data_file = self.cfg.get('training_data_file', '/var/lib/xdp-defense/training_data.csv')
with open(data_file, 'w', newline='') as f:
writer = csv.writer(f)
writer.writerow([
'total_packets', 'total_bytes', 'tcp_syn_count', 'tcp_other_count',
'udp_count', 'icmp_count', 'other_proto_count', 'unique_ips_approx',
'small_pkt_count', 'large_pkt_count',
'syn_ratio', 'udp_ratio', 'icmp_ratio', 'small_pkt_ratio', 'avg_pkt_size'
])
writer.writerows(self.training_data)
log.info("AI model trained and saved to %s", model_file)
except Exception as e:
log.error("AI training failed: %s", e)
def load_model(self):
"""Load a previously trained model."""
model_file = self.cfg.get('model_file', '/var/lib/xdp-defense/ai_model.pkl')
if not os.path.exists(model_file):
return False
try:
with open(model_file, 'rb') as f:
data = pickle.load(f)
self.model = data['model']
self.scaler = data['scaler']
self.is_learning = False
log.info("AI model loaded from %s", model_file)
return True
except Exception as e:
log.error("Failed to load AI model: %s", e)
return False
def predict(self, features):
"""Run anomaly detection. Returns (is_anomaly, score)."""
if not self.enabled or self.model is None:
return False, 0.0
try:
import numpy as np
X = np.array([features])
X_scaled = self.scaler.transform(X)
score = self.model.decision_function(X_scaled)[0]
threshold = self.cfg.get('anomaly_threshold', -0.3)
return score < threshold, float(score)
except Exception as e:
log.error("AI prediction error: %s", e)
return False, 0.0
# ==================== ProfileManager ====================
class ProfileManager:
"""Manage time-based rate limit profiles."""
def __init__(self, rate_cfg):
self.cfg = rate_cfg
self.current_profile = 'default'
def check_and_apply(self):
"""Check current time and apply matching profile."""
from xdp_common import write_rate_config
profiles = self.cfg.get('profiles', {})
now = datetime.now()
current_hour = now.hour
current_min = now.minute
current_time = current_hour * 60 + current_min
weekday = now.strftime('%a').lower()
matched_profile = None
matched_name = 'default'
for name, profile in profiles.items():
hours = profile.get('hours', '')
weekdays = profile.get('weekdays', '')
if weekdays:
day_range = weekdays.lower().split('-')
day_names = ['mon', 'tue', 'wed', 'thu', 'fri', 'sat', 'sun']
if len(day_range) == 2:
try:
start_idx = day_names.index(day_range[0])
end_idx = day_names.index(day_range[1])
current_idx = day_names.index(weekday)
if start_idx <= end_idx:
if not (start_idx <= current_idx <= end_idx):
continue
else:
if not (current_idx >= start_idx or current_idx <= end_idx):
continue
except ValueError:
continue
if hours:
try:
start_str, end_str = hours.split('-')
sh, sm = map(int, start_str.split(':'))
eh, em = map(int, end_str.split(':'))
start_min = sh * 60 + sm
end_min = eh * 60 + em
if start_min <= end_min:
if not (start_min <= current_time < end_min):
continue
else:
if not (current_time >= start_min or current_time < end_min):
continue
except (ValueError, AttributeError):
continue
matched_profile = profile
matched_name = name
break
if matched_name != self.current_profile:
if matched_profile:
pps = matched_profile.get('pps', self.cfg.get('default_pps', 1000))
bps = matched_profile.get('bps', self.cfg.get('default_bps', 0))
else:
pps = self.cfg.get('default_pps', 1000)
bps = self.cfg.get('default_bps', 0)
window = self.cfg.get('window_sec', 1)
try:
write_rate_config(pps, bps, window * 1_000_000_000)
log.info("Profile switched: %s -> %s (pps=%d)", self.current_profile, matched_name, pps)
self.current_profile = matched_name
except Exception as e:
log.error("Failed to apply profile %s: %s", matched_name, e)
# ==================== DDoSDaemon ====================
class DDoSDaemon:
"""Main daemon orchestrator."""
def __init__(self, config_path=CONFIG_PATH):
self.config_path = config_path
self.cfg = load_config(config_path)
self.running = False
self._stop_event = threading.Event()
self._setup_components()
def _setup_components(self):
self.violation_tracker = ViolationTracker(self.cfg['escalation'])
self.ewma_analyzer = EWMAAnalyzer(
alpha=self.cfg['ewma'].get('alpha', 0.3),
threshold_multiplier=self.cfg['ewma'].get('threshold_multiplier', 3.0),
)
self.ai_detector = AIDetector(self.cfg['ai'])
self.profile_manager = ProfileManager(self.cfg['rate_limits'])
if self.ai_detector.enabled:
self.ai_detector.load_model()
level = self.cfg['general'].get('log_level', 'info').upper()
log.setLevel(getattr(logging, level, logging.INFO))
def _write_pid(self):
pid_file = self.cfg['general'].get('pid_file', '/var/lib/xdp-defense/daemon.pid')
os.makedirs(os.path.dirname(pid_file), exist_ok=True)
with open(pid_file, 'w') as f:
f.write(str(os.getpid()))
def _remove_pid(self):
pid_file = self.cfg['general'].get('pid_file', '/var/lib/xdp-defense/daemon.pid')
try:
os.unlink(pid_file)
except OSError:
pass
def _ensure_single_instance(self):
"""Stop any existing daemon before starting."""
pid_file = self.cfg['general'].get('pid_file', '/var/lib/xdp-defense/daemon.pid')
if not os.path.exists(pid_file):
return
try:
with open(pid_file) as f:
old_pid = int(f.read().strip())
os.kill(old_pid, 0)
log.info("Stopping existing daemon (PID %d)...", old_pid)
os.kill(old_pid, signal.SIGTERM)
for _ in range(30):
time.sleep(1)
try:
os.kill(old_pid, 0)
except OSError:
log.info("Old daemon stopped")
return
log.warning("Daemon PID %d did not stop, sending SIGKILL", old_pid)
os.kill(old_pid, signal.SIGKILL)
time.sleep(1)
except (ValueError, OSError):
pass
def _handle_sighup(self, signum, frame):
log.info("SIGHUP received, reloading config...")
self.cfg = load_config(self.config_path)
self._setup_components()
log.info("Config reloaded")
def _handle_sigterm(self, signum, frame):
log.info("SIGTERM received, shutting down...")
self.running = False
self._stop_event.set()
def _handle_sigusr1(self, signum, frame):
log.info("SIGUSR1 received, requesting AI retrain...")
self.ai_detector.request_retrain()
# ---- Worker Threads ----
def _ewma_thread(self):
"""Poll rate counters, compute EWMA, detect violations, escalate."""
from xdp_common import dump_rate_counters, block_ip
interval = self.cfg['ewma'].get('poll_interval', 1)
prev_counters = {}
while self.running:
try:
entries = dump_rate_counters('rate_counter_v4', top_n=1000)
active_ips = []
for ip_str, pkts, bts, last_seen in entries:
active_ips.append(ip_str)
prev = prev_counters.get(ip_str, 0)
delta = pkts - prev if pkts >= prev else pkts
prev_counters[ip_str] = pkts
if delta <= 0:
continue
pps = delta / max(interval, 0.1)
is_anomalous = self.ewma_analyzer.update(ip_str, pps)
if is_anomalous:
level = self.violation_tracker.record_violation(ip_str)
ew = self.ewma_analyzer.get_stats(ip_str)
log.warning(
"EWMA anomaly: %s pps=%.1f ewma=%.1f baseline=%.1f -> %s",
ip_str, pps, ew['ewma'], ew['baseline'], level
)
if level == 'temp_block':
dur = self.cfg['escalation'].get('temp_block_duration', 300)
try:
block_ip(ip_str, dur)
log.warning("TEMP BLOCK: %s for %ds", ip_str, dur)
except Exception as e:
log.error("Failed to temp-block %s: %s", ip_str, e)
elif level == 'perm_block':
try:
block_ip(ip_str, 0)
log.warning("PERM BLOCK: %s", ip_str)
except Exception as e:
log.error("Failed to perm-block %s: %s", ip_str, e)
self.ewma_analyzer.cleanup_stale(active_ips)
except Exception as e:
log.error("EWMA thread error: %s", e)
self._stop_event.wait(interval)
def _ai_thread(self):
"""Read traffic features, run AI inference or collect training data."""
from xdp_common import read_percpu_features, dump_rate_counters, block_ip
interval = self.cfg['ai'].get('poll_interval', 5)
prev_features = None
while self.running:
try:
if not self.ai_detector.enabled:
self._stop_event.wait(interval)
continue
features = read_percpu_features()
if not features:
self._stop_event.wait(interval)
continue
feature_names = [
'total_packets', 'total_bytes', 'tcp_syn_count', 'tcp_other_count',
'udp_count', 'icmp_count', 'other_proto_count', 'unique_ips_approx',
'small_pkt_count', 'large_pkt_count'
]
if prev_features is not None:
deltas = []
for name in feature_names:
cur = features.get(name, 0)
prev = prev_features.get(name, 0)
deltas.append(max(0, cur - prev))
min_pkts = self.cfg['ai'].get('min_packets_for_sample', 20)
if deltas[0] < min_pkts:
prev_features = features
self._stop_event.wait(interval)
continue
total = deltas[0] + 1e-6
syn_ratio = deltas[2] / total
udp_ratio = deltas[4] / total
icmp_ratio = deltas[5] / total
small_pkt_ratio = deltas[8] / total
avg_pkt_size = deltas[1] / total
deltas.extend([syn_ratio, udp_ratio, icmp_ratio, small_pkt_ratio, avg_pkt_size])
if self.ai_detector.is_learning:
self.ai_detector.collect_sample(deltas)
if len(self.ai_detector.training_data) % 100 == 0:
log.debug("AI learning: %d samples collected",
len(self.ai_detector.training_data))
else:
is_anomaly, score = self.ai_detector.predict(deltas)
if is_anomaly:
log.warning(
"AI ANOMALY detected: score=%.4f deltas=%s",
score, dict(zip(feature_names, deltas[:len(feature_names)]))
)
top_ips = dump_rate_counters('rate_counter_v4', top_n=5)
for ip_str, pkts, bts, _ in top_ips:
level = self.violation_tracker.record_violation(ip_str)
log.warning("AI escalation: %s -> %s", ip_str, level)
if level == 'temp_block':
dur = self.cfg['escalation'].get('temp_block_duration', 300)
try:
block_ip(ip_str, dur)
log.warning("AI TEMP BLOCK: %s for %ds", ip_str, dur)
except Exception as e:
log.error("Failed to AI temp-block %s: %s", ip_str, e)
elif level == 'perm_block':
try:
block_ip(ip_str, 0)
log.warning("AI PERM BLOCK: %s", ip_str)
except Exception as e:
log.error("Failed to AI perm-block %s: %s", ip_str, e)
prev_features = features
except Exception as e:
log.error("AI thread error: %s", e)
self._stop_event.wait(interval)
def _profile_thread(self):
"""Check time-of-day and switch rate profiles."""
while self.running:
try:
self.profile_manager.check_and_apply()
except Exception as e:
log.error("Profile thread error: %s", e)
self._stop_event.wait(60)
def _cleanup_thread(self):
"""Periodically clean up expired blocked IPs and stale violations."""
from xdp_common import dump_blocked_ips, unblock_ip
while self.running:
try:
with open('/proc/uptime') as f:
now_ns = int(float(f.read().split()[0]) * 1_000_000_000)
for map_name in ['blocked_ips_v4', 'blocked_ips_v6']:
entries = dump_blocked_ips(map_name)
for ip_str, expire_ns, blocked_at, drop_count in entries:
if expire_ns != 0 and now_ns > expire_ns:
try:
unblock_ip(ip_str)
self.violation_tracker.clear(ip_str)
log.info("Expired block removed: %s (dropped %d pkts)", ip_str, drop_count)
except Exception as e:
log.error("Failed to remove expired block %s: %s", ip_str, e)
self.violation_tracker.cleanup_expired()
except Exception as e:
log.error("Cleanup thread error: %s", e)
self._stop_event.wait(60)
# ---- Main Loop ----
def run(self):
"""Start the daemon."""
log.info("XDP Defense Daemon starting...")
signal.signal(signal.SIGHUP, self._handle_sighup)
signal.signal(signal.SIGTERM, self._handle_sigterm)
signal.signal(signal.SIGINT, self._handle_sigterm)
signal.signal(signal.SIGUSR1, self._handle_sigusr1)
self._ensure_single_instance()
self._write_pid()
self.running = True
threads = [
threading.Thread(target=self._ewma_thread, name='ewma', daemon=True),
threading.Thread(target=self._ai_thread, name='ai', daemon=True),
threading.Thread(target=self._profile_thread, name='profile', daemon=True),
threading.Thread(target=self._cleanup_thread, name='cleanup', daemon=True),
]
for t in threads:
t.start()
log.info("Started %s thread", t.name)
log.info("Daemon running (PID %d)", os.getpid())
try:
while self.running:
self._stop_event.wait(1)
except KeyboardInterrupt:
pass
log.info("Shutting down...")
self.running = False
self._stop_event.set()
for t in threads:
t.join(timeout=5)
self._remove_pid()
log.info("Daemon stopped")
# ==================== Entry Point ====================
def main():
config_path = CONFIG_PATH
if len(sys.argv) > 1:
config_path = sys.argv[1]
daemon = DDoSDaemon(config_path)
daemon.run()
if __name__ == '__main__':
main()

256
lib/xdp_whitelist.py Executable file
View File

@@ -0,0 +1,256 @@
#!/usr/bin/env python3
"""
XDP Whitelist Manager - Fast batch IP whitelisting
Supports presets like Cloudflare, AWS, Google, etc.
"""
import sys
import json
import urllib.request
from pathlib import Path
from xdp_common import get_map_id, batch_map_operation, classify_cidrs
WHITELIST_DIR = Path("/etc/xdp-blocker/whitelist")
# Preset URLs for trusted services
PRESETS = {
"cloudflare": {
"v4": "https://www.cloudflare.com/ips-v4",
"v6": "https://www.cloudflare.com/ips-v6",
"desc": "Cloudflare CDN/Proxy"
},
"aws": {
"v4": "https://ip-ranges.amazonaws.com/ip-ranges.json",
"desc": "Amazon Web Services (all regions)"
},
"google": {
"v4": "https://www.gstatic.com/ipranges/cloud.json",
"desc": "Google Cloud Platform"
},
"github": {
"v4": "https://api.github.com/meta",
"desc": "GitHub Services"
}
}
def download_cloudflare():
"""Download Cloudflare IP ranges"""
cidrs = []
try:
req = urllib.request.Request(
PRESETS["cloudflare"]["v4"],
headers={"User-Agent": "xdp-whitelist/1.0"}
)
with urllib.request.urlopen(req) as r:
cidrs.extend(r.read().decode().strip().split('\n'))
print(f" Downloaded {len(cidrs)} IPv4 ranges")
except Exception as e:
print(f" [WARN] Failed to download IPv4: {e}")
return cidrs
def download_aws():
"""Download AWS IP ranges"""
cidrs = []
try:
with urllib.request.urlopen(PRESETS["aws"]["v4"]) as r:
data = json.loads(r.read().decode())
for prefix in data.get("prefixes", []):
cidrs.append(prefix["ip_prefix"])
print(f" Downloaded {len(cidrs)} IPv4 ranges")
except Exception as e:
print(f" [WARN] Failed to download: {e}")
return cidrs
def download_google():
"""Download Google Cloud IP ranges"""
cidrs = []
try:
with urllib.request.urlopen(PRESETS["google"]["v4"]) as r:
data = json.loads(r.read().decode())
for prefix in data.get("prefixes", []):
if "ipv4Prefix" in prefix:
cidrs.append(prefix["ipv4Prefix"])
print(f" Downloaded {len(cidrs)} IPv4 ranges")
except Exception as e:
print(f" [WARN] Failed to download: {e}")
return cidrs
def download_github():
"""Download GitHub IP ranges"""
cidrs = []
try:
req = urllib.request.Request(
PRESETS["github"]["v4"],
headers={"User-Agent": "xdp-whitelist"}
)
with urllib.request.urlopen(req) as r:
data = json.loads(r.read().decode())
for key in ["hooks", "web", "api", "git", "packages", "pages", "importer", "actions", "dependabot"]:
if key in data:
cidrs.extend(data[key])
cidrs = list(set(c for c in cidrs if ':' not in c))
print(f" Downloaded {len(cidrs)} IPv4 ranges")
except Exception as e:
print(f" [WARN] Failed to download: {e}")
return cidrs
def add_whitelist(name, cidrs=None):
"""Add IPs to whitelist"""
name = name.lower()
WHITELIST_DIR.mkdir(parents=True, exist_ok=True)
wl_file = WHITELIST_DIR / f"{name}.txt"
if cidrs is None and wl_file.exists():
with open(wl_file) as f:
cidrs = [line.strip() for line in f if line.strip() and ':' not in line]
if cidrs:
print(f"[INFO] Using cached {name} ({len(cidrs)} CIDRs)")
if cidrs is None:
if name == "cloudflare":
print(f"[INFO] Downloading Cloudflare IP ranges...")
cidrs = download_cloudflare()
elif name == "aws":
print(f"[INFO] Downloading AWS IP ranges...")
cidrs = download_aws()
elif name == "google":
print(f"[INFO] Downloading Google Cloud IP ranges...")
cidrs = download_google()
elif name == "github":
print(f"[INFO] Downloading GitHub IP ranges...")
cidrs = download_github()
else:
print(f"[ERROR] Unknown preset: {name}")
print(f"Available presets: {', '.join(PRESETS.keys())}")
return False
if not cidrs:
print("[ERROR] No CIDRs to add")
return False
with open(wl_file, 'w') as f:
f.write('\n'.join(cidrs))
map_id = get_map_id("whitelist_v4")
if not map_id:
print("[ERROR] whitelist_v4 map not found. Is XDP loaded?")
return False
v4_cidrs, v6_cidrs = classify_cidrs(cidrs)
if v4_cidrs:
print(f"[INFO] Adding {len(v4_cidrs)} IPv4 CIDRs to whitelist...")
batch_map_operation(map_id, v4_cidrs, operation="update", batch_size=500)
if v6_cidrs:
map_id_v6 = get_map_id("whitelist_v6")
if map_id_v6:
print(f"[INFO] Adding {len(v6_cidrs)} IPv6 CIDRs to whitelist...")
batch_map_operation(map_id_v6, v6_cidrs, operation="update", batch_size=500, ipv6=True)
print(f"[OK] Whitelisted {name}: {len(v4_cidrs)} v4 + {len(v6_cidrs)} v6 CIDRs")
return True
def del_whitelist(name):
"""Remove IPs from whitelist"""
name = name.lower()
wl_file = WHITELIST_DIR / f"{name}.txt"
if not wl_file.exists():
print(f"[ERROR] {name} is not whitelisted")
return False
map_id = get_map_id("whitelist_v4")
if not map_id:
print("[ERROR] whitelist_v4 map not found")
return False
with open(wl_file) as f:
cidrs = [line.strip() for line in f if line.strip()]
v4_cidrs, v6_cidrs = classify_cidrs(cidrs)
if v4_cidrs:
print(f"[INFO] Removing {len(v4_cidrs)} IPv4 CIDRs from whitelist...")
batch_map_operation(map_id, v4_cidrs, operation="delete", batch_size=500)
if v6_cidrs:
map_id_v6 = get_map_id("whitelist_v6")
if map_id_v6:
print(f"[INFO] Removing {len(v6_cidrs)} IPv6 CIDRs from whitelist...")
batch_map_operation(map_id_v6, v6_cidrs, operation="delete", batch_size=500, ipv6=True)
wl_file.unlink()
print(f"[OK] Removed {name} from whitelist")
return True
def list_whitelist():
"""List whitelisted services"""
print("=== Whitelisted Services ===")
if not WHITELIST_DIR.exists():
print(" (none)")
return
files = list(WHITELIST_DIR.glob("*.txt"))
if not files:
print(" (none)")
return
for wl_file in sorted(files):
name = wl_file.stem
count = sum(1 for line in open(wl_file) if line.strip() and ':' not in line)
desc = PRESETS.get(name, {}).get("desc", "Custom")
print(f" {name}: {count} CIDRs ({desc})")
def show_presets():
"""Show available presets"""
print("=== Available Presets ===")
for name, info in PRESETS.items():
print(f" {name}: {info['desc']}")
def show_help():
print("""XDP Whitelist Manager - Fast batch IP whitelisting
Usage: xdp-whitelist <command> [args]
Commands:
add <preset|file> Whitelist a preset or custom file
del <name> Remove from whitelist
list List whitelisted services
presets Show available presets
Presets:
cloudflare Cloudflare CDN/Proxy IPs
aws Amazon Web Services
google Google Cloud Platform
github GitHub Services
Examples:
xdp-whitelist add cloudflare # Whitelist Cloudflare
xdp-whitelist add aws # Whitelist AWS
xdp-whitelist del cloudflare # Remove Cloudflare
xdp-whitelist list
""")
def main():
if len(sys.argv) < 2:
show_help()
return
cmd = sys.argv[1]
if cmd == "add" and len(sys.argv) >= 3:
add_whitelist(sys.argv[2])
elif cmd == "del" and len(sys.argv) >= 3:
del_whitelist(sys.argv[2])
elif cmd == "list":
list_whitelist()
elif cmd == "presets":
show_presets()
else:
show_help()
if __name__ == "__main__":
main()