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

A developer built a minimal Security Information and Event Management (SIEM) system using Python, SQLite, and Telegram alerts, requiring less than 400 lines of code and no external infrastructure. The system tails log files, stores events in SQLite with WAL mode for concurrent access, and runs detection rules to alert on threats like SSH brute-force attempts.

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<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: 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.