initial commit

.gitignore

@@ -0,0 +1,8 @@
+bin
+lib
+lib64
+include
+__pycache__
+*.cfg
+*.toml
+*.log

README.md

@@ -0,0 +1,153 @@
+# 🚨 sshd-bruteforce-detector
+
+A lightweight, real-time SSH brute-force detection tool for Linux systems using `systemd-journal`.
+
+It monitors `sshd` logs, groups failed login attempts by source IP, and alerts you when suspicious activity crosses a defined threshold.
+
+Because manually reading logs is fun… until it isn’t.
+
+---
+
+## ✨ Features
+
+* 📡 Real-time monitoring via `systemd.journal`
+* 🔍 Detects repeated failed SSH authentication attempts
+* 🌐 Groups events per remote host (IP)
+* ⏱ Configurable time window (e.g. `5m`, `1h`)
+* 🚦 Configurable max attempts threshold
+* 🧾 Supports config file (`.toml`) and CLI args
+* 🪶 Lightweight and dependency-minimal
+
+---
+
+## ⚙️ How It Works
+
+1. Listens to `sshd.service` logs from systemd journal
+2. Filters **failed authentication messages**
+3. Groups attempts by `rhost` (IP address)
+4. Keeps only recent attempts within a time window
+5. Triggers alert if attempts ≥ threshold
+
+---
+
+## 🚀 Installation
+
+```bash
+git clone https://github.com/yourusername/sshd-bruteforce-detector.git
+cd sshd-bruteforce-detector
+pip install -r requirements.txt
+```
+
+> Requires Linux with `systemd` and Python 3.11+ (for `tomllib`).
+
+---
+
+## ▶️ Usage
+
+### Basic run
+
+```bash
+python main.py
+```
+
+### With arguments
+
+```bash
+python main.py -w 5m -m 10
+```
+
+### With config file
+
+```bash
+python main.py -c config.toml
+```
+
+---
+
+## 🧩 Configuration
+
+You can configure via CLI or TOML file.
+
+### Example `config.toml`
+
+```toml
+window = "5m"
+max_attempts = 5
+```
+
+### Options
+
+| Option                 | Description                                       | Default |
+| ---------------------- | ------------------------------------------------- | ------- |
+| `--window`, `-w`       | Time window to group attempts (`1h`, `5m`, `30s`) | `5m`    |
+| `--max-attempts`, `-m` | Max failed attempts before alert                  | `5`     |
+| `--config`, `-c`       | Path to TOML config file                          | —       |
+
+CLI args override config file values.
+
+---
+
+## 📊 Example Output
+
+```text
+INFO  Authentication failure from host 192.168.1.10 for user root on sshd
+INFO  Authentication failure from host 192.168.1.10 for user admin on sshd
+CRITICAL 5 authentication failure events from 192.168.1.10 on sshd in 42 seconds.
+```
+
+---
+
+## 🧠 Detection Logic
+
+The detection is intentionally simple:
+
+```python
+len(recent_attempts) >= max_attempts
+```
+
+No AI, no magic. Just effective signal over noise.
+
+---
+
+## 🛑 Stopping the Program
+
+Handles signals gracefully:
+
+* `Ctrl + C` → exits with code `130`
+* `SIGTERM` → clean exit
+
+---
+
+## 📁 Project Structure
+
+```
+.
+├── main.py            # Entry point
+├── detector.py        # Detection logic
+├── config.py          # CLI + config handling
+├── message_parse.py   # Log parsing
+├── utils.py           # Helpers (window parsing, filtering)
+```
+
+---
+
+## ⚠️ Limitations
+
+* Only works on systems using `systemd`
+* Depends on log format consistency (`authentication failure;`)
+* No automatic banning (yet 👀)
+
+---
+
+## 💡 Future Ideas
+
+* Auto-ban IPs via `iptables` / `nftables`
+* Export metrics (Prometheus)
+* Web dashboard
+* Alert integrations (Slack, email, etc.)
+
+---
+
+## 📜 License
+
+MIT — do whatever you want, just don’t blame me if your server catches fire.

config.py

@@ -0,0 +1,66 @@
+import argparse
+import sys
+import tomllib
+import utils
+import logging
+
+
+class Config:
+    def __init__(self):
+        self.logger = logging.getLogger("sshd-bruteforce-detector")
+        parser = argparse.ArgumentParser(
+            prog="sshd-bruteforce-detector",
+            description="detects SSH bruteforce attacks by reading sshd's logs from systemd journal",
+        )
+
+        parser.add_argument(
+            "-c", "--config", nargs="?", help="provide a path for the config file"
+        )
+        parser.add_argument(
+            "-m",
+            "--max-attempts",
+            nargs="?",
+            help="Maximum number of failed login attempts allowed within the time window before triggering an alert. (default: 5)",
+        )
+        parser.add_argument(
+            "-w",
+            "--window",
+            nargs="?",
+            help="Time window used to group failed login attempts for detection (default: 5m)",
+        )
+
+        self.args = parser.parse_args()
+        self.config = {}
+        if getattr(self.args, "config", None):
+            self.__load_config(self.args.config)
+
+    def __load_config(self, path: str):
+        try:
+            with open(path, "rb") as f:
+                self.config = tomllib.load(f)
+        except FileNotFoundError:
+            self.logger.info("Config file not found. Stopping manually...")
+            sys.exit(1)
+        except tomllib.TOMLDecodeError:
+            self.logger.info(
+                "Config file not in valid TOML format. Stopping manually..."
+            )
+            sys.exit(1)
+
+    def __get_config_option(self, option):
+        opt = getattr(self.args, option, None)
+        opt_config = self.config.get(option, None)
+        if opt is not None:
+            return opt
+        elif opt_config is not None:
+            return opt_config
+        else:
+            return None
+
+    def get_window(self):
+        window = self.__get_config_option("window")
+        return utils.parse_window(window) if window is not None else 300
+
+    def get_max_attempts(self):
+        max_attempts = self.__get_config_option("max_attempts")
+        return int(max_attempts) if max_attempts is not None else 5

