ids_detector.py

#!/usr/bin/env python3
# Javier Ferrándiz Fernández | https://github.com/javisys

import argparse
import threading
import time
import sqlite3
import subprocess
import signal
import sys
from collections import defaultdict, deque
from datetime import datetime, timedelta
import json
import os

from scapy.all import sniff, TCP, IP, ARP, DNS, DNSQR, UDP, ICMP, conf
import requests
from tabulate import tabulate
from datetime import datetime, timezone


def ensure_root():
    """Abort if not running as root. Many features (sniffing, iptables) require elevated privileges."""
    try:
        if hasattr(os, "geteuid"):
            if os.geteuid() != 0:
                print("ERROR: This script must be run as root (sudo). Exiting.")
                sys.exit(1)
        else:
            # On non-Unix systems, warn but continue
            print("WARNING: Unable to verify root privileges on this platform. Ensure you run this as administrator/root.")
    except Exception as e:
        print("Warning checking for root privileges:", e)

# Config (valores por defecto)
CONFIG = {
    'db_path': 'ids_events.db',
    'syn_window': 10,           # segundos
    'syn_threshold': 100,       # SYNs por ventana => alerta
    'scan_window': 15,          # segundos
    'scan_port_threshold': 50,  # puertos distintos en ventana => escaneo
    'icmp_window': 10,
    'icmp_threshold': 200,
    'dns_window': 30,
    'dns_subdomains_threshold': 100,
    'dns_name_length_threshold': 120,
    'ban_duration': 300,        # segundos para ban temporal
    'alert_webhook': None,      # URL opcional para POST con alertas
    'interface': None,
    'whitelist': [],            # IPs que nunca banear
    'blacklist': [],            # IPs que siempre banear al detectarlas
    'prometheus_port': None,    # si se quiere exponer métricas
}


def init_db(path):
    conn = sqlite3.connect(path, check_same_thread=False)
    cur = conn.cursor()
    cur.execute('''CREATE TABLE IF NOT EXISTS events (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        ts TEXT,
        src TEXT,
        type TEXT,
        details TEXT
    )''')
    conn.commit()
    return conn


DB = None

def log_event(src, ev_type, details):
    ts = datetime.now(timezone.utc).isoformat()
    details_json = json.dumps(details, default=str)
    if DB:
        try:
            cur = DB.cursor()
            cur.execute(
                'INSERT INTO events (ts,src,type,details) VALUES (?,?,?,?)',
                (ts, src, ev_type, details_json)
            )
            DB.commit()
        except Exception as e:
            print("DB write error:", e)
    print(f"[{ts}] {ev_type} - {src} - {details_json}")
    if CONFIG['alert_webhook']:
        try:
            requests.post(
                CONFIG['alert_webhook'],
                json={'ts': ts, 'src': src, 'type': ev_type, 'details': details},
                timeout=3
            )
        except Exception as e:
            print("Warning: failed to send alert webhook:", e)


BAN_LOCK = threading.Lock()
BANNED = {}

def apply_iptables_block(src):
    if src in CONFIG['whitelist']:
        print(f"{src} is whitelisted; skipping ban")
        return
    with BAN_LOCK:
        if src in BANNED:
            print(f"{src} already banned until {BANNED[src]}")
            return
        try:
            subprocess.run(['iptables', '-I', 'INPUT', '-s', src, '-j', 'DROP'], check=True)
            # timestamp aware usando timezone.utc
            unban_time = datetime.now(timezone.utc) + timedelta(seconds=CONFIG['ban_duration'])
            BANNED[src] = unban_time
            log_event(src, 'ban_applied', {'until': unban_time.isoformat()})
        except Exception as e:
            print(f"Failed to apply iptables rule for {src}: {e}")

def remove_iptables_block(src):
    with BAN_LOCK:
        try:
            subprocess.run(['iptables', '-D', 'INPUT', '-s', src, '-j', 'DROP'], check=True)
            log_event(src, 'ban_removed', {})
        except Exception as e:
            print(f"Failed to remove iptables rule for {src}: {e}")
        finally:
            if src in BANNED:
                del BANNED[src]

def unban_worker():
    while True:
        now = datetime.now(timezone.utc)
        to_unban = []
        with BAN_LOCK:
            for src, until in list(BANNED.items()):
                if now >= until:
                    to_unban.append(src)
        for src in to_unban:
            remove_iptables_block(src)
        time.sleep(5)


syn_records = defaultdict(deque)
port_records = defaultdict(deque)
port_sets = defaultdict(set)
icmp_records = defaultdict(deque)
dns_names = defaultdict(lambda: deque())
arp_table = {}

last_alert = defaultdict(lambda: defaultdict(lambda: datetime(1970,1,1)))

ALERT_MIN_INTERVAL = timedelta(seconds=30)  # alert spam


