Most teams can't afford Splunk, and Elastic SIEM takes real time to tune. Yet you still need to know when someone is brute-forcing your SSH, when a suspicious process spawns at 3 AM, or when a web server starts returning a flood of 500s. A minimal SIEM built with Python, SQLite, and Telegram can cover these cases with less than 400 lines of code and zero additional infrastructure.
This article walks through building a working prototype you can deploy and extend immediately.
A Security Information and Event Management system does three things:
That's it. Everything else — dashboards, threat intelligence enrichment, ML anomaly detection — is additive. The goal here is a foundation that runs on a single VM, a Raspberry Pi, or a $6/month VPS without any external service dependencies.
SQLite is the right tool for a minimal SIEM. It's file-based, needs no server process, and handles tens of millions of rows comfortably. We use two tables: events
for raw log lines and detections
for fired rules.
import sqlite3
from pathlib import Path
DB_PATH = Path("/var/db/siem.db")
def init_db(db_path: Path = DB_PATH) -> sqlite3.Connection:
conn = sqlite3.connect(db_path)
conn.execute("PRAGMA journal_mode=WAL") # concurrent reads won't block writes
conn.execute("PRAGMA synchronous=NORMAL")
conn.executescript("""
CREATE TABLE IF NOT EXISTS events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ts INTEGER NOT NULL, -- unix epoch ms
source TEXT NOT NULL, -- e.g. sshd, nginx, auditd
host TEXT NOT NULL,
raw TEXT NOT NULL, -- original log line
severity TEXT DEFAULT 'info'
);
CREATE TABLE IF NOT EXISTS detections (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ts INTEGER NOT NULL,
rule_id TEXT NOT NULL,
event_ids TEXT NOT NULL, -- JSON array of matching event IDs
summary TEXT NOT NULL,
notified INTEGER DEFAULT 0
);
CREATE INDEX IF NOT EXISTS idx_events_ts ON events(ts);
CREATE INDEX IF NOT EXISTS idx_events_source ON events(source);
""")
conn.commit()
return conn
One critical detail: WAL
mode allows a reader and a writer to work simultaneously. Without it, your log ingestion locks the database and the detection loop stalls.
We tail /var/log/auth.log
using subprocess
. The same pattern works for nginx, auditd, or any file-based log source.
import subprocess
import time
import re
from dataclasses import dataclass
@dataclass
class LogEvent:
ts: int
source: str
host: str
raw: str
severity: str = "info"
SSH_FAILED = re.compile(
r"Failed password for (?:\w+ )?(?P<user>\S+) from (?P<ip>[\d\.]+)"
)
def tail_auth_log(path: str = "/var/log/auth.log"):
proc = subprocess.Popen(
["tail", "-F", "-n", "0", path],
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
text=True,
)
try:
for line in proc.stdout:
line = line.rstrip()
ts = int(time.time() * 1000)
severity = "warning" if SSH_FAILED.search(line) else "info"
yield LogEvent(ts=ts, source="sshd", host="localhost",
raw=line, severity=severity)
finally:
proc.terminate()
def ingest(conn, event: LogEvent):
conn.execute(
"INSERT INTO events (ts, source, host, raw, severity) VALUES (?,?,?,?,?)",
(event.ts, event.source, event.host, event.raw, event.severity),
)
conn.commit()
For multi-host environments, ship logs via syslog-ng to a central host and swap the tail
for a UDP listener. The schema doesn't change.
The detection loop queries recent events and matches them against rule functions. Here's a brute-force SSH rule that fires when the same IP produces more than 5 failed logins within 60 seconds:
import json
def detect_ssh_brute_force(conn, window_sec: int = 60, threshold: int = 5):
now_ms = int(time.time() * 1000)
since_ms = now_ms - window_sec * 1000
rows = conn.execute(
"SELECT id, raw FROM events "
"WHERE source = 'sshd' AND severity = 'warning' AND ts >= ?",
(since_ms,),
).fetchall()
ip_events: dict[str, list[int]] = {}
for row_id, raw in rows:
m = SSH_FAILED.search(raw)
if m:
ip = m.group("ip")
ip_events.setdefault(ip, []).append(row_id)
results = []
for ip, event_ids in ip_events.items():
if len(event_ids) >= threshold:
summary = (
f"SSH brute force from {ip}: "
f"{len(event_ids)} attempts in {window_sec}s"
)
results.append(("ssh_brute_force", json.dumps(event_ids), summary))
return results
def run_detections(conn):
for rule_fn in [detect_ssh_brute_force]:
for rule_id, event_ids, summary in rule_fn(conn):
existing = conn.execute(
"SELECT 1 FROM detections "
"WHERE rule_id=? AND event_ids=? AND ts >= ?",
(rule_id, event_ids, int(time.time() * 1000) - 600_000),
).fetchone()
if not existing:
conn.execute(
"INSERT INTO detections (ts, rule_id, event_ids, summary) "
"VALUES (?,?,?,?)",
(int(time.time() * 1000), rule_id, event_ids, summary),
)
conn.commit()
The deduplication check is not optional. Without it, an active brute-force campaign fires a fresh alert every 30 seconds until the detection window moves past it.
Telegram's Bot API is the lowest-friction notification channel available: no SMTP setup, no PagerDuty account, no webhook hosting required. Get a bot token from @botfather and your personal chat ID from @userinfobot — both free.
import urllib.request
TELEGRAM_TOKEN = "your-bot-token"
TELEGRAM_CHAT_ID = "your-chat-id"
def send_telegram(message: str):
url = f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage"
payload = json.dumps({
"chat_id": TELEGRAM_CHAT_ID,
"text": message,
"parse_mode": "Markdown",
}).encode()
req = urllib.request.Request(
url, data=payload,
headers={"Content-Type": "application/json"},
)
with urllib.request.urlopen(req, timeout=10) as resp:
return json.loads(resp.read())
def flush_alerts(conn):
rows = conn.execute(
"SELECT id, summary FROM detections WHERE notified = 0"
).fetchall()
for det_id, summary in rows:
try:
send_telegram(f"🚨 *SIEM Alert*\n{summary}")
conn.execute(
"UPDATE detections SET notified = 1 WHERE id = ?", (det_id,)
)
conn.commit()
except Exception as e:
print(f"[alert] failed to notify: {e}")
Using urllib.request
keeps the dependency list empty. For higher-volume deployments, swap this for httpx
with async support.
Run ingestion in a background thread; detection and alerting run on the main thread every 30 seconds.
import threading
def main():
conn = init_db()
def ingest_loop():
for event in tail_auth_log():
ingest(conn, event)
t = threading.Thread(target=ingest_loop, daemon=True)
t.start()
while True:
run_detections(conn)
flush_alerts(conn)
time.sleep(30)
if __name__ == "__main__":
main()
Drop it under systemd for automatic restart on failure:
[Unit]
Description=Minimal SIEM
After=network.target
[Service]
ExecStart=/usr/bin/python3 /opt/siem/main.py
Restart=always
RestartSec=5
[Install]
WantedBy=multi-user.target
This stack — Python + SQLite + Telegram — covers 80% of what small teams need from a SIEM at essentially zero infrastructure cost. The rule system is deliberately simple: add a new function that returns (rule_id, event_ids, summary)
tuples and register it in run_detections
. No framework, no YAML configuration, no agent to maintain.
Reasonable next steps: rules for failed sudo
attempts, unexpected cron jobs, and outbound connections via auditd
; a read-only UI with Datasette; AbuseIPDB enrichment on flagged IPs. For a structured checklist of what to monitor and in what priority order, our free security hardening checklists map detection goals to MITRE ATT&CK tactics.
I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.