idps.py

import os
import sys
import time
import fnmatch
import threading
import argparse
try:
    from termcolor import colored
except ImportError:
    def colored(text, color):
        return text
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from watchdog.events import FileCreatedEvent, FileDeletedEvent, FileMovedEvent, FileModifiedEvent

from monitor import monitor_network_connections, monitor_system_processes
from detector import AdvancedAnomalyDetector



class IDPSEventHandler(FileSystemEventHandler):
    """Custom event handler for file system monitoring with anomaly detection."""
    def __init__(self, ignore_patterns=None, anomaly_detector=None):
        super().__init__()
        self.ignore_patterns = ignore_patterns or []
        self.anomaly_detector = anomaly_detector


    def _get_event_type(self, event):
        if isinstance(event, FileCreatedEvent):
            return 0
        elif isinstance(event, FileDeletedEvent):
            return 1
        elif isinstance(event, FileMovedEvent):
            return 2
        elif isinstance(event, FileModifiedEvent):
            return 3
        else:
            return -1

    def _get_event_vector(self, event):
        event_type = self._get_event_type(event)
        if event_type == -1:
            return None

        file_size = 0
        if os.path.exists(event.src_path):
            file_size = os.path.getsize(event.src_path)

        return [event_type, file_size]

    def should_ignore(self, path):
        for pattern in self.ignore_patterns:
            if fnmatch.fnmatch(path, pattern):
                return True
        return False
    
    def log_event(self, event_type, path):
        timestamp = time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime())
        with open("./logs/file_log.txt", "a") as log_file:
            log_file.write(f"{timestamp} - {event_type} - {path}\n")

    def on_created(self, event):
        if self.should_ignore(event.src_path):
            return
        feature_vector = self._get_event_vector(event)
        if feature_vector is not None:
            self.anomaly_detector.add_event(feature_vector)
        print(colored(f"[ALERT] {event.src_path} has been created.", "green"))
        self.log_event("created", event.src_path)

    def on_deleted(self, event):
        if self.should_ignore(event.src_path):
            return
        feature_vector = self._get_event_vector(event)
        if feature_vector is not None:
            self.anomaly_detector.add_event(feature_vector)
        print(colored(f"[ALERT] {event.src_path} has been deleted.", "red"))
        self.log_event("deleted", event.src_path)

    def on_moved(self, event):
        if self.should_ignore(event.src_path) and self.should_ignore(event.dest_path):
            return
        feature_vector = self._get_event_vector(event)
        if feature_vector is not None:
            self.anomaly_detector.add_event(feature_vector)
        print(colored(f"[ALERT] {event.src_path} has been moved to {event.dest_path}.", "yellow"))
        self.log_event("moved", f"{event.src_path} -> {event.dest_path}")

    def on_modified(self, event):
        if self.should_ignore(event.src_path):
            return
        feature_vector = self._get_event_vector(event)
        if feature_vector is not None:
            self.anomaly_detector.add_event(feature_vector)
        print(colored(f"[ALERT] {event.src_path} has been modified.", "blue"))
        self.log_event("modified", event.src_path)


def main():
    """Entry point for the IDPS. Allows user to specify monitored paths via command line."""
    parser = argparse.ArgumentParser(description="Smart Intrusion Detection & Prevention System by Ankush RFawat")
    parser.add_argument('-p', '--paths', nargs='+', default=["./lab"], help='Directories to monitor (space separated)')
    args = parser.parse_args()

    paths = args.paths
    ignore_patterns = ["*.tmp", "*.log"]
    def alert_callback(msg):
        print(colored(f"[ANOMALY ALERT] {msg}", "magenta"))

    anomaly_detector = AdvancedAnomalyDetector(threshold=10, time_window=60, alert_callback=alert_callback)
    event_handler = IDPSEventHandler(ignore_patterns=ignore_patterns, anomaly_detector=anomaly_detector)
    observer = Observer()

    for path in paths:
        observer.schedule(event_handler, path, recursive=True)

    observer.start()

    network_monitor_thread = threading.Thread(target=monitor_network_connections)
    network_monitor_thread.start()

    process_monitor_thread = threading.Thread(target=monitor_system_processes)
    process_monitor_thread.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()
    network_monitor_thread.join()
    process_monitor_thread.join()

if __name__ == "__main__":
    main()