Build a Local AI-Powered SIEM Tool
Build a Python SIEM tool that detects threats and uses a local LLM to analyze them.
Introduction
30 Second Summary
Security teams monitor millions of log events every day looking for signs of attacks. Modern security tools use AI to cut through the noise and surface what actually matters.
In this project, you will build a local AI-powered SIEM (Security Information and Event Management) tool that simulates security logs, detects threats using rule-based logic, and sends flagged events to a locally-running AI model for intelligent analysis.
What You'll Build
A Python-based SIEM tool that generates realistic security events, flags threats with configurable rules, and displays AI-powered analysis in a color-coded terminal dashboard.
By the end of this project, you'll have:
- A log simulator and detection engine that generate realistic security events and flag threats using configurable threshold-based rules.
- An AI threat analyzer powered by a local Gemma 3 model that provides threat assessments and response recommendations for each alert.
- A Rich terminal dashboard with color-coded severity levels and real-time AI-generated threat intelligence.
- Secret Mission: Add MITRE ATT&CK technique mapping so each alert includes its tactic and technique ID.
Are there any prerequisites?
This project assumes intermediate Python knowledge (functions, classes, dictionaries, and loops). You also need a Mac with Apple Silicon (M1 or later) to run the local AI model.
Set Up Ollama and Project Environment
Before you can build a security tool that uses AI, you need a local LLM running on your machine.
In this step, you will install Ollama to run AI models locally, pull the Gemma 3 model, set up a Python project environment, and confirm everything works end-to-end.
In this step, get ready to:
- Install Ollama and pull the Gemma 3 model.
- Create a Python virtual environment and install dependencies.
- Verify the full pipeline by writing a test script that talks to the local model.
Install Ollama and pull the Gemma 3 model
Ollama lets you run large language models entirely on your Mac. No cloud API keys, no usage limits, and your data never leaves your machine.
- Install Ollama by running this command in your terminal:
curl -fsSL https://ollama.com/install.sh | shWhat does this command do?
This downloads and runs the official Ollama installer script. It installs both the Ollama desktop application and the ollama command-line tool.
- After installation completes, the Ollama app should launch automatically. You will see a llama icon in your menu bar at the top of your screen.
- If Ollama did not launch automatically, press Cmd+Space to open Spotlight, type Ollama, and press Enter to open it.
Not seeing the llama icon in your menu bar?
Make sure Ollama is running before proceeding. Look for the llama icon in your menu bar (top-right area of the screen). If the install script showed an error, check that you are running macOS 14 Sonoma or later.
Help me troubleshoot my Ollama installation on macOS.
- Pull the Gemma 3 4B model by running this command:
ollama pull gemma3:4bWhat is Gemma 3?
Gemma 3 is a lightweight language model from Google DeepMind. The 4b variant has 4 billion parameters, which is small enough to run fast on your M4 Pro while still being capable enough for security analysis tasks.
This download is around 3 GB. It may take a few minutes depending on your internet speed.
- Verify the model is available by running:
ollama list- You should see gemma3:4b listed in the output with its size and modification date.
Don't see gemma3:4b in the list?
Make sure the ollama pull command finished without errors. If you see a connection error, confirm the Ollama app is running (llama icon in the menu bar). Try running ollama pull gemma3:4b again.
Help me fix my Ollama model pull.
Create a Python virtual environment and install dependencies
Your SIEM tool needs two Python packages: ollama (to talk to the local model) and rich (to build the terminal dashboard). A virtual environment keeps these packages isolated from the rest of your system.
- Move to your Desktop by running this command:
cd ~/Desktop- Create your project folder and move into it by running these commands:
mkdir ai-siem
cd ai-siem- Confirm the folder exists by running:
ls ~/Desktop- You should see ai-siem listed in the output.
- Create a Python virtual environment by running this command:
python3 -m venv venv- Activate the virtual environment by running:
source venv/bin/activate- You should see (venv) appear at the beginning of your terminal prompt. This confirms the virtual environment is active.
What is a virtual environment?
A virtual environment is an isolated Python workspace. Packages you install here stay in this folder and do not affect your system Python. This prevents conflicts between projects that need different package versions.
- Install the two required packages by running:
pip install ollama rich- You should see output ending with a success message confirming both packages installed.
Seeing a permission error or pip not found?
Make sure you activated the virtual environment first. Your terminal prompt should start with (venv). If you do not see it, run source venv/bin/activate and try the install again.
Help me fix my pip install in a virtual environment.
Verify the full pipeline with a test script
Now that Ollama is running and your packages are installed, you will write a quick script to confirm Python can talk to the local Gemma 3 model.
- Open the ai-siem folder in VS Code. Press Cmd+Space to open Spotlight, type VS Code, and press Enter to open it.
- In VS Code, click File in the top menu bar.
- Click Open Folder.
- Navigate to Desktop, select the ai-siem folder, and click Open.
- Create a new file called test_ollama.py by clicking the new file icon in the VS Code Explorer sidebar (the page icon with a + sign).
- Type test_ollama.py as the file name and press Enter.
- Add the following code to test_ollama.py:
from ollama import chat
response = chat(
model="gemma3:4b",
messages=[{"role": "user", "content": "Say hello and confirm you are working."}],
)
print(response.message.content)What does this code do?
The chat function from the ollama library sends a message to the locally running Gemma 3 model. It works just like calling a cloud API, except everything stays on your machine.
The response object contains the model's reply in response.message.content. You print it to confirm the full pipeline (Python to Ollama to Gemma 3 and back) works.
- Save test_ollama.py by pressing Cmd+S.
- Run the test script in your terminal by running:
python3 test_ollama.py- You should see an AI-generated response in your terminal. The model will say something like a greeting confirming it is operational. The exact wording varies each time.
Seeing a connection error or timeout?
This means the Ollama server is not running. Check that the llama icon is visible in your menu bar. If it is not there, open the Ollama app again. Then re-run the script.
If you see ModuleNotFoundError: No module named 'ollama', make sure your virtual environment is active. Your prompt should start with (venv).
Help me fix my test_ollama.py script.
✔️ Awesome, I've got everything!
Great. Double check you have saved test_ollama.py and confirmed a response from the model.
ⓧ I'd like to double check the full code
Here is the complete test_ollama.py file:
from ollama import chat
response = chat(
model="gemma3:4b",
messages=[{"role": "user", "content": "Say hello and confirm you are working."}],
)
print(response.message.content)Your local AI engine is ready. Next up, you will build a log simulator that generates realistic security events for your SIEM tool to analyze.
Build the Log Simulator
Your environment is locked and loaded. Now it is time to build the first core component of your SIEM tool: a log simulator that produces realistic security events.
Real SIEM systems ingest logs from firewalls, servers, and applications. Instead of parsing real system logs (which vary wildly between machines), you will build a simulator that generates structured events mimicking SSH brute force attacks, port scans, and privilege escalation attempts.
In this step, get ready to:
- Create a log event generator that produces structured security events with realistic data.
- Build attack scenario functions that simulate SSH brute force, port scanning, and privilege escalation.
- Wire everything into a continuous event stream that mixes normal and malicious traffic.
Create the log event generator
Every security event your SIEM processes needs a consistent structure. You will define the data pools (usernames, IP addresses) and a function that generates normal-looking traffic from legitimate users.
- In VS Code, create a new file called log_simulator.py in your project folder.
- Add the imports and data constants by pasting this code:
import random
import time
from datetime import datetime
# Legitimate users who generate normal traffic
NORMAL_USERS = ["alice", "bob", "charlie", "deploy", "jenkins"]
# Common usernames attackers try during brute force attempts
ATTACK_USERS = ["root", "admin", "test", "user", "oracle", "postgres"]
# Internal network IPs (trusted traffic sources)
INTERNAL_IPS = ["10.0.1.50", "10.0.1.51", "10.0.1.52", "10.0.2.10"]
# External IPs that simulate attacker origins
EXTERNAL_IPS = ["203.0.113.42", "198.51.100.77", "192.0.2.99", "185.220.101.5"]What does this code do?
These four lists define the data pools your simulator draws from. NORMAL_USERS contains legitimate-sounding usernames. ATTACK_USERS lists common targets that attackers try during brute force attempts (like root and admin).
The IP pools separate trusted internal traffic (10.0.x.x range) from external attacker IPs. This separation lets the detection engine identify suspicious sources later.
- Below the constants, add the function that generates a single normal event:
def generate_normal_event():
# Template pairs: (event_type, log message format)
event_types = [
("ssh_success", "Accepted publickey for {user} from {ip} port 22"),
("sudo_command", "{user} : TTY=pts/0 ; PWD=/home/{user} ; COMMAND=/bin/ls"),
("file_access", "{user} accessed /var/log/syslog from {ip}"),
]
event_type, template = random.choice(event_types)
user = random.choice(NORMAL_USERS)
ip = random.choice(INTERNAL_IPS)
# Return a structured event dictionary matching SIEM log schema
return {
"timestamp": datetime.now().isoformat(),
"source_ip": ip,
"event_type": event_type,
"username": user,
"message": template.format(user=user, ip=ip),
}What does this code do?
generate_normal_event() picks a random event type (SSH login, sudo command, or file access), a random legitimate user, and an internal IP. It returns a dictionary with five fields: timestamp, source_ip, event_type, username, and message.
This consistent schema is what makes a SIEM powerful. Every event, whether normal or malicious, follows the same structure so the detection engine can process them uniformly.
- Save log_simulator.py.
- Test the normal event generator by running this command in your terminal:
python -c "from log_simulator import generate_normal_event; print(generate_normal_event())"- You should see a dictionary printed with a timestamp, source_ip, event_type, username, and message field.
Seeing a ModuleNotFoundError?
Make sure your terminal is in the same folder as log_simulator.py. Run ls to confirm you see the file listed. Also check that your virtual environment is still active (you should see the environment name in your terminal prompt).
help me fix a ModuleNotFoundError when importing log_simulator
Build attack scenario functions
Normal traffic is the baseline. Now you need functions that generate bursts of malicious activity representing real attack patterns. Each function creates a cluster of related events from the same attacker IP.
- In log_simulator.py, add the SSH brute force generator below generate_normal_event():
def generate_brute_force_events(count=8):
# Single attacker IP tries many passwords against one username
ip = random.choice(EXTERNAL_IPS)
user = random.choice(ATTACK_USERS)
events = []
for _ in range(count):
events.append({
"timestamp": datetime.now().isoformat(),
"source_ip": ip,
"event_type": "ssh_failed_login",
"username": user,
"message": f"Failed password for {user} from {ip} port 22 ssh2",
})
return eventsWhat does this code do?
generate_brute_force_events() simulates an SSH brute force attack. One external IP repeatedly tries to log in with the same username, generating 8 failed login events by default. The key signature is many ssh_failed_login events from the same IP in a short time window.
- Add the port scan generator below the brute force function:
def generate_port_scan_events(count=12):
# Single IP probes many different ports looking for open services
ip = random.choice(EXTERNAL_IPS)
events = []
ports = random.sample(range(1, 65535), count)
for port in ports:
events.append({
"timestamp": datetime.now().isoformat(),
"source_ip": ip,
"event_type": "port_scan",
"username": "unknown",
"message": f"Connection attempt from {ip} to port {port}/tcp",
})
return eventsWhat does this code do?
generate_port_scan_events() simulates network reconnaissance. A single IP probes 12 random ports, looking for open services. The random.sample() call picks unique port numbers from the full TCP range (1-65535). This pattern is a classic indicator of attacker reconnaissance.
- Add the privilege escalation generator below the port scan function:
def generate_privilege_escalation_events():
# Simulates: failed logins -> successful login -> sudo command
ip = random.choice(EXTERNAL_IPS)
user = random.choice(ATTACK_USERS)
events = []
# Initial brute force attempts
for _ in range(5):
events.append({
"timestamp": datetime.now().isoformat(),
"source_ip": ip,
"event_type": "ssh_failed_login",
"username": user,
"message": f"Failed password for {user} from {ip} port 22 ssh2",
})
# Attacker gains access
events.append({
"timestamp": datetime.now().isoformat(),
"source_ip": ip,
"event_type": "ssh_success",
"username": user,
"message": f"Accepted password for {user} from {ip} port 22 ssh2",
})
# Attacker escalates to root shell
events.append({
"timestamp": datetime.now().isoformat(),
"source_ip": ip,
"event_type": "sudo_command",
"username": user,
"message": f"{user} : TTY=pts/1 ; PWD=/root ; COMMAND=/bin/bash",
})
return eventsWhat does this code do?
generate_privilege_escalation_events() simulates the most dangerous attack pattern: an attacker brute-forces their way in, succeeds, then immediately runs a sudo command to gain root access. This three-stage sequence (failed logins, successful login, privilege escalation) is exactly what a real SOC analyst would flag as critical.
- Save log_simulator.py.
- Test the attack generators by running:
python -c "from log_simulator import generate_brute_force_events; events = generate_brute_force_events(3); [print(e) for e in events]"- You should see 3 event dictionaries printed, all sharing the same source_ip and showing ssh_failed_login as the event type.
Seeing a SyntaxError?
Check for missing commas at the end of dictionary entries inside events.append({...}). Each key-value pair except the last needs a trailing comma. Also make sure the f-string quotes match (use double quotes for the outer string since the f-string is inside the dictionary).
help me debug a SyntaxError in my log_simulator.py attack functions
Create the event stream generator
Your individual functions work. Now you need a continuous stream that mixes normal traffic with random attack bursts, just like a real network produces.
What is a Python generator?
A generator function uses yield instead of return. Each time it yields a value, it pauses and remembers its state. The next time you ask for a value, it resumes right where it left off. This is perfect for an infinite event stream that never runs out of data.
- At the bottom of log_simulator.py, add the event stream generator:
def event_stream(attack_probability=0.3):
# Infinite generator mixing normal and attack traffic
while True:
if random.random() < attack_probability:
# Pick a random attack scenario and yield its events
attack = random.choice([
generate_brute_force_events,
generate_port_scan_events,
generate_privilege_escalation_events,
])
for event in attack():
yield event
time.sleep(0.1)
else:
# Normal traffic with longer delay between events
yield generate_normal_event()
time.sleep(0.5)What does this code do?
event_stream() is an infinite generator. On each iteration, it rolls a random number. If it falls below the attack_probability (30% by default), it picks a random attack scenario and yields each event in that burst with a 0.1-second delay.
Otherwise, it yields a single normal event with a longer 0.5-second delay. The time.sleep() calls simulate realistic timing gaps between log entries. This gives the detection engine time to process events just like a real SIEM would.
- Save log_simulator.py.
- Test the full event stream by running:
python -c "from log_simulator import event_stream; g = event_stream(); [print(next(g)) for _ in range(10)]"- You should see 10 event dictionaries printed, with a mix of normal events (ssh_success, file_access) and attack events (ssh_failed_login, port_scan). The output takes a few seconds because of the built-in delays.
Only seeing normal events?
With a 30% attack probability and only 10 events, you might occasionally get a run of all normal traffic. Run the command again. If you still only see normal events, double-check that event_stream references the three attack functions correctly (no typos in function names).
help me debug why my event_stream only produces normal events
✔️ Awesome, I've got everything!
Great. Double check you have saved log_simulator.py and your test output shows a mix of normal and attack events.
ⓧ I'd like to double check the full code
Here is the complete log_simulator.py file. Compare yours against it:
import random
import time
from datetime import datetime
NORMAL_USERS = ["alice", "bob", "charlie", "deploy", "jenkins"]
ATTACK_USERS = ["root", "admin", "test", "user", "oracle", "postgres"]
INTERNAL_IPS = ["10.0.1.50", "10.0.1.51", "10.0.1.52", "10.0.2.10"]
EXTERNAL_IPS = ["203.0.113.42", "198.51.100.77", "192.0.2.99", "185.220.101.5"]
def generate_normal_event():
event_types = [
("ssh_success", "Accepted publickey for {user} from {ip} port 22"),
("sudo_command", "{user} : TTY=pts/0 ; PWD=/home/{user} ; COMMAND=/bin/ls"),
("file_access", "{user} accessed /var/log/syslog from {ip}"),
]
event_type, template = random.choice(event_types)
user = random.choice(NORMAL_USERS)
ip = random.choice(INTERNAL_IPS)
return {
"timestamp": datetime.now().isoformat(),
"source_ip": ip,
"event_type": event_type,
"username": user,
"message": template.format(user=user, ip=ip),
}
def generate_brute_force_events(count=8):
ip = random.choice(EXTERNAL_IPS)
user = random.choice(ATTACK_USERS)
events = []
for _ in range(count):
events.append({
"timestamp": datetime.now().isoformat(),
"source_ip": ip,
"event_type": "ssh_failed_login",
"username": user,
"message": f"Failed password for {user} from {ip} port 22 ssh2",
})
return events
def generate_port_scan_events(count=12):
ip = random.choice(EXTERNAL_IPS)
events = []
ports = random.sample(range(1, 65535), count)
for port in ports:
events.append({
"timestamp": datetime.now().isoformat(),
"source_ip": ip,
"event_type": "port_scan",
"username": "unknown",
"message": f"Connection attempt from {ip} to port {port}/tcp",
})
return events
def generate_privilege_escalation_events():
ip = random.choice(EXTERNAL_IPS)
user = random.choice(ATTACK_USERS)
events = []
for _ in range(5):
events.append({
"timestamp": datetime.now().isoformat(),
"source_ip": ip,
"event_type": "ssh_failed_login",
"username": user,
"message": f"Failed password for {user} from {ip} port 22 ssh2",
})
events.append({
"timestamp": datetime.now().isoformat(),
"source_ip": ip,
"event_type": "ssh_success",
"username": user,
"message": f"Accepted password for {user} from {ip} port 22 ssh2",
})
events.append({
"timestamp": datetime.now().isoformat(),
"source_ip": ip,
"event_type": "sudo_command",
"username": user,
"message": f"{user} : TTY=pts/1 ; PWD=/root ; COMMAND=/bin/bash",
})
return events
def event_stream(attack_probability=0.3):
while True:
if random.random() < attack_probability:
attack = random.choice([
generate_brute_force_events,
generate_port_scan_events,
generate_privilege_escalation_events,
])
for event in attack():
yield event
time.sleep(0.1)
else:
yield generate_normal_event()
time.sleep(0.5)Your log simulator is producing a realistic stream of security events. Next up, you will build a detection engine that watches this stream and flags suspicious patterns in real time.
Build the Detection Engine
Your log simulator generates a stream of realistic security events, including attack scenarios. Now you need something that watches that stream and decides when to raise an alarm.
Real SIEM tools use detection rules to spot suspicious patterns in log data. In this step, you will build a detection engine that maintains a sliding time window of events and fires alerts when thresholds are breached.
In this step, get ready to:
- Create a detection engine class with a sliding time window and threshold-based rules.
- Implement three detection rules: SSH brute force, port scanning, and privilege escalation.
- Test the engine by feeding simulated attack events through it.
Create the detection engine class
A detection engine needs to remember recent events so it can spot patterns over time. Your engine will keep a sliding window of the last 300 seconds of events, discarding anything older each time a new event arrives.
- Create a new file called detection_engine.py in your project folder.
- Add the class skeleton with the sliding window and rule dispatcher by pasting this code:
from collections import defaultdict
from datetime import datetime, timedelta
class DetectionEngine:
def __init__(self):
# Store recent events, generated alerts, and per-IP alert cooldowns
self.events = []
self.alerts = []
self.alerted_ips = defaultdict(dict)
def ingest(self, event):
# Add the event, then trim anything older than 300 seconds
self.events.append(event)
cutoff = datetime.now() - timedelta(seconds=300)
self.events = [
e for e in self.events
if datetime.fromisoformat(e["timestamp"]) > cutoff
]
new_alerts = self._run_rules(event)
self.alerts.extend(new_alerts)
return new_alerts
def _run_rules(self, event):
# Run each detection rule against the incoming event
alerts = []
alert = self._check_brute_force(event)
if alert:
alerts.append(alert)
alert = self._check_port_scan(event)
if alert:
alerts.append(alert)
alert = self._check_privilege_escalation(event)
if alert:
alerts.append(alert)
return alertsWhat does this code do?
The DetectionEngine class stores three things: a list of recent events, a list of generated alerts, and a dictionary tracking when each IP last triggered an alert (to prevent duplicate alerts).
The ingest() method is the entry point. Every new event flows through it. It trims events older than 300 seconds, then runs all detection rules against the current event.
Why a sliding window?
Security detection needs context. A single failed login is normal. Five failed logins from the same IP in 60 seconds is suspicious. The sliding window gives each rule access to recent history without memory growing forever.
The skeleton references three detection methods that do not exist yet. The first one detects SSH brute force attacks by counting failed login attempts from the same IP.
- Add the brute force detection method directly below _run_rules:
def _check_brute_force(self, event):
# Only trigger on failed SSH logins
if event["event_type"] != "ssh_failed_login":
return None
ip = event["source_ip"]
# Cooldown: don't re-alert for the same IP within 60 seconds
if "brute_force" in self.alerted_ips[ip]:
elapsed = datetime.now() - self.alerted_ips[ip]["brute_force"]
if elapsed < timedelta(seconds=60):
return None
# Count failed logins from this IP in the last 60 seconds
cutoff = datetime.now() - timedelta(seconds=60)
failed = [
e for e in self.events
if e["event_type"] == "ssh_failed_login"
and e["source_ip"] == ip
and datetime.fromisoformat(e["timestamp"]) > cutoff
]
# Threshold: 5 or more failed logins triggers the alert
if len(failed) >= 5:
self.alerted_ips[ip]["brute_force"] = datetime.now()
return {
"timestamp": datetime.now().isoformat(),
"severity": "HIGH",
"rule": "SSH Brute Force",
"source_ip": ip,
"description": f"{len(failed)} failed SSH logins from {ip} in 60s",
"events": failed,
}
return NoneWhat does this rule do?
This method only activates on ssh_failed_login events. It counts how many failed logins came from the same source IP in the last 60 seconds. If the count reaches 5 or more, it generates a HIGH severity alert.
The alerted_ips cooldown prevents the engine from spamming duplicate alerts for the same ongoing attack. Once an IP triggers a brute force alert, it will not trigger again for 60 seconds.
- Save detection_engine.py.
You will not see output yet. The remaining rules come next.
Add port scan and privilege escalation rules
The brute force rule catches one attack pattern. Real detection engines need multiple rules running simultaneously. Port scanning and privilege escalation each follow the same structure but watch for different event types and use different thresholds.
- Add the port scan detection method below _check_brute_force:
def _check_port_scan(self, event):
# Only trigger on port scan events
if event["event_type"] != "port_scan":
return None
ip = event["source_ip"]
# Cooldown: don't re-alert for the same IP within 120 seconds
if "port_scan" in self.alerted_ips[ip]:
elapsed = datetime.now() - self.alerted_ips[ip]["port_scan"]
if elapsed < timedelta(seconds=120):
return None
# Count port connections from this IP in the last 120 seconds
cutoff = datetime.now() - timedelta(seconds=120)
scans = [
e for e in self.events
if e["event_type"] == "port_scan"
and e["source_ip"] == ip
and datetime.fromisoformat(e["timestamp"]) > cutoff
]
# Threshold: 8 or more port connections triggers the alert
if len(scans) >= 8:
self.alerted_ips[ip]["port_scan"] = datetime.now()
return {
"timestamp": datetime.now().isoformat(),
"severity": "MEDIUM",
"rule": "Port Scan",
"source_ip": ip,
"description": f"{len(scans)} port connections from {ip} in 120s",
"events": scans,
}
return NoneWhat does this rule do?
Port scanning is when an attacker probes many ports on a target to find open services. This rule watches for 8 or more port connection attempts from the same IP within 120 seconds. It generates a MEDIUM severity alert because port scans are reconnaissance, not an active compromise.
- Add the privilege escalation detection method below _check_port_scan:
def _check_privilege_escalation(self, event):
# Only trigger on sudo commands
if event["event_type"] != "sudo_command":
return None
ip = event["source_ip"]
# Cooldown: don't re-alert for the same IP within 300 seconds
if "priv_esc" in self.alerted_ips[ip]:
elapsed = datetime.now() - self.alerted_ips[ip]["priv_esc"]
if elapsed < timedelta(seconds=300):
return None
# Look for failed logins from this IP in the last 300 seconds
cutoff = datetime.now() - timedelta(seconds=300)
failed = [
e for e in self.events
if e["event_type"] == "ssh_failed_login"
and e["source_ip"] == ip
and datetime.fromisoformat(e["timestamp"]) > cutoff
]
# Threshold: sudo after 3+ failed logins means possible compromise
if len(failed) >= 3:
self.alerted_ips[ip]["priv_esc"] = datetime.now()
return {
"timestamp": datetime.now().isoformat(),
"severity": "CRITICAL",
"rule": "Privilege Escalation",
"source_ip": ip,
"description": f"Sudo command from {ip} after {len(failed)} failed logins",
"events": failed + [event],
}
return NoneWhy is privilege escalation CRITICAL?
This rule correlates two event types. It fires when someone runs a sudo command from an IP that recently had 3 or more failed login attempts. That pattern suggests an attacker gained access after brute-forcing credentials and is now elevating privileges.
This is the most dangerous scenario because it indicates an active compromise in progress. That is why it gets CRITICAL severity.
- Save detection_engine.py.
Seeing a syntax error?
Make sure each method is indented with exactly 4 spaces inside the DetectionEngine class. Python uses indentation to determine class membership.
Check that the f" strings in each alert dictionary use curly braces around variables like {ip} and {len(failed)}.
Help me fix a syntax error in my detection_engine.py file.
✔️ Awesome, I've got everything!
Great. Double check you have saved detection_engine.py before moving on.
ⓧ I'd like to double check the full code
from collections import defaultdict
from datetime import datetime, timedelta
class DetectionEngine:
def __init__(self):
self.events = []
self.alerts = []
self.alerted_ips = defaultdict(dict)
def ingest(self, event):
self.events.append(event)
cutoff = datetime.now() - timedelta(seconds=300)
self.events = [
e for e in self.events
if datetime.fromisoformat(e["timestamp"]) > cutoff
]
new_alerts = self._run_rules(event)
self.alerts.extend(new_alerts)
return new_alerts
def _run_rules(self, event):
alerts = []
alert = self._check_brute_force(event)
if alert:
alerts.append(alert)
alert = self._check_port_scan(event)
if alert:
alerts.append(alert)
alert = self._check_privilege_escalation(event)
if alert:
alerts.append(alert)
return alerts
def _check_brute_force(self, event):
if event["event_type"] != "ssh_failed_login":
return None
ip = event["source_ip"]
if "brute_force" in self.alerted_ips[ip]:
elapsed = datetime.now() - self.alerted_ips[ip]["brute_force"]
if elapsed < timedelta(seconds=60):
return None
cutoff = datetime.now() - timedelta(seconds=60)
failed = [
e for e in self.events
if e["event_type"] == "ssh_failed_login"
and e["source_ip"] == ip
and datetime.fromisoformat(e["timestamp"]) > cutoff
]
if len(failed) >= 5:
self.alerted_ips[ip]["brute_force"] = datetime.now()
return {
"timestamp": datetime.now().isoformat(),
"severity": "HIGH",
"rule": "SSH Brute Force",
"source_ip": ip,
"description": f"{len(failed)} failed SSH logins from {ip} in 60s",
"events": failed,
}
return None
def _check_port_scan(self, event):
if event["event_type"] != "port_scan":
return None
ip = event["source_ip"]
if "port_scan" in self.alerted_ips[ip]:
elapsed = datetime.now() - self.alerted_ips[ip]["port_scan"]
if elapsed < timedelta(seconds=120):
return None
cutoff = datetime.now() - timedelta(seconds=120)
scans = [
e for e in self.events
if e["event_type"] == "port_scan"
and e["source_ip"] == ip
and datetime.fromisoformat(e["timestamp"]) > cutoff
]
if len(scans) >= 8:
self.alerted_ips[ip]["port_scan"] = datetime.now()
return {
"timestamp": datetime.now().isoformat(),
"severity": "MEDIUM",
"rule": "Port Scan",
"source_ip": ip,
"description": f"{len(scans)} port connections from {ip} in 120s",
"events": scans,
}
return None
def _check_privilege_escalation(self, event):
if event["event_type"] != "sudo_command":
return None
ip = event["source_ip"]
if "priv_esc" in self.alerted_ips[ip]:
elapsed = datetime.now() - self.alerted_ips[ip]["priv_esc"]
if elapsed < timedelta(seconds=300):
return None
cutoff = datetime.now() - timedelta(seconds=300)
failed = [
e for e in self.events
if e["event_type"] == "ssh_failed_login"
and e["source_ip"] == ip
and datetime.fromisoformat(e["timestamp"]) > cutoff
]
if len(failed) >= 3:
self.alerted_ips[ip]["priv_esc"] = datetime.now()
return {
"timestamp": datetime.now().isoformat(),
"severity": "CRITICAL",
"rule": "Privilege Escalation",
"source_ip": ip,
"description": f"Sudo command from {ip} after {len(failed)} failed logins",
"events": failed + [event],
}
return NoneTest the detection engine with simulated events
The detection engine and log simulator are separate modules. The final step is to feed simulated attack events through the engine and confirm it generates the correct alerts.
- Test the brute force detection by running this command in your terminal:
python3 -c '
from log_simulator import generate_brute_force_events
from detection_engine import DetectionEngine
engine = DetectionEngine()
for event in generate_brute_force_events():
alerts = engine.ingest(event)
for a in alerts:
print("ALERT:", a["severity"], "-", a["rule"], "from", a["source_ip"])
print(" ", a["description"])
'You should see output like ALERT: HIGH - SSH Brute Force from 203.0.113.42 followed by a description of how many failed logins were detected. The exact IP will vary because the simulator picks randomly.
- Test the privilege escalation detection by running:
python3 -c '
from log_simulator import generate_privilege_escalation_events
from detection_engine import DetectionEngine
engine = DetectionEngine()
for event in generate_privilege_escalation_events():
alerts = engine.ingest(event)
for a in alerts:
print("ALERT:", a["severity"], "-", a["rule"], "from", a["source_ip"])
print(" ", a["description"])
'You should see two alerts fire. First a HIGH severity SSH Brute Force alert (from the failed login burst), then a CRITICAL Privilege Escalation alert when the sudo command arrives after those failures.
Why two alerts from one attack scenario?
The privilege escalation scenario includes 5 failed logins followed by a successful login and a sudo command. The brute force rule fires on the 5th failed login. The privilege escalation rule fires on the sudo command because it sees prior failed logins from the same IP.
This layered detection is intentional. Real SOC teams want to see every suspicious signal, even when they overlap. Each alert gives a different angle on the same attack.
No alerts printed?
Make sure your terminal is in the same directory as both log_simulator.py and detection_engine.py. Python needs both files to be importable.
Confirm your virtual environment is still active. You should see the environment name in parentheses at the start of your terminal prompt.
Help me debug why my detection engine is not producing alerts.
Your detection engine is working. It ingests raw events, applies threshold-based rules, and outputs structured alerts with severity levels. Next up, you will connect these alerts to a local AI model that provides intelligent threat analysis for each one.
Integrate the AI Threat Analyzer
Your detection engine flags suspicious activity, but raw alerts only tell you what happened. A security analyst's real job is interpreting those alerts and recommending a response.
In this step, you will build an AI-powered threat analyzer that sends each detection alert to your local Gemma 3 model. The model acts as a virtual SOC analyst, providing threat assessments, impact descriptions, and recommended response actions for every alert.
In this step, get ready to:
- Build a threat analyzer with a system prompt that establishes the AI as a security analyst.
- Send detection alerts to the local LLM and attach AI-generated analysis to each alert.
- Test the full pipeline from log generation through AI-enriched alerts.
Create the threat analyzer system prompt
In a real Security Operations Center, analysts follow a structured process when reviewing alerts. Your system prompt tells the LLM to behave as a senior analyst and respond in a consistent format every time.
- Create a new file called threat_analyzer.py in your project folder.
- Add the import and system prompt by pasting this code:
from ollama import chat
SYSTEM_PROMPT = """You are a senior security analyst at a Security Operations Center (SOC).
When given a security alert, provide a brief threat analysis with:
1. Threat Assessment: What is happening and how serious is it (1-2 sentences)
2. Potential Impact: What could happen if this is not addressed (1-2 sentences)
3. Recommended Actions: 2-3 specific steps to respond (bullet points)
Keep your response concise and actionable. Do not use markdown headers."""What does this code do?
The from ollama import chat line imports the same chat function you used in your test script earlier.
The SYSTEM_PROMPT constant defines a persona and response format for the model. By specifying the exact structure (threat assessment, impact, actions), you ensure every alert gets a consistent, actionable analysis.
The last line ("Do not use markdown headers") prevents the model from inserting # symbols that would clutter the terminal output.
- Save threat_analyzer.py.
Why a system prompt instead of instructions in the user message?
System prompts set the model's behavior for the entire conversation. User messages describe the specific task. Separating them means you can change which alert you analyze without rewriting the analyst persona every time.
Build the analyze_threat function
This function takes a detection alert, formats its fields into a prompt, sends it to the model, and attaches the response back to the alert object.
- In threat_analyzer.py, add the following function below the SYSTEM_PROMPT:
def analyze_threat(alert):
prompt = f"""Analyze this security alert:
Rule: {alert['rule']}
Severity: {alert['severity']}
Source IP: {alert['source_ip']}
Description: {alert['description']}
Number of triggering events: {len(alert['events'])}
Provide your threat analysis."""
response = chat(
model="gemma3:4b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": prompt},
],
)
alert["ai_analysis"] = response.message.content
return alertWhat does this function do?
The prompt f-string pulls key fields from the alert dictionary: the rule name, severity level, source IP, description, and how many events triggered it. This gives the model all the context it needs to assess the threat.
The chat() call sends two messages: the system prompt (analyst persona) and the user prompt (the alert data). The model responds with its structured analysis.
Finally, alert["ai_analysis"] = response.message.content attaches the AI's text response directly onto the alert dictionary and returns it. This means any code that receives the alert after analysis can read the ai_analysis field immediately.
- Save threat_analyzer.py.
Why does the model take a few seconds to respond?
Gemma 3 runs entirely on your Mac's GPU and CPU. Each response requires the model to generate tokens one at a time, which typically takes 3-10 seconds depending on response length. In the next step, you will add visual feedback so the user knows analysis is in progress.
Getting an import error?
Make sure your virtual environment is still active. You should see (venv) at the start of your terminal prompt. If not, activate it again before running any Python code.
Also confirm Ollama is still running in the background. If you closed the terminal where Ollama was running, the model will not be reachable.
Help me fix my threat_analyzer.py import error
✔️ Awesome, I've got everything!
Great. Double check you have saved threat_analyzer.py with both the SYSTEM_PROMPT constant and the analyze_threat function.
ⓧ I'd like to double check the full code
from ollama import chat
SYSTEM_PROMPT = """You are a senior security analyst at a Security Operations Center (SOC).
When given a security alert, provide a brief threat analysis with:
1. Threat Assessment: What is happening and how serious is it (1-2 sentences)
2. Potential Impact: What could happen if this is not addressed (1-2 sentences)
3. Recommended Actions: 2-3 specific steps to respond (bullet points)
Keep your response concise and actionable. Do not use markdown headers."""
def analyze_threat(alert):
prompt = f"""Analyze this security alert:
Rule: {alert['rule']}
Severity: {alert['severity']}
Source IP: {alert['source_ip']}
Description: {alert['description']}
Number of triggering events: {len(alert['events'])}
Provide your threat analysis."""
response = chat(
model="gemma3:4b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": prompt},
],
)
alert["ai_analysis"] = response.message.content
return alertTest the full pipeline
Time to connect all three modules together. You will generate attack events, run them through the detection engine, and send the resulting alert to your AI analyzer.
- Run the full pipeline test by running this command in your terminal:
python3 -c "
from log_simulator import generate_brute_force_events
from detection_engine import DetectionEngine
from threat_analyzer import analyze_threat
engine = DetectionEngine()
for event in generate_brute_force_events():
alerts = engine.ingest(event)
for alert in alerts:
result = analyze_threat(alert)
print(result['ai_analysis'])
"What does this test do?
This inline script imports all three of your modules and chains them together. It generates a burst of SSH brute force events, feeds them into the detection engine, and when the engine produces an alert (after 5+ failed logins), it sends that alert to your AI threat analyzer.
The model will take a few seconds to respond. When it finishes, you will see a structured analysis printed in your terminal with a threat assessment, potential impact, and recommended actions.
- Wait 5-10 seconds for the model to generate its response.
- You should see a multi-line AI-generated analysis in your terminal that includes a threat assessment, potential impact description, and 2-3 recommended response actions.
Not seeing any output?
If the command hangs for more than 30 seconds, Ollama may not be running. Open a new terminal tab and run ollama list to check. If you see gemma3:4b listed, the model is available and the delay is just generation time.
If you see an error about the module not being found, make sure you are running the command from inside your project folder where all three .py files live.
Help me debug my pipeline test
Your AI threat analyzer is working. Every alert that passes through it now carries intelligent context that a human analyst would normally write. Next up, you will build a Rich terminal dashboard that displays these AI-enriched alerts in real time with color-coded severity levels.
Build the Terminal Dashboard
Your SIEM pipeline now generates events, detects threats, and enriches alerts with AI analysis. The final piece is a way to actually see all of this working together in real time.
In this step, you'll build a professional terminal dashboard using Rich that ties your entire pipeline together. The dashboard will display color-coded alerts as they appear and show the full AI threat intelligence for each one.
In this step, get ready to:
- Create a Rich-based terminal dashboard with color-coded severity levels.
- Build an alerts table showing the most recent threats with AI summaries.
- Wire a main event loop that runs the full SIEM pipeline and displays results live.
Create the dashboard file with imports and color mapping
The dashboard needs to pull in all three modules you've built so far plus Rich's display components. It also needs a helper to map severity levels to terminal colors.
- Create a new file called dashboard.py in your project folder.
- Add the following imports, console instance, and severity color helper:
from rich.console import Console
from rich.table import Table
from rich.panel import Panel
from log_simulator import event_stream
from detection_engine import DetectionEngine
from threat_analyzer import analyze_threat
console = Console()
def severity_color(severity):
colors = {
"CRITICAL": "bold red",
"HIGH": "yellow",
"MEDIUM": "cyan",
"LOW": "green",
}
return colors.get(severity, "white")What does this code do?
The first three imports bring in Rich's display components. Console handles all terminal output. Table creates formatted tables. Panel wraps content in a bordered box.
The next three imports pull in your own modules so the dashboard can generate events, run detection, and analyze threats all in one place.
The severity_color function maps each severity string to a Rich color style. CRITICAL alerts render in bold red, while LOW alerts appear in green. If an unknown severity appears, it falls back to white.
- Save dashboard.py.
- You won't see output yet. The next chunk builds the table that displays your alerts.
Build the alerts table function
Real SOC dashboards prioritize scanability. An analyst needs to see severity, source, and context at a glance without reading paragraphs of text.
- In dashboard.py, add the following function below severity_color:
def build_alerts_table(alerts):
table = Table(title="SIEM Alerts", show_lines=True)
table.add_column("Time", style="dim", width=10)
table.add_column("Severity", width=10)
table.add_column("Rule", width=22)
table.add_column("Source IP", width=16)
table.add_column("AI Summary", width=50)
for alert in alerts[-10:]:
time_str = alert["timestamp"][11:19]
severity = alert["severity"]
summary = alert.get("ai_analysis", "Analyzing...")[:80]
table.add_row(
time_str,
f"[{severity_color(severity)}]{severity}[/]",
alert["rule"],
alert["source_ip"],
summary,
)
return tableWhat does this code do?
- Creates a Rich Table with five columns: Time, Severity, Rule, Source IP, and AI Summary. Each column has a fixed width to keep the layout stable.
- The alerts[-10:] slice means only the 10 most recent alerts display. This keeps the table manageable as alerts accumulate.
- Rich markup like [{severity_color(severity)}]{severity}[/] applies color inline. CRITICAL shows bold red text, HIGH shows yellow, and so on.
- The AI summary is truncated to 80 characters with [:80] so it fits in the table without wrapping.
- Save dashboard.py.
- You won't see output yet. The main loop in the next chunk brings everything to life.
Wire the main event loop with the detail panel
The main function ties together every component you've built. It starts the event stream, feeds events into the detection engine, sends flagged alerts to the AI analyzer, and prints the results in a formatted dashboard.
- In dashboard.py, add the following main function below build_alerts_table:
def main():
console.print(
Panel("AI-Powered SIEM Tool - Local Threat Intelligence", style="bold blue")
)
console.print("[dim]Press Ctrl+C to stop monitoring[/dim]\n")
engine = DetectionEngine()
analyzed_alerts = []
try:
for event in event_stream():
new_alerts = engine.ingest(event)
for alert in new_alerts:
console.print(f"[bold yellow]New alert detected:[/] {alert['rule']}")
console.print("[dim]Running AI analysis...[/dim]")
analyzed = analyze_threat(alert)
analyzed_alerts.append(analyzed)
console.print()
console.print(build_alerts_table(analyzed_alerts))
console.print()
console.print(
Panel(
analyzed["ai_analysis"],
title=f"[bold]{analyzed['rule']}[/bold] - {analyzed['source_ip']}",
border_style=severity_color(analyzed["severity"]),
)
)
console.print()
except KeyboardInterrupt:
console.print("\n[bold green]Monitoring stopped.[/bold green]")
console.print(f"Total alerts generated: {len(analyzed_alerts)}")
if __name__ == "__main__":
main()What does this code do?
- The opening Panel prints a bold blue header banner identifying the tool. The dim text tells the user how to stop it.
- The try/except KeyboardInterrupt pattern lets the user press Ctrl+C to gracefully stop monitoring instead of crashing with a traceback.
- For each new alert, the loop prints the alerts table (showing all alerts so far) and a detail Panel containing the full AI analysis. The Panel's border color matches the alert severity.
- The detail Panel shows the complete threat assessment, potential impact, and recommended actions from the AI. This gives the analyst full context without needing to expand anything.
- Save dashboard.py.
- Run the dashboard by entering this command in your terminal:
python dashboard.py- You should see a blue "AI-Powered SIEM Tool" banner appear, followed by alerts popping up every few seconds. Each alert shows a color-coded severity, the rule that fired, and an AI-generated summary in the table.
- Below the table, a bordered panel displays the full AI threat analysis for the most recent alert.
- Press Ctrl+C to stop the dashboard when you're done watching.
Why a terminal dashboard?
Many real SOC tools present data in the terminal. A terminal dashboard avoids the overhead of building a web frontend and keeps you focused on the security logic. Rich gives you tables, color, and panels with zero HTML or CSS.
Not seeing any alerts?
- Alerts appear only when attack scenarios trigger. Wait 10-15 seconds. The simulator injects attacks at a 30% rate, so normal events appear first.
- If you see an import error, make sure all four files (log_simulator.py, detection_engine.py, threat_analyzer.py, dashboard.py) are in the same folder.
- If the AI analysis hangs, confirm Ollama is still running. Open a new terminal tab and run ollama list to verify gemma3:4b is available.
Help me debug my SIEM dashboard - it's not showing alerts or the AI analysis is not working.
✔️ Awesome, I've got everything!
Great. Double check you've saved your file and that the dashboard is displaying alerts with color-coded severity and AI analysis panels.
ⓧ I'd like to double check the full code
Here is the complete dashboard.py file:
from rich.console import Console
from rich.table import Table
from rich.panel import Panel
from log_simulator import event_stream
from detection_engine import DetectionEngine
from threat_analyzer import analyze_threat
console = Console()
def severity_color(severity):
colors = {
"CRITICAL": "bold red",
"HIGH": "yellow",
"MEDIUM": "cyan",
"LOW": "green",
}
return colors.get(severity, "white")
def build_alerts_table(alerts):
table = Table(title="SIEM Alerts", show_lines=True)
table.add_column("Time", style="dim", width=10)
table.add_column("Severity", width=10)
table.add_column("Rule", width=22)
table.add_column("Source IP", width=16)
table.add_column("AI Summary", width=50)
for alert in alerts[-10:]:
time_str = alert["timestamp"][11:19]
severity = alert["severity"]
summary = alert.get("ai_analysis", "Analyzing...")[:80]
table.add_row(
time_str,
f"[{severity_color(severity)}]{severity}[/]",
alert["rule"],
alert["source_ip"],
summary,
)
return table
def main():
console.print(
Panel("AI-Powered SIEM Tool - Local Threat Intelligence", style="bold blue")
)
console.print("[dim]Press Ctrl+C to stop monitoring[/dim]\n")
engine = DetectionEngine()
analyzed_alerts = []
try:
for event in event_stream():
new_alerts = engine.ingest(event)
for alert in new_alerts:
console.print(f"[bold yellow]New alert detected:[/] {alert['rule']}")
console.print("[dim]Running AI analysis...[/dim]")
analyzed = analyze_threat(alert)
analyzed_alerts.append(analyzed)
console.print()
console.print(build_alerts_table(analyzed_alerts))
console.print()
console.print(
Panel(
analyzed["ai_analysis"],
title=f"[bold]{analyzed['rule']}[/bold] - {analyzed['source_ip']}",
border_style=severity_color(analyzed["severity"]),
)
)
console.print()
except KeyboardInterrupt:
console.print("\n[bold green]Monitoring stopped.[/bold green]")
console.print(f"Total alerts generated: {len(analyzed_alerts)}")
if __name__ == "__main__":
main()Your SIEM tool is fully operational. You now have a complete pipeline from simulated log events through rule-based detection to AI-powered threat intelligence, all displayed in a professional terminal dashboard. Next up, a secret mission to add industry-standard threat classification.
Secret mission
Add MITRE ATT&CK Mapping
Your SIEM tool detects threats, but real SOC analysts need to classify those threats using an industry-standard framework. In this secret mission, you'll integrate MITRE ATT&CK technique mapping so every alert gets tagged with its official tactic and technique ID.
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely on your local machine, so there are no ongoing cloud costs or subscriptions to worry about. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.
Resources you used:
- Ollama application with the gemma3:4b model.
- Python virtual environment in your project folder.
- Project files: log_simulator.py, detection_engine.py, threat_analyzer.py, dashboard.py, and test_ollama.py.
Keep everything running
No action needed. Choose this if you're still actively building or plan to extend your SIEM tool with new detection rules or log sources.
- Ollama will continue running in the background and unload models from memory after 5 minutes of inactivity.
- Your Python virtual environment and project files remain on your Desktop ready to use.
Pause - I'll come back to this later
Shut down running processes to free up memory while keeping all your files intact for later.
- Stop Ollama by clicking the Ollama icon in your menu bar and selecting Quit Ollama.
- Deactivate your Python virtual environment by running:
deactivate- Your project folder and all files remain on your Desktop. You can reactivate the virtual environment and restart Ollama when you return.
Delete - I don't want to use this again
Remove all project resources and start fresh. This frees disk space used by the model and project files.
- Stop Ollama by clicking the Ollama icon in your menu bar and selecting Quit Ollama.
- Remove the gemma3:4b model by running:
ollama rm gemma3:4b- Remove the Ollama application entirely by running:
sudo rm -rf /Applications/Ollama.app
sudo rm /usr/local/bin/ollama
rm -rf ~/.ollamaWhat does this remove?
The first command removes the Ollama application. The second removes the CLI symlink. The third deletes all downloaded models, configuration, and cache data stored locally.
- Delete your project folder and virtual environment by running:
cd ~/Desktop
rm -rf ai-siemAdjust the folder name
Replace ai-siem with whatever you named your project folder on the Desktop.
Nice Work!
Nice Work!
Nice work! You just built a fully local, AI-powered SIEM tool that detects threats and generates intelligent analysis without sending a single byte of data to the cloud.
You've learned how to:
- Build a complete SIEM pipeline from scratch, including log ingestion, event parsing, threshold-based detection rules, and structured alerting.
- Integrate a locally-running LLM (Gemma 3 via Ollama) into a Python security application for intelligent threat analysis with zero cloud costs or API keys.
- Design a professional Rich terminal dashboard with color-coded severity levels and real-time alert monitoring.
- Secret Mission: Added MITRE ATT&CK framework mapping to connect detected threats to industry-standard technique classifications.
Ready to quiz yourself?