Deteccao de anomalias em logs de servidor com Python e estatistica

Detecção de Anomalias em Logs com Python e Estatística: Seu Radar de Ameaças Antes do Apocalipse Digital

PorOlivetto

Às 3:47 da manhã, enquanto você dormia, alguém tentou fazer 4.237 requisições ao seu servidor em menos de 90 segundos. Não foi um bot de SEO. Não foi um crawler amigável. Foi alguém testando suas portas, procurando por um /wp-admin esquecido, um endpoint exposto, uma falha de autenticação que você jurou que ia consertar “depois do almoço” — três meses atrás.

E sabe qual foi a resposta do seu sistema?

Silêncio.

Porque você não tinha um sistema de detecção de anomalias. Tinha só um log crescendo silenciosamente, como um tumor digital, até o dia em que seu servidor simplesmente sumiu da internet. E aí você descobriu que o ataque não começou naquele dia. Começou semanas antes, em pequenas sondagens que pareciam normais demais para alertar qualquer um.

O Problema: Logs São Oceânicos e Você Navega de Bote

A média de um servidor web modesto gera entre 10MB e 500MB de logs por dia. Em produção, isso facilmente passa de gigabytes. E cada linha é uma história: um 200 que deveria ser 404, um 403 suspeito, um spike de latência que durou exatos 47 segundos e sumiu antes de você abrir o terminal.

O padrão atual é reativo: algo quebra, você olha os logs. Mas quando o “algo” é um ataque silencioso ou uma vulnerabilidade sendo explorada em camadas que você nem monitora, reagir é tarde demais.

O que você precisa é de um sistema de detecção de anomalias que funcione em tempo real, sem depender de SaaS caro, sem mandar seus dados para nuvem de terceiro, e — mais importante — sem exigir um PhD em machine learning.

A Solução: Estatística Simples que Funciona Melhor que ML Complicado

Aqui vai uma verdade que ninguém te conta: para 95% dos casos de detecção de anomalias em logs, estatística básica funciona tão bem quanto deep learning — e roda num Raspberry Pi.

O segredo está em monitorar métricas agregadas ao invés de linhas individuais. Não importa se o ataque veio de um IP na Ucrânia ou de uma botnet com milhares de origens. O que importa é o padrão:

  • Taxa de erro abrupta: seu 404 médio é 2%/min. De repente pulou para 40%/min. Anomalia.
  • Latência fora do envelope: 95% das requisições respondem em <200ms. De repente 30% estão em >2s. Anomalia.
  • Volume de tráfego fora do padrão: segunda-feira 14h sempre tem 150 req/min. Hoje tem 1.200. Anomalia.
  • Padrão temporal estranho: seu sistema nunca recebeu requisições às 4h da manhã. Agora recebe. Anomalia.

Construindo o Detector em Python Puro

Vou te mostrar um sistema completo que eu construí depois de tomar um susto de verdade: descobri que um servidor meu estava sendo escaneado há 23 dias sem que eu soubesse. O pior? O log tinha todas as evidências. Eu simplesmente não estava olhando.

Estrutura do Sistema

O detector funciona em três camadas:

  • Coletor: consome logs em tempo real (ou batch) e extrai métricas agregadas
  • Modelo: mantém uma baseline estatística do comportamento normal
  • Detector: compara observações atuais contra a baseline e dispara alertas

Camada 1: O Coletor de Métricas

Em vez de parsear cada linha de log (o que é lento e frágil), extraímos métricas numa janela deslizante. Para logs Nginx/Apache, isso é trivial:

import re
from collections import defaultdict
from datetime import datetime, timedelta

class LogMetricsCollector:
    def __init__(self, window_minutes=5):
        self.window = timedelta(minutes=window_minutes)
        self.requests = []
        self.pattern = re.compile(
            r'(\S+) - - \[(.+?)\] "(\S+) (\S+) \S+" (\d{3}) (\d+) "[^"]*" "([^"]*)"'
        )

    def parse_line(self, line):
        m = self.pattern.match(line)
        if not m:
            return None
        ip, time_str, method, path, status, size, ua = m.groups()
        dt = datetime.strptime(time_str, "%d/%b/%Y:%H:%M:%S %z")
        return {
            "ip": ip, "dt": dt, "method": method,
            "path": path, "status": int(status),
            "size": int(size), "ua": ua
        }

    def add(self, line):
        entry = self.parse_line(line)
        if entry:
            self.requests.append(entry)
        cutoff = datetime.now(entry["dt"].tzinfo) - self.window
        self.requests = [r for r in self.requests if r["dt"] > cutoff]

    def snapshot(self):
        if not self.requests:
            return None
        total = len(self.requests)
        errors = sum(1 for r in self.requests if r["status"] >= 400)
        sizes = [r["size"] for r in self.requests]
        unique_ips = len(set(r["ip"] for r in self.requests))
        return {
            "total": total,
            "error_rate": errors / total,
            "avg_size": sum(sizes) / len(sizes),
            "unique_ips": unique_ips,
            "timestamp": datetime.now().isoformat()
        }