def now_ts():
    return datetime.utcnow().isoformat()

def should_alert(src, ev_type):
    last = last_alert[src][ev_type]
    if datetime.utcnow() - last >= ALERT_MIN_INTERVAL:
        last_alert[src][ev_type] = datetime.utcnow()
        return True
    return False

# Packet handlers / detections
def handle_syn(pkt):
    if not pkt.haslayer(TCP) or not pkt.haslayer(IP):
        return
    tcp = pkt[TCP]
    ip = pkt[IP]
    if tcp.flags & 0x02 and not (tcp.flags & 0x10):
        src = ip.src
        now = time.time()
        dq = syn_records[src]
        dq.append(now)
        # remove old entries
        while dq and now - dq[0] > CONFIG['syn_window']:
            dq.popleft()
        count = len(dq)
        if count >= CONFIG['syn_threshold'] and should_alert(src, 'syn_flood'):
            details = {'count': count, 'window': CONFIG['syn_window']}
            log_event(src, 'syn_flood', details)
            if src not in CONFIG['whitelist'] and src not in CONFIG['blacklist']:
                apply_iptables_block(src)
        port = tcp.dport
        add_port_observation(src, port)

def add_port_observation(src, port):
    t = time.time()
    dq = port_records[src]
    dq.append((t, port))
    port_sets[src].add(port)
    while dq and t - dq[0][0] > CONFIG['scan_window']:
        old = dq.popleft()
        ports = {p for _, p in dq}
        port_sets[src] = ports
    if len(port_sets[src]) >= CONFIG['scan_port_threshold'] and should_alert(src, 'port_scan'):
        details = {'unique_ports': len(port_sets[src]), 'window': CONFIG['scan_window']}
        log_event(src, 'port_scan', details)
        if src not in CONFIG['whitelist'] and src not in CONFIG['blacklist']:
            apply_iptables_block(src)

def handle_icmp(pkt):
    if not pkt.haslayer(ICMP) or not pkt.haslayer(IP):
        return
    ip = pkt[IP]
    src = ip.src
    now = time.time()
    dq = icmp_records[src]
    dq.append(now)
    while dq and now - dq[0] > CONFIG['icmp_window']:
        dq.popleft()
    count = len(dq)
    if count >= CONFIG['icmp_threshold'] and should_alert(src, 'icmp_flood'):
        details = {'count': count, 'window': CONFIG['icmp_window']}
        log_event(src, 'icmp_flood', details)
        if src not in CONFIG['whitelist'] and src not in CONFIG['blacklist']:
            apply_iptables_block(src)

def handle_arp(pkt):
    if not pkt.haslayer(ARP):
        return
    arp = pkt[ARP]
    ip = arp.psrc
    mac = arp.hwsrc
    if ip in arp_table and arp_table[ip] != mac:
        # MAC changed for same IP -> possible ARP spoofing
        src = ip
        details = {'previous_mac': arp_table[ip], 'new_mac': mac}
        log_event(src, 'arp_spoofing_detected', details)
        if should_alert(src, 'arp_spoofing') and src not in CONFIG['whitelist'] and src not in CONFIG['blacklist']:
            apply_iptables_block(src)
    arp_table[ip] = mac

def handle_dns(pkt):
    # Inspect UDP DNS queries
    if not pkt.haslayer(DNS) or not pkt.haslayer(IP):
        return
    dns = pkt[DNS]
    ip = pkt[IP]
    src = ip.src
    if dns.qr == 0 and dns.qd is not None:
        qname = dns.qd.qname.decode(errors='ignore') if isinstance(dns.qd.qname, bytes) else str(dns.qd.qname)
        qname = qname.rstrip('.')
        now = time.time()
        dq = dns_names[src]
        dq.append((now, qname))
        while dq and now - dq[0][0] > CONFIG['dns_window']:
            dq.popleft()
        if len(qname) >= CONFIG['dns_name_length_threshold'] and should_alert(src, 'dns_long_name'):
            details = {'qname': qname, 'length': len(qname)}
            log_event(src, 'dns_long_name', details)
        names = {n for _, n in dq}
        if len(names) >= CONFIG['dns_subdomains_threshold'] and should_alert(src, 'dns_high_variety'):
            details = {'unique_names': len(names), 'window': CONFIG['dns_window']}
            log_event(src, 'dns_high_variety', details)
            if src not in CONFIG['whitelist'] and src not in CONFIG['blacklist']:
                apply_iptables_block(src)

# Generic sniff callback that dispatches by protocol
def packet_callback(pkt):
    try:
        if pkt.haslayer(ARP):
            handle_arp(pkt)
        if pkt.haslayer(TCP):
            handle_syn(pkt)
        if pkt.haslayer(ICMP):
            handle_icmp(pkt)
        if pkt.haslayer(DNS):
            handle_dns(pkt)
    except Exception as e:
        print("Error handling packet:", e)