detector.py

@@ -0,0 +1,5 @@
+from message_parse import Message
+
+
+def detect_suspect_messages(recent_messages: list[Message], max_events: int):
+    return len(recent_messages) >= max_events

logs.py

@@ -0,0 +1,23 @@
+import logging
+import sys
+
+
+def setup_logging():
+    logger = logging.getLogger("sshd-bruteforce-detector")
+    logger.setLevel(logging.INFO)
+
+    if logger.handlers:
+        return logger
+
+    fmt = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s")
+
+    console = logging.StreamHandler(sys.stdout)
+    console.setFormatter(fmt)
+
+    file = logging.FileHandler("sshd-bruteforce-detector.log")
+    file.setFormatter(fmt)
+
+    logger.addHandler(console)
+    logger.addHandler(file)
+
+    return logger

main.py

@@ -0,0 +1,89 @@
+import logging
+from systemd import journal
+from datetime import timedelta
+from config import Config
+from utils import filter_message_list
+from message_parse import Message
+from sys import exit
+import detector
+import signal
+import time
+import logs
+
+
+running = True
+logger = logging.getLogger("sshd-bruteforce-detector")
+
+
+def handle_exit_signal(signum, _):
+    global running
+    global logger
+    running = False
+    logger.info(f"{signal.Signals(signum).name} received. Stopping manually...")
+    if signum == signal.SIGINT:
+        exit(130)
+    else:
+        exit(0)
+
+
+def main():
+    logs.setup_logging()
+
+    global logger
+
+    config = Config()
+
+    logger = logging.getLogger("sshd-bruteforce-detector")
+    window = config.get_window()
+    max_attempts = config.get_max_attempts()
+
+    j = journal.Reader()
+
+    j.add_match(_SYSTEMD_UNIT="sshd.service")
+    j.seek_tail()
+    j.get_previous()
+
+    messages: dict[str, list] = {}
+
+    message_detected = False
+    logger.info(
+        f"Starting sshd-bruteforce-detector with configuration: window={window} max_attempts={max_attempts}"
+    )
+
+    global running
+
+    while running:
+        for entry in j:
+            message_str = entry["MESSAGE"]
+            if Message.is_failed_auth_message(message_str):
+                message = Message(message_str, entry["__REALTIME_TIMESTAMP"])
+                if not messages.get(str(message.rhost)):
+                    messages[str(message.rhost)] = []
+                messages[str(message.rhost)].append(message)
+                logger.info(
+                    f"Authentication failure from host {message.rhost} for user {message.user} on sshd"
+                )
+                message_detected = True
+
+        new_messages = messages.copy()
+        for ip, message_list in messages.items():
+            new_messages[ip] = filter_message_list(
+                message_list, timedelta(seconds=window)
+            )
+            if (
+                detector.detect_suspect_messages(new_messages[ip], max_attempts)
+                and message_detected
+            ):
+                alert_window = new_messages[ip][-1].date - new_messages[ip][0].date
+                logger.critical(
+                    f"{len(new_messages[ip])} authentication failure events from {ip} on sshd in {alert_window.seconds} seconds."
+                )
+                message_detected = False
+
+        time.sleep(0.3)
+
+
+if __name__ == "__main__":
+    signal.signal(signal.SIGTERM, handle_exit_signal)
+    signal.signal(signal.SIGINT, handle_exit_signal)
+    main()

message_parse.py

@@ -0,0 +1,16 @@
+from datetime import datetime
+import re
+
+
+class Message:
+    def __init__(self, message: str, date: datetime):
+        self.date = date
+        self.user = re.search(r"user=([^\s]+)", message)
+        self.rhost = re.search(r"rhost=([^\s]+)", message)
+
+        self.user = self.user.group(1) if self.user else None
+        self.rhost = self.rhost.group(1) if self.rhost else None
+
+    @staticmethod
+    def is_failed_auth_message(message: str):
+        return "authentication failure;" in message

requirements.txt

@@ -0,0 +1 @@
+systemd-python

utils.py

@@ -0,0 +1,24 @@
+from datetime import datetime, timedelta, timezone
+from message_parse import Message
+import re
+
+
+def filter_message_list(message_list: list[Message], window: timedelta):
+    now = datetime.now(timezone.utc)
+    return [m for m in message_list if now - m.date <= window]
+
+
+def parse_window(s: str) -> int:
+    pattern = r"(?:(\d+)h)?(?:(\d+)m)?(?:(\d+)s)?$"
+    match = re.fullmatch(pattern, s.strip())
+
+    if not match:
+        raise ValueError(f"Invalid duration: {s}")
+
+    h, m, sec = match.groups()
+
+    return (
+        (int(h) * 3600 if h else 0)
+        + (int(m) * 60 if m else 0)
+        + (int(sec) if sec else 0)
+    )