Simples, direto, e processa milhares de linhas por segundo porque não faz I/O nem mantém estado pesado.

Camada 2: A Baseline Estatística

Aqui está onde a mágica acontece. Usamos desvio padrão móvel com média exponencial para manter uma baseline que se adapta a mudanças legítimas (tipo um lançamento que aumenta o tráfego) mas ainda detecta anomalias:

import statistics
from collections import deque

class AdaptiveBaseline:
    def __init__(self, history_size=100, sensitivity=3.0):
        self.history = deque(maxlen=history_size)
        self.sensitivity = sensitivity
        self.ema = None
        self.emvar = None
        self.alpha = 2.0 / (history_size + 1)

    def update(self, value):
        self.history.append(value)
        if self.ema is None:
            self.ema = value
            self.emvar = 0.0
            return
        delta = value - self.ema
        self.ema += self.alpha * delta
        self.emvar = (1 - self.alpha) * (self.emvar + self.alpha * delta ** 2)

    def is_anomaly(self, value):
        if len(self.history) < 10 or self.emvar is None:
            return False
        std = self.emvar ** 0.5
        threshold = self.sensitivity * std
        return abs(value - self.ema) > threshold

    def zscore(self, value):
        if self.emvar is None or self.emvar == 0:
            return 0.0
        return (value - self.ema) / (self.emvar ** 0.5)

A média exponencial (EMA) é perfeita porque:

  • Adapta-se a mudanças graduais (seu site cresceu, a baseline cresce junto)
  • Não precisa manter todo o histórico em memória (só o EMA e a variância)
  • Detecta mudanças abruptas instantaneamente (ataques, falhas)

Camada 3: O Detector e Alertador

Agora juntamos tudo num pipeline que consome logs, mantém baselines para cada métrica, e dispara quando algo sai do normal:

class AnomalyDetector:
    def __init__(self):
        self.collector = LogMetricsCollector(window_minutes=5)
        self.baselines = {
            "error_rate": AdaptiveBaseline(sensitivity=2.5),
            "total": AdaptiveBaseline(sensitivity=3.0),
            "unique_ips": AdaptiveBaseline(sensitivity=2.0),
        }
        self.alert_cooldown = {}

    def process_line(self, line):
        self.collector.add(line)
        snapshot = self.collector.snapshot()
        if not snapshot:
            return []
        alerts = []
        for metric, baseline in self.baselines.items():
            value = snapshot[metric]
            baseline.update(value)
            if baseline.is_anomaly(value):
                z = baseline.zscore(value)
                alerts.append({
                    "metric": metric,
                    "value": round(value, 4),
                    "zscore": round(z, 2),
                    "severity": "critical" if abs(z) > 5 else "warning"
                })
        return alerts

Cada métrica tem sua própria sensibilidade porque algumas coisas variam mais que outras. Taxa de erro nunca deve passar de 2.5σ. Volume de tráfego pode variar mais (3σ). IPs únicos são os mais voláteis (2σ, porque um ataque distribuído aparece logo).

Testando com Dados Reais (ou Quase)

Vamos simular um cenário real: seu servidor normal recebe ~100 req/5min, com 2% de erro. De repente, um ataque de força bruta começa:

import random

def simulate_normal():
    return f'127.0.0.1 - - [{datetime.now().strftime("%d/%b/%Y:%H:%M:%S %z")}] '
           f'"GET /api/data HTTP/1.1" 200 {random.randint(500,1500)} '
           f'"-" "Mozilla/5.0"'

def simulate_attack():
    ip = f"192.168.{random.randint(1,255)}.{random.randint(1,255)}"
    return f'{ip} - - [{datetime.now().strftime("%d/%b/%Y:%H:%M:%S %z")}] '
           f'"POST /wp-login.php HTTP/1.1" 403 42 "-" "evil-bot/1.0"'

detector = AnomalyDetector()

# 200 requisições normais
for _ in range(200):
    detector.process_line(simulate_normal())

print("Baseline estabelecida.")
print(f"Error rate EMA: {detector.baselines['error_rate'].ema:.4f}")
print(f"Total EMA: {detector.baselines['total'].ema:.1f}")

# Agora o ataque: 300 requisições 403 em 30 segundos
for _ in range(300):
    alerts = detector.process_line(simulate_attack())
    if alerts:
        print(f"🚨 ALERTA: {alerts[0]}")
        break

Execute isso e você verá o detector disparar na 15ª requisição maliciosa. 15 requisições. Não 4.237. Não depois de 23 dias. Em segundos.

O Box Perrengue: Quando Tudo Dá Errado ao Mesmo Tempo

CENÁRIO DO INFERNO: É sexta-feira, 18h. Você está prestes a sair. Seu detector dispara: “anomalia em total de requisições”. Você olha: tráfego 10x maior. Mas todos os status são 200. É um ataque? É um viral? É um bot de SEO que finalmente descobriu seu site?

