monitor.py

#!/usr/bin/env python3

from scapy.all import sniff, IP, TCP, UDP
from collections import defaultdict, deque
from datetime import datetime, timedelta
import argparse
import logging
import json
import signal
import sys

# -----------------------------
# Argument Parsing
# -----------------------------

def parse_args():
    parser = argparse.ArgumentParser(description="Lightweight Intrusion Detection Prototype")

    parser.add_argument("-i", "--interface", required=True,
                        help="Network interface (e.g., eth0)")

    parser.add_argument("--time-window", type=int, default=10,
                        help="Time window in seconds for analysis")

    parser.add_argument("--scan-threshold", type=int, default=20,
                        help="Unique ports accessed within window before flagging scan")

    parser.add_argument("--traffic-threshold", type=int, default=150,
                        help="Packets per IP within window before flagging high traffic")

    parser.add_argument("--cooldown", type=int, default=30,
                        help="Seconds before repeating same alert for same IP")

    parser.add_argument("--logfile", default="alerts.json",
                        help="Log file (JSON format)")

    return parser.parse_args()


# -----------------------------
# Detection Engine
# -----------------------------

class IDS:

    SUSPICIOUS_PORTS = {23, 4444, 1337, 3389}

    def __init__(self, time_window, scan_threshold, traffic_threshold, cooldown):
        self.time_window = timedelta(seconds=time_window)
        self.scan_threshold = scan_threshold
        self.traffic_threshold = traffic_threshold
        self.cooldown = timedelta(seconds=cooldown)

        self.packet_times = defaultdict(deque)
        self.port_activity = defaultdict(set)
        self.last_alert_time = {}

    # Cleanup old packet timestamps
    def _cleanup(self, ip, now):
        while self.packet_times[ip] and now - self.packet_times[ip][0] > self.time_window:
            self.packet_times[ip].popleft()

    # Prevent alert spam
    def _should_alert(self, ip, alert_type, now):
        key = (ip, alert_type)
        if key in self.last_alert_time:
            if now - self.last_alert_time[key] < self.cooldown:
                return False
        self.last_alert_time[key] = now
        return True

    def analyze(self, packet):

        if not packet.haslayer(IP):
            return None

        src_ip = packet[IP].src
        now = datetime.now()

        self.packet_times[src_ip].append(now)
        self._cleanup(src_ip, now)

        # High traffic detection
        if len(self.packet_times[src_ip]) > self.traffic_threshold:
            if self._should_alert(src_ip, "high_traffic", now):
                return ("high_traffic", f"High traffic detected from {src_ip}")

        # TCP logic
        if packet.haslayer(TCP):
            dport = packet[TCP].dport
            self.port_activity[src_ip].add(dport)

            if dport in self.SUSPICIOUS_PORTS:
                if self._should_alert(src_ip, "suspicious_port", now):
                    return ("suspicious_port",
                            f"Suspicious TCP connection from {src_ip} to port {dport}")

            if len(self.port_activity[src_ip]) > self.scan_threshold:
                if self._should_alert(src_ip, "port_scan", now):
                    return ("port_scan",
                            f"Possible port scan detected from {src_ip}")

        # UDP logic
        if packet.haslayer(UDP):
            dport = packet[UDP].dport
            if dport in self.SUSPICIOUS_PORTS:
                if self._should_alert(src_ip, "suspicious_port", now):
                    return ("suspicious_port",
                            f"Suspicious UDP connection from {src_ip} to port {dport}")

        return None


# -----------------------------
# Logging
# -----------------------------

def log_json(logfile, alert_type, message):
    entry = {
        "timestamp": datetime.now().isoformat(),
        "type": alert_type,
        "message": message
    }

    print(f"[ALERT] {message}")

    with open(logfile, "a") as f:
        f.write(json.dumps(entry) + "\n")


# -----------------------------
# Main
# -----------------------------

def main():
    args = parse_args()

    ids = IDS(
        time_window=args.time_window,
        scan_threshold=args.scan_threshold,
        traffic_threshold=args.traffic_threshold,
        cooldown=args.cooldown
    )

    print("Lightweight IDS started")
    print(f"Monitoring interface: {args.interface}")
    print("Press CTRL+C to stop\n")

    def packet_handler(packet):
        result = ids.analyze(packet)
        if result:
            alert_type, message = result
            log_json(args.logfile, alert_type, message)

    signal.signal(signal.SIGINT, lambda s, f: sys.exit(0))

    sniff(
        iface=args.interface,
        prn=packet_handler,
        store=False
    )


if __name__ == "__main__":
    main()