# CLI / control functions
def print_status():
    rows = []

    syn_top = [(k, len(v)) for k, v in syn_records.items()]
    syn_top = sorted(syn_top, key=lambda x: x[1], reverse=True)[:10]

    scan_top = [(k, len(port_sets[k])) for k in port_sets]
    scan_top = sorted(scan_top, key=lambda x: x[1], reverse=True)[:10]

    rows.append(["SYN counts (top)", syn_top])
    rows.append(["Port scan unique ports (top)", scan_top])

    ban_list = [(k, v.isoformat()) for k, v in BANNED.items()]
    rows.append(["Banned", ban_list])

    print(tabulate(rows, headers=["Metric", "Top 10"]))


def interactive_shell():
    print("Entering interactive shell. Type 'help' for commands.")
    try:
        while True:
            cmd = input('ids> ').strip()
            if not cmd:
                continue
            if cmd in ('q','quit','exit'):
                print("Exiting shell...")
                break
            elif cmd == 'status':
                print_status()
            elif cmd.startswith('ban '):
                ip = cmd.split()[1]
                apply_iptables_block(ip)
            elif cmd.startswith('unban '):
                ip = cmd.split()[1]
                remove_iptables_block(ip)
            elif cmd == 'events':
                cur = DB.execute('SELECT id,ts,src,type,details FROM events ORDER BY id DESC LIMIT 50')
                rows = cur.fetchall()
                for r in rows:
                    print(r)
            elif cmd == 'help':
                print("commands: status, ban <ip>, unban <ip>, events, help, exit") 
            else:
                print("Unknown command. Type 'help'.")
    except KeyboardInterrupt:
        print("Shell interrupted.")


# main loop
def parse_args():
    p = argparse.ArgumentParser(description='Lightweight Python IDS (educational)')
    p.add_argument('--interface', '-i', help='Interface to sniff (default: scapy conf.iface)', default=None)
    p.add_argument('--db', help='SQLite DB path', default=CONFIG['db_path'])
    p.add_argument('--webhook', help='Alert webhook URL', default=None)
    p.add_argument('--whitelist', help='Comma separated IPs to whitelist', default='')
    p.add_argument('--blacklist', help='Comma separated IPs to blacklist', default='')
    p.add_argument('--ban-duration', type=int, help='Ban duration in seconds', default=CONFIG['ban_duration'])
    p.add_argument('--syn-threshold', type=int, help='SYN threshold', default=CONFIG['syn_threshold'])
    p.add_argument('--scan-threshold', type=int, help='Scan (unique ports) threshold', default=CONFIG['scan_port_threshold'])
    p.add_argument('--prometheus-port', type=int, help='Expose prometheus metrics on this port (optional)', default=None)
    return p.parse_args()

def start_sniffer(interface):
    print(f"Starting sniff on interface: {interface}")
    # L2 sniffing
    try:
        sniff(iface=interface, prn=packet_callback, store=False)
    except Exception as e:
        print("Sniffing error:", e)

def main():
    # Check for root permissions
    if os.geteuid() != 0:
        print("Error: Este script necesita privilegios de root (ejecuta con sudo).")
        sys.exit(1)

    global DB
    args = parse_args()
    # require root for sniffing/iptables
    ensure_root()

    # apply args to config
    CONFIG['db_path'] = args.db
    CONFIG['alert_webhook'] = args.webhook
    if args.whitelist:
        CONFIG['whitelist'] = [x.strip() for x in args.whitelist.split(',') if x.strip()]
    if args.blacklist:
        CONFIG['blacklist'] = [x.strip() for x in args.blacklist.split(',') if x.strip()]
    CONFIG['ban_duration'] = args.ban_duration
    CONFIG['syn_threshold'] = args.syn_threshold
    CONFIG['scan_port_threshold'] = args.scan_threshold
    if args.prometheus_port:
        CONFIG['prometheus_port'] = args.prometheus_port

    DB = init_db(CONFIG['db_path'])

    for ip in CONFIG['blacklist']:
        try:
            apply_iptables_block(ip)
        except Exception as e:
            print(f"Failed to blacklist {ip}: {e}")

    t_unban = threading.Thread(target=unban_worker, daemon=True)
    t_unban.start()

    if args.interface:
        CONFIG['interface'] = args.interface
        conf.iface = args.interface

    t_sniff = threading.Thread(target=start_sniffer, args=(CONFIG['interface'],), daemon=True)
    t_sniff.start()

    try:
        interactive_shell()
    except KeyboardInterrupt:
        print("Interrupted by user, shutting down...")
    finally:
        print("Cleaning up bans...")
        for src in list(BANNED.keys()):
            try:
                remove_iptables_block(src)
            except Exception as e:
                print(e)
        print("Done")

if __name__ == '__main__':
    main()