Você decide ignorar. Sábado de manhã, o servidor está offline. O que aconteceu? O “tráfego” era um ataque de reflexão/amplificação NTP. Os 200s eram respostas ao ataque, e seu link de rede saturou. O detector acertou. Você que errou ao não investigar.

Moral: anomalias são sintomas, não diagnósticos. O detector te diz “algo está estranho”. Você precisa olhar e entender o porquê. Nunca ignore um alerta só porque os status parecem normais.

Integração com Alertas Reais

Detectar é metade da batalha. A outra metade é te acordar às 3:47 da manhã quando necessário. Vamos adicionar notificações via webhook (que você pode rotear para seu próprio handler seguro ou direto pro Telegram/Discord):

import requests
import json
from datetime import datetime

class AlertNotifier:
    def __init__(self, webhook_url=None):
        self.webhook_url = webhook_url

    def notify(self, alerts, snapshot):
        if not alerts:
            return
        msg = f"🚨 *ANOMALIA DETECTADA* — {datetime.now().isoformat()}\n\n"
        for a in alerts:
            emoji = "🔴" if a["severity"] == "critical" else "🟡"
            msg += f"{emoji} `{a['metric']}`: {a['value']} (z={a['zscore']})\n"
        msg += f"\n📊 Snapshot: {json.dumps(snapshot, indent=2)}"
        if self.webhook_url:
            try:
                requests.post(self.webhook_url, json={"text": msg}, timeout=10)
            except Exception as e:
                print(f"Falha ao notificar: {e}")
        else:
            print(msg)

Em produção, eu recomendo usar o Circuit Breaker no notifier também. Se seu webhook cair, você não quer que o detector pare de funcionar por causa disso.

Executando em Produção: systemd e Rotação de Logs

Para rodar isso como serviço, crie um script que consome com tail -f e um serviço systemd:

#!/usr/bin/env python3
# /opt/monitor/detectord.py
import subprocess, sys, signal
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent))
from detector import AnomalyDetector, AlertNotifier

detector = AnomalyDetector()
notifier = AlertNotifier(webhook_url="https://seu-webhook.com/alerts")

def handle(sig, frame):
    print("Desligando graciosamente...")
    sys.exit(0)

signal.signal(signal.SIGTERM, handle)
signal.signal(signal.SIGINT, handle)

process = subprocess.Popen(
    ["tail", "-f", "-n", "0", "/var/log/nginx/access.log"],
    stdout=subprocess.PIPE, text=True
)

for line in process.stdout:
    alerts = detector.process_line(line.strip())
    if alerts:
        snapshot = detector.collector.snapshot()
        notifier.notify(alerts, snapshot)

E o serviço systemd:

# /etc/systemd/system/log-anomaly-detector.service
[Unit]
Description=Detector de Anomalias em Logs
After=network.target

[Service]
Type=simple
ExecStart=/usr/bin/python3 /opt/monitor/detectord.py
Restart=always
RestartSec=5
User=monitor
Group=monitor

[Install]
WantedBy=multi-user.target

Limitações e Quando Ir Além

Este sistema é intencionalmente simples. Ele detecta anomalias estatísticas, não semânticas. Ele não vai saber que um payload SQL injection é malicioso — só que a taxa de erro subiu. Para isso, você precisa de análise de assinaturas ou um WAF.

Mas aqui está a verdade: 95% dos ataques deixam rastros estatísticos óbvios se você estiver olhando. Scans de portas aumentam conexões. Força bruta aumenta requisições POST. DDoS aumenta volume. Data exfiltration aumenta bytes de saída. Estatística captura tudo isso, em tempo real, com código que cabe numa tela.

Conclusão: Seu Servidor Precisa de um Sistema Nervoso

O que construímos aqui é um sistema nervoso digital: ele sente, compara, e grita quando algo doi. Não é um cérebro — não precisa ser. Um cérebro é você, com seu contexto, sua intuição, sua capacidade de olhar pra um z-score de 4.2 e decidir se é um lançamento de produto ou um ataque em andamento.

O detector é o reflexo. A consciência é sua.

E o custo? Zero em licenças. Python puro. Roda em qualquer lugar. Sem dependências externas. Sem SaaS. Sem “me envie seus logs para nossa nuvem segura (confia)”. Seus dados, suas regras, seus alertas.

Configure isso hoje. Não amanhã. Porque o ataque que vai te pegar desprevenido provavelmente já começou — você só não sabe ainda.

Qual Automação Você Quer Ver no Próximo Post?

Eu escrevi isso porque precisava de um detector que funcionasse agora, não depois de configurar ELK, Prometheus, e mais 47 containers Docker. Mas sei que cada infraestrutura é diferente.

Qual automação de segurança ou monitoramento você está precisando? Um detector de anomalias para banco de dados? Um monitor de integridade de arquivos críticos? Um sistema de honeypot local? Deixe nos comentários — a fila de prioridade do Lab da Garra é ditada por quem precisa de ajuda de verdade.

Posts Similares