Real-Time Auth Log Parser and Automated IP Sanitization Engine

This real-time security automation daemon streams authentication logs to detect brute-force intrusion attempts and dynamically updates host firewall rules. Internally, it processes log file changes from /var/log/auth.log via file tailing, matches failed access sequences using regular expressions, tracks failure frequencies within rolling time windows, and invokes system firewall utilities via subprocess calls to drop suspicious IP addresses.

Deploying this tool on edge hosts, bastion instances, and publicly exposed infrastructure delivers automated protection against SSH password-guessing attacks and automated credential spraying. It functions as an integrated defense layer to reduce system exposure without depending on external log aggregation services.

Unsanitized IP strings read directly from log files can lead to command injection if passed to shell interfaces without validation using Python ipaddress modules. Overly aggressive threshold settings risk accidentally blocking valid administrative IP ranges or facilitating denial-of-service conditions through source IP spoofing.

> System.Execute()

import time, re, subprocess, ipaddress

LOG_FILE = "/var/log/auth.log"
FAIL_THRESHOLD = 5
failed_attempts = {}

def block_ip(ip):
    try:
        ip_obj = ipaddress.ip_address(ip)
        subprocess.run(["nft", "add", "element", "inet", "filter", "blackhole", f"{{ {ip_obj} }}"], check=True)
        print(f"[+] Blocked IP via nftables: {ip_obj}")
    except Exception as e:
        print(f"[-] Blocking failed for {ip}: {e}")

def monitor_log():
    regex = re.compile(r"Failed password for .* from ([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)")
    with open(LOG_FILE, "r") as f:
        f.seek(0, 2)
        while True:
            line = f.readline()
            if not line:
                time.sleep(0.5)
                continue
            match = regex.search(line)
            if match:
                ip = match.group(1)
                failed_attempts[ip] = failed_attempts.get(ip, 0) + 1
                if failed_attempts[ip] >= FAIL_THRESHOLD:
                    block_ip(ip)
                    failed_attempts[ip] = 0

monitor_log()

> Architecture & Security Context

Verified for Valters IT Hub. When deploying this configuration in a production homelab or enterprise environment, ensure that you monitor system resources and implement strict access controls. Administrator privileges are required.

ID: VALTERSIT-A70D005D4D64