# How to Build a Minimal SIEM with Python, SQLite and Telegram Alerts > Source: > Published: 2026-08-14 10:05:10+00:00 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. ``` python 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. ``` python 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. 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: ``` python 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. 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](https://dev.to/botfather) and your personal chat ID from @userinfobot — both free. ``` python 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. ``` python 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](https://datasette.io); AbuseIPDB enrichment on flagged IPs. For a structured checklist of what to monitor and in what priority order, [our free security hardening checklists](https://ayinedjimi-consultants.fr/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.*