How to Build a Minimal SIEM with Python, SQLite and Telegram Alerts

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. What You Actually Need from a SIEM A Security Information and Event Management system does three things: Collect logs from multiple sources Correlate events against detection rules Alert when a rule fires 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. Setting Up the Storage Layer 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. Ingesting Logs 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" # Matches: "Failed password for root from 1.2.3.4" SSH_FAILED = re.compile( r"Failed password for (?:\w+ )?(?P\S+) from (?P[\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. Writing Detection Rules 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): # Deduplicate: skip if same rule+payload fired in the last 10 minutes 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. Sending Telegram Alerts 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. Putting It All Together 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 The Takeaway 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.