Building a Real-Time Anomaly Detection Engine for Cloud Storage Security

Real-Time Anomaly Detection Engine for a Cloud Storage Platform

Timilehin Obalereko developed a Python-based security daemon that monitors Nginx logs to identify and block attackers automatically. The system utilizes a statistical Z-score threshold of 3.0 to achieve 99.87% detection accuracy for abnormal traffic spikes.

Why This Matters

Static security thresholds often fail in dynamic cloud environments where traffic patterns fluctuate by time of day; a fixed limit causes false positives during peaks and misses attacks during lulls. This system addresses technical reality by implementing an adaptive baseline with a ‘spike guard’ to prevent malicious traffic from corrupting the statistical model, ensuring the firewall remains effective even under sustained assault.

Key Insights

• Statistical Z-score detection: Using (current_rate - mean) / stddev to identify anomalies that are 3.0 standard deviations above the mean, representing a 99.87% probability of deviation.
• Adaptive Baseline with Spike Guard: The engine recalculates mean and standard deviation every 60 seconds but discards any traffic over 10x the current mean to prevent attackers from skewing the ‘normal’ definition.
• Kernel-Level Blocking with iptables: By using ‘iptables -I INPUT 1’, the system drops malicious packets at the network layer before they reach application code, reducing CPU overhead during attacks.
• Sliding Window Architecture: Implementing Python’s collections.deque for a 60-second sliding window ensures real-time rate calculation that catches short bursts missed by per-minute counters.
• Error Surge Detection: The system automatically tightens Z-score thresholds from 3.0 to 2.0 if an IP generates excessive 404 or 401 status codes, indicating vulnerability probing.

Working Examples

Implementation of a sliding window using collections.deque to calculate requests per second.

from collections import deque, defaultdict
import time

class SlidingWindowDetector:
    def __init__(self, config):
        self.window_seconds = 60
        self.ip_windows = defaultdict(deque)

    def get_ip_rate(self, ip):
        now = time.time()
        dq = self.ip_windows[ip]
        cutoff = now - self.window_seconds
        while dq and dq[0] < cutoff:
            dq.popleft()
        return len(dq) / self.window_seconds

Function to programmatically block malicious IPs at the Linux kernel level.

def ban_ip(self, ip):
    # -I INPUT 1 = insert at top priority
    # -s {ip} = source IP
    # -j DROP = discard packet
    subprocess.run([
        "iptables", "-I", "INPUT", "1",
        "-s", ip, "-j", "DROP"
    ])

Practical Applications

• Use Case: Public cloud storage platforms like Nextcloud use this engine to block brute-force bots hammering Nginx endpoints. Pitfall: Hardcoding thresholds without standard deviation leads to false positives during legitimate traffic surges.
• Use Case: Botnet detection via global window monitoring that spots distributed attacks where no single IP exceeds limits. Pitfall: Failing to implement an auto-unban backoff schedule can permanently lock out legitimate users behind shared corporate NATs.

References:

• https://dev.to/devopstimi/how-i-built-a-real-time-anomaly-detection-engine-for-a-cloud-storage-platform-30o3
• https://github.com/devops-timi/anomaly-detection-engine