WAF em Python puro — firewall de aplicação web que bloqueia SQL injection, XSS e path traversal sem depender de Cloudflare

WAF em Python Puro: O Firewall de Aplicação Web Que Bloqueia SQL Injection, XSS e Path Traversal Sem Depender de Cloudflare

Era 3 da manhã quando o Slack explodiu. Não era um deploy que deu errado — era algo pior. Alguém tinha encontrado o endpoint de busca do nosso sistema e estava testando, com paciência cirúrgica, cada vetor de SQL injection que existe. ' OR 1=1 --, union selects aninhados, blind injection com time-based delays. O banco de dados sobreviveu porque as queries eram parametrizadas. Mas os logs? Os logs pareciam um diário de guerra.

Naquele momento eu percebi que rate limiter sozinho não resolve. Você precisa de algo que entenda o que está chegando — não apenas quantas requisições por segundo, mas o conteúdo delas. Um Web Application Firewall (WAF). E eu queria construir um em Python puro, sem depender de Cloudflare, ModSecurity ou nginx.

Neste artigo, vou te mostrar exatamente como construí esse WAF — um middleware que inspeciona cada request HTTP, detecta os vetores de ataque mais comuns (SQL injection, XSS, path traversal, command injection) e bloqueia tudo antes de chegar na sua aplicação. Sem dependências. Sem configuração de servidor. Sem pagar plano Enterprise de CDN.

Se você já implementou um rate limiter e agora quer dar o próximo passo na segurança, chegou ao lugar certo.

O Que É um WAF e Por Que o Seu Rate Limiter Não Substitui Um

Um rate limiter responde a pergunta: “quantas requisições este cliente fez nos últimos X segundos?”. É volumétrico. Trata todos os requests como iguais.

Um WAF responde a uma pergunta diferente: “o que este request está tentando fazer?”. É semântico. Lê headers, query strings, body, cookies e decide se o conteúdo é malicioso.

Na prática, os dois se complementam:

  • Rate Limiter: bloqueia o atacante que manda 10.000 requests/s
  • WAF: bloqueia o atacante que manda um request com '; DROP TABLE users; -- no campo de busca

Eu aprendi isso da forma difícil. Tinha rate limiter. Tinha honeypot. Mas o atacante paciente — aquele que manda uma requisição a cada 30 segundos — passava limpo. Precisava de algo que lesse o conteúdo.

Arquitetura do WAF: O Pipeline de Inspeção

O WAF que vamos construir funciona como um pipeline de inspeção. Cada request passa por uma sequência de verificadores. Se qualquer um deles detectar algo suspeito, o request é bloqueado imediatamente.

class WAFMiddleware:
    """Middleware WAF que inspeciona requests HTTP e bloqueia ataques."""

    def __init__(self, strict_mode: bool = True):
        self.strict_mode = strict_mode
        self.blocked_count = 0
        self.total_count = 0
        self._rules = self._load_rules()

    def inspect(self, request: dict) -> WAFResult:
        """Pipeline principal de inspeção."""
        self.total_count += 1

        # Extrair todos os vetores de ataque do request
        vectors = self._extract_vectors(request)

        # Aplicar cada regra em sequência
        for rule in self._rules:
            match = rule.check(vectors)
            if match:
                self.blocked_count += 1
                return WAFResult(
                    blocked=True,
                    rule_name=rule.name,
                    severity=rule.severity,
                    detail=match.detail
                )

        return WAFResult(blocked=False)

    def _extract_vectors(self, request: dict) -> AttackVectors:
        """Extrai todos os campos inspecionáveis de um request."""
        return AttackVectors(
            query_string=request.get("query_string", ""),
            body=request.get("body", ""),
            headers=request.get("headers", {}),
            cookies=request.get("cookies", ""),
            path=request.get("path", ""),
            user_agent=request.get("headers", {}).get("User-Agent", ""),
        )

Repare na simplicidade: extrair vetores → aplicar regras → bloquear ou liberar. A mágica está nos verificadores.

Detectando SQL Injection: Além do OR 1=1

A maioria dos tutoriais de SQL injection detection para no OR 1=1. Na vida real, atacantes usam técnicas muito mais sofisticadas: union selects aninhados, encoding tricks, comentários SQL para bypass de filtros simples.

O nosso verificador precisa cobrir múltiplos vetores:

import re
from dataclasses import dataclass, field
from typing import Optional, List
from enum import Enum

class Severity(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

@dataclass
class MatchDetail:
    pattern_matched: str
    vector: str
    payload_sample: str

@dataclass
class WAFResult:
    blocked: bool
    rule_name: str = ""
    severity: Severity = Severity.LOW
    detail: Optional[MatchDetail] = None

@dataclass
class AttackVectors:
    query_string: str = ""
    body: str = ""
    headers: dict = field(default_factory=dict)
    cookies: str = ""
    path: str = ""
    user_agent: str = ""

    @property
    def all_text(self) -> str:
        """Concatena todos os vetores para inspeção unificada."""
        parts = [
            self.query_string, self.body, self.cookies,
            self.path, self.user_agent
        ]
        parts.extend(self.headers.values())
        return " ".join(parts)


class SQLInjectionRule:
    """Detecta SQL injection em múltiplos padrões."""

    name = "sql_injection"
    severity = Severity.CRITICAL

    # Padrões organizados por sofisticação
    PATTERNS = [
        # Básicos
        (r"(?i)(\b(SELECT|INSERT|UPDATE|DELETE|DROP|UNION|ALTER)\b.*\b(FROM|INTO|TABLE|WHERE|SET)\b)", "SQL keyword chain"),
        (r"(?i)('|\"|;)\s*(OR|AND)\s+.*=.*", "Boolean-based injection"),
        (r"(?i)\bUNION\b\s+(ALL\s+)?\bSELECT\b", "UNION SELECT"),

        # Intermediários
        (r"(?i)--\s*$", "SQL comment terminator"),
        (r"(?i)/\*.*\*/", "Block comment injection"),
        (r"(?i)\b(SLEEP|BENCHMARK|WAITFOR)\s*\(", "Time-based blind injection"),

        # Avançados
        (r"(?i)\bCHAR\s*\(\s*\d+", "CHAR() encoding bypass"),
        (r"(?i)0x[0-9a-fA-F]{4,}", "Hex encoding bypass"),
        (r"(?i)\bCONCAT\s*\(", "CONCAT string assembly"),
        (r"(?i)\bINFORMATION_SCHEMA\b", "Schema enumeration"),
        (r"(?i)\bLOAD_FILE\s*\(", "File read attempt"),
        (r"(?i)\bINTO\s+OUTFILE\b", "File write attempt"),
    ]

    def __init__(self):
        self._compiled = [(re.compile(p), desc) for p, desc in self.PATTERNS]

    def check(self, vectors: AttackVectors) -> Optional[MatchDetail]:
        """Verifica SQL injection em todos os vetores."""
        targets = [
            ("query_string", vectors.query_string),
            ("body", vectors.body),
            ("cookies", vectors.cookies),
            ("path", vectors.path),
        ]

        for vector_name, text in targets:
            if not text:
                continue
            for pattern, description in self._compiled:
                match = pattern.search(text)
                if match:
                    return MatchDetail(
                        pattern_matched=description,
                        vector=vector_name,
                        payload_sample=text[:100]
                    )
        return None

A chave aqui é a camada de padrões. Não tentamos pegar tudo com uma regex monstro — cada padrão é específico e tem um descritor legível. Quando o log diz “UNION SELECT detectado em query_string”, qualquer pessoa do time entende o que aconteceu.

Proteção Contra XSS: Cross-Site Scripting em Todas as Suas Formas

XSS é o ataque camaleão. Pode vir como <script>alert(1)</script> no campo de comentário, ou como javascript:alert(document.cookie) no atributo href de um link, ou ainda como <img onerror=...> escondido dentro de um avatar upload.

Existem três tipos principais que precisamos cobrir:

  • Reflected XSS: payload no query string, refletido na resposta
  • Stored XSS: payload salvo no banco, executado em outros usuários
  • DOM-based XSS: manipulação do DOM no client-side via JavaScript
class XSSRule:
    """Detecta Cross-Site Scripting em múltiplas variações."""

    name = "xss"
    severity = Severity.HIGH

    PATTERNS = [
        # Script tags diretas
        (r"(?i)<\s*script[^>]*>", "Direct script tag"),
        (r"(?i)<\s*/\s*script\s*>", "Script closing tag"),

        # Event handlers
        (r"(?i)\bon\w+\s*=\s*['\"]?[^'\">\s]", "Event handler injection"),
        (r"(?i)\b(onerror|onload|onmouseover|onfocus|onblur)\s*=", "Common event handler"),

        # Protocol handlers
        (r"(?i)javascript\s*:", "JavaScript protocol handler"),
        (r"(?i)vbscript\s*:", "VBScript protocol handler"),
        (r"(?i)data\s*:\s*text/html", "Data URI HTML injection"),

        # Encoding tricks
        (r"(?i)&#\d+;", "HTML entity encoding"),
        (r"(?i)&#x[0-9a-fA-F]+;", "Hex entity encoding"),
        (r"(?i)%3Cscript", "URL-encoded script tag"),

        # Tag injection
        (r"(?i)<\s*(img|iframe|object|embed|svg|math)[^>]*>", "Dangerous HTML tag"),
        (r"(?i)<\s*link[^>]*>", "Link tag injection"),
        (r"(?i)<\s*meta[^>]*>", "Meta tag injection"),
    ]

    def __init__(self):
        self._compiled = [(re.compile(p), desc) for p, desc in self.PATTERNS]

    def check(self, vectors: AttackVectors) -> Optional[MatchDetail]:
        targets = [
            ("query_string", vectors.query_string),
            ("body", vectors.body),
            ("cookies", vectors.cookies),
        ]

        for vector_name, text in targets:
            if not text:
                continue
            for pattern, description in self._compiled:
                match = pattern.search(text)
                if match:
                    return MatchDetail(
                        pattern_matched=description,
                        vector=vector_name,
                        payload_sample=text[:100]
                    )
        return None

Path Traversal: Quando o Atacante Quer Ler /etc/passwd

Path traversal é um dos ataques mais antigos e ainda surpreendentemente eficaz. A ideia é simples: o atacante manipula parâmetros de URL que representam caminhos de arquivo para acessar arquivos fora do diretório permitido.

O vetor clássico é ../../etc/passwd, mas existem variações que burlam filtros ingênuos:

  • ..%2F..%2Fetc%2Fpasswd — URL encoding
  • ....//....//etc/passwd — double-dot bypass
  • %252e%252e%252f — double encoding
  • ..\\..\\windows\\system32\\config\\sam — backslash em Windows
class PathTraversalRule:
    """Detecta tentativas de path traversal."""

    name = "path_traversal"
    severity = Severity.HIGH

    # Arquivos sensíveis que atacantes costumam buscar
    SENSITIVE_FILES = [
        "/etc/passwd", "/etc/shadow", "/etc/hosts",
        "/proc/self/environ", "/proc/self/cmdline",
        "/.env", "/.git/config", "/.aws/credentials",
        "wp-config.php", "web.config", ".htaccess",
        "/windows/system32", "boot.ini",
    ]

    PATTERNS = [
        # Sequências de traversal
        (r"(\.\./|\.\.\\|%2e%2e%2f|%2e%2e/|\.\.%2f|%252e%252e%252f)", "Directory traversal sequence"),

        # Dupla sequência
        (r"(\.\.\.\./|\.\.//)", "Double traversal attempt"),

        # Null byte injection (para truncar extensão)
        (r"(%00|\\x00)", "Null byte injection"),
    ]

    def __init__(self):
        self._compiled = [(re.compile(p, re.IGNORECASE), desc) for p, desc in self.PATTERNS]

    def check(self, vectors: AttackVectors) -> Optional[MatchDetail]:
        # Verificar padrões de traversal
        targets = [
            ("query_string", vectors.query_string),
            ("path", vectors.path),
            ("body", vectors.body),
        ]

        for vector_name, text in targets:
            if not text:
                continue

            # Checar padrões de traversal
            for pattern, description in self._compiled:
                match = pattern.search(text)
                if match:
                    return MatchDetail(
                        pattern_matched=description,
                        vector=vector_name,
                        payload_sample=text[:100]
                    )

            # Checar arquivos sensíveis
            text_lower = text.lower()
            for sensitive in self.SENSITIVE_FILES:
                if sensitive.lower() in text_lower:
                    return MatchDetail(
                        pattern_matched=f"Sensitive file access: {sensitive}",
                        vector=vector_name,
                        payload_sample=text[:100]
                    )

        return None

Command Injection: Quando o Hacker Quer Shell

Command injection acontece quando o atacante consegue injetar comandos do sistema operacional através de inputs da aplicação. É o ataque mais devastador da lista — dá acesso direto ao servidor.

Vetores comuns incluem operadores de shell (;, |, &&, ||), backticks para subshell, e variáveis de ambiente como $HOME ou ${IFS}.

class CommandInjectionRule:
    """Detecta tentativas de command injection."""

    name = "command_injection"
    severity = Severity.CRITICAL

    # Comandos que atacantes costumam executar
    DANGEROUS_COMMANDS = [
        "cat", "ls", "id", "whoami", "uname",
        "wget", "curl", "nc", "ncat", "netcat",
        "python", "perl", "ruby", "php",
        "bash", "sh", "zsh", "cmd", "powershell",
        "chmod", "chown", "mkdir", "rm", "mv",
    ]

    PATTERNS = [
        # Operadores de shell
        (r"(?<=\s|^);\s*\w+", "Shell command chaining with ;"),
        (r"\|\s*\w+", "Pipe to command"),
        (r"&&\s*\w+", "Command chaining with &&"),
        (r"\|\|\s*\w+", "Command chaining with ||"),

        # Subshell
        (r"`[^`]+`", "Backtick subshell"),
        (r"\$\([^)]+\)", "Dollar-paren subshell"),

        # Redirecionamento
        (r">\s*/", "Output redirect to path"),
        (r"<\s*/", "Input redirect from path"),

        # Variáveis de ambiente
        (r"\$\{?\w+\}?", "Environment variable access"),

        # Newline injection
        (r"(%0a|%0d|\\n|\\r)", "Newline injection"),
    ]

    def __init__(self):
        self._compiled = [(re.compile(p), desc) for p, desc in self.PATTERNS]

    def check(self, vectors: AttackVectors) -> Optional[MatchDetail]:
        targets = [
            ("query_string", vectors.query_string),
            ("body", vectors.body),
            ("cookies", vectors.cookies),
        ]

        for vector_name, text in targets:
            if not text:
                continue
            for pattern, description in self._compiled:
                match = pattern.search(text)
                if match:
                    # Verificar se é um comando perigoso
                    for cmd in self.DANGEROUS_COMMANDS:
                        if cmd in match.group(0).lower():
                            return MatchDetail(
                                pattern_matched=f"{description} ({cmd})",
                                vector=vector_name,
                                payload_sample=text[:100]
                            )
        return None

Montando o Pipeline Completo

Agora vamos juntar tudo. O WAF completo é um pipeline onde cada regra é um filtro independente. Se qualquer regra disparar, o request é bloqueado.

class WAFMiddleware:
    """WAF completo com pipeline de inspeção."""

    def __init__(self, strict_mode: bool = True, log_file: str = "waf.log"):
        self.strict_mode = strict_mode
        self.log_file = log_file
        self.blocked_count = 0
        self.total_count = 0

        # Pipeline de regras (ordem importa)
        self._rules = [
            SQLInjectionRule(),
            XSSRule(),
            PathTraversalRule(),
            CommandInjectionRule(),
        ]

    def inspect(self, request: dict) -> WAFResult:
        """Pipeline principal — aplica todas as regras em sequência."""
        self.total_count += 1
        vectors = self._extract_vectors(request)

        for rule in self._rules:
            match = rule.check(vectors)
            if match:
                self.blocked_count += 1
                self._log_block(rule, match, request)
                return WAFResult(
                    blocked=True,
                    rule_name=rule.name,
                    severity=rule.severity,
                    detail=match
                )

        return WAFResult(blocked=False)

    def _extract_vectors(self, request: dict) -> AttackVectors:
        """Extrai todos os campos inspecionáveis."""
        return AttackVectors(
            query_string=request.get("query_string", ""),
            body=request.get("body", ""),
            headers=request.get("headers", {}),
            cookies=request.get("cookies", ""),
            path=request.get("path", ""),
            user_agent=request.get("headers", {}).get("User-Agent", ""),
        )

    def _log_block(self, rule, match: MatchDetail, request: dict):
        """Registra bloqueio em log estruturado."""
        import json, datetime
        entry = {
            "timestamp": datetime.datetime.utcnow().isoformat(),
            "rule": rule.name,
            "severity": rule.severity.value,
            "pattern": match.pattern_matched,
            "vector": match.vector,
            "payload": match.payload_sample,
            "ip": request.get("ip", "unknown"),
        }
        with open(self.log_file, "a") as f:
            f.write(json.dumps(entry) + "\n")

    @property
    def stats(self) -> dict:
        """Estatísticas de bloqueio."""
        return {
            "total_requests": self.total_count,
            "blocked_requests": self.blocked_count,
            "block_rate": f"{self.blocked_count / max(self.total_count, 1) * 100:.1f}%",
            "rules_active": len(self._rules),
        }

Com isso, você tem um WAF funcional. Mas um WAF de verdade precisa de mais: testes.

Testando o WAF: Payloads Reais de Ataque

Um WAF sem testes é como um firewall sem regras — parece bonito no dashboard, mas não protege nada. Vamos montar uma suíte de testes com payloads reais, organizados por categoria de ataque.

import unittest

class TestWAF(unittest.TestCase):
    """Testes com payloads reais de ataque."""

    def setUp(self):
        self.waf = WAFMiddleware()

    # === SQL INJECTION ===
    def test_basic_sqli(self):
        req = {"query_string": "id=1' OR '1'='1"}
        result = self.waf.inspect(req)
        self.assertTrue(result.blocked)
        self.assertEqual(result.rule_name, "sql_injection")

    def test_union_select(self):
        req = {"query_string": "id=1 UNION SELECT username,password FROM users"}
        result = self.waf.inspect(req)
        self.assertTrue(result.blocked)

    def test_time_based_blind(self):
        req = {"query_string": "id=1; SLEEP(5)--"}
        result = self.waf.inspect(req)
        self.assertTrue(result.blocked)

    def test_information_schema(self):
        req = {"query_string": "table=INFORMATION_SCHEMA.TABLES"}
        result = self.waf.inspect(req)
        self.assertTrue(result.blocked)

    # === XSS ===
    def test_script_tag(self):
        req = {"body": "comment=<script>alert('XSS')</script>"}
        result = self.waf.inspect(req)
        self.assertTrue(result.blocked)
        self.assertEqual(result.rule_name, "xss")

    def test_event_handler(self):
        req = {"body": 'name=<img onerror=alert(1) src=x>'}
        result = self.waf.inspect(req)
        self.assertTrue(result.blocked)

    def test_svg_onload(self):
        req = {"body": "<svg/onload=alert(1)>"}
        result = self.waf.inspect(req)
        self.assertTrue(result.blocked)

    # === PATH TRAVERSAL ===
    def test_etc_passwd(self):
        req = {"path": "/files/../../etc/passwd"}
        result = self.waf.inspect(req)
        self.assertTrue(result.blocked)
        self.assertEqual(result.rule_name, "path_traversal")

    def test_encoded_traversal(self):
        req = {"query_string": "file=%2e%2e%2f%2e%2e%2fetc%2fshadow"}
        result = self.waf.inspect(req)
        self.assertTrue(result.blocked)

    def test_env_file(self):
        req = {"path": "/download/.env"}
        result = self.waf.inspect(req)
        self.assertTrue(result.blocked)

    # === COMMAND INJECTION ===
    def test_semicolon_chain(self):
        req = {"body": "host=127.0.0.1; cat /etc/passwd"}
        result = self.waf.inspect(req)
        self.assertTrue(result.blocked)
        self.assertEqual(result.rule_name, "command_injection")

    def test_pipe_command(self):
        req = {"body": "input=test | whoami"}
        result = self.waf.inspect(req)
        self.assertTrue(result.blocked)

    # === LEGITIMATE REQUESTS (false positive check) ===
    def test_normal_search(self):
        req = {"query_string": "q=python tutorial for beginners"}
        result = self.waf.inspect(req)
        self.assertFalse(result.blocked)

    def test_normal_json_body(self):
        req = {"body": '{"name": "João", "email": "joao@example.com"}'}
        result = self.waf.inspect(req)
        self.assertFalse(result.blocked)

    def test_url_with_dots(self):
        req = {"path": "/files/report.2024.pdf"}
        result = self.waf.inspect(req)
        self.assertFalse(result.blocked)

if __name__ == "__main__":
    unittest.main()

Os testes de false positives são tão importantes quanto os de detecção. Um WAF que bloqueia requests legítimos é pior do que não ter WAF — seus usuários vão embora, e o time de suporte vai ficar maluco tentando entender por que formulários normais não funcionam.

Integrando com Flask, FastAPI e Django

O WAF que construímos é agnóstico de framework. Aqui está como integrá-lo nos três frameworks mais populares de Python:

Flask

from flask import Flask, request, abort, jsonify

app = Flask(__name__)
waf = WAFMiddleware()

@app.before_request
def waf_check():
    req = {
        "query_string": request.query_string.decode("utf-8", errors="ignore"),
        "body": request.get_data(as_text=True),
        "headers": dict(request.headers),
        "cookies": request.cookies,
        "path": request.path,
        "ip": request.remote_addr,
    }
    result = waf.inspect(req)
    if result.blocked:
        return jsonify({
            "error": "Request blocked by WAF",
            "rule": result.rule_name,
        }), 403

@app.route("/")
def index():
    return f"Requests inspecionados: {waf.stats}"

FastAPI

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()
waf = WAFMiddleware()

@app.middleware("http")
async def waf_middleware(request: Request, call_next):
    body = await request.body()
    req = {
        "query_string": str(request.query_params),
        "body": body.decode("utf-8", errors="ignore"),
        "headers": dict(request.headers),
        "cookies": dict(request.cookies),
        "path": request.url.path,
        "ip": request.client.host if request.client else "unknown",
    }
    result = waf.inspect(req)
    if result.blocked:
        return JSONResponse(
            status_code=403,
            content={"error": "Request blocked by WAF", "rule": result.rule_name}
        )
    return await call_next(request)

Django

# waf_middleware.py
import json
from django.http import JsonResponse

waf = WAFMiddleware()

class WAFMiddlewareDjango:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        req = {
            "query_string": request.META.get("QUERY_STRING", ""),
            "body": request.body.decode("utf-8", errors="ignore"),
            "headers": {
                k.replace("HTTP_", ""): v
                for k, v in request.META.items()
                if k.startswith("HTTP_")
            },
            "cookies": request.COOKIES,
            "path": request.path,
            "ip": request.META.get("REMOTE_ADDR", "unknown"),
        }
        result = waf.inspect(req)
        if result.blocked:
            return JsonResponse(
                {"error": "Request blocked by WAF", "rule": result.rule_name},
                status=403
            )
        return self.get_response(request)

Em settings.py, adicione ao middleware:

MIDDLEWARE = [
    # ... outros middlewares
    "waf_middleware.WAFMiddlewareDjango",
]

Dashboard de Bloqueios em Tempo Real

De nada adianta ter um WAF se você não consegue ver o que ele está bloqueando. Aqui está um gerador de relatório que lê os logs estruturados e produz um dashboard em texto:

import json
from collections import Counter, defaultdict
from datetime import datetime

def waf_dashboard(log_file: str = "waf.log") -> str:
    """Gera dashboard de bloqueios a partir dos logs."""
    entries = []
    try:
        with open(log_file, "r") as f:
            for line in f:
                line = line.strip()
                if line:
                    entries.append(json.loads(line))
    except FileNotFoundError:
        return "Nenhum log encontrado. O WAF ainda não bloqueou nada."

    if not entries:
        return "Nenhum bloqueio registrado."

    # Contadores
    rule_counts = Counter(e["rule"] for e in entries)
    severity_counts = Counter(e["severity"] for e in entries)
    ip_counts = Counter(e.get("ip", "unknown") for e in entries)
    vector_counts = Counter(e["vector"] for e in entries)

    # Últimos 24h
    now = datetime.utcnow()
    recent = [
        e for e in entries
        if (now - datetime.fromisoformat(e["timestamp"])).total_seconds() < 86400
    ]

    # Top IPs
    top_ips = ip_counts.most_common(5)

    # Montar relatório
    lines = [
        "═" * 60,
        "  🔒 WAF DASHBOARD — Relatório de Bloqueios",
        "═" * 60,
        f"\n📊 Total de bloqueios: {len(entries)}",
        f"🕐 Últimas 24h: {len(recent)}",
        "\n📋 Bloqueios por Regra:",
    ]
    for rule, count in rule_counts.most_common():
        bar = "█" * min(count, 40)
        lines.append(f"  {rule:25s} {count:5d} {bar}")

    lines.append("\n⚠️  Bloqueios por Severidade:")
    for sev, count in sorted(severity_counts.items(), key=lambda x: -x[1]):
        lines.append(f"  {sev:10s} {count:5d}")

    lines.append("\n🌐 Top 5 IPs Bloqueados:")
    for ip, count in top_ips:
        lines.append(f"  {ip:20s} {count:5d}")

    lines.append("\n📍 Vetores de Ataque Mais Explorados:")
    for vector, count in vector_counts.most_common():
        lines.append(f"  {vector:20s} {count:5d}")

    lines.append("\n" + "═" * 60)
    return "\n".join(lines)

# Uso:
# print(waf_dashboard())

Esse dashboard pode ser executado via cron e enviado por email, ou integrado a um bot de Slack/Telegram. O ponto é: logs sem análise são ruído. Você precisa de dashboards.

Limitações e Próximos Passos

Sendo honesto sobre o que este WAF não faz:

  • Não faz decode automático — se o atacante usa base64 encoding no payload, o WAF não vai detectar. Você precisaria adicionar um step de decode pré-inspeção.
  • Não tem machine learning — é baseado em regras fixas. Ataques zero-day podem passar. Um WAF de produção combina regras com ML.
  • Não protege contra DDoS — para isso você precisa do rate limiter trabalhando em conjunto.
  • Não substitui validação de input — o WAF é uma camada adicional, não a única. Continue validando e sanitizando inputs na camada da aplicação.

Para evoluir, considere:

  1. Modo aprendizado: rode o WAF em modo log-only por uma semana para calibrar thresholds de falso positivo
  2. Regras customizadas: adicione regras específicas para a sua aplicação (ex.: se você tem upload de arquivos, adicione verificação de MIME type malicioso)
  3. Integração com fail2ban: quando o WAF bloquear N requests do mesmo IP, adicione ao ban list do firewall do sistema
  4. Correlação com honeypot: combine com um honeypot para enriquecer inteligência de ameaças

Conclusão: Segurança em Camadas, Não em Milagres

Aquela madrugada de ataque SQL injection me ensinou uma coisa que nenhum curso de segurança ensina: segurança não é um produto, é um processo. Não existe bala de prata. O que existe é uma pilha de defesas onde cada camada compensa as limitações da outra.

O WAF que construímos aqui é a camada semântica — aquela que o que está chegando e decide se faz sentido. Combinado com rate limiter (camada volumétrica), honeypot (camada de inteligência), e validação de input (camada de aplicação), você tem uma defesa em profundidade que não depende de nenhum serviço pago.

O código completo tem menos de 300 linhas. Zero dependências. Funciona com Flask, FastAPI e Django. E, mais importante: você entende cada linha. Quando algo der errado — e vai dar errado — você sabe exatamente onde olhar.

Qual é a próxima camada de segurança que você quer ver implementada em Python puro? Um detector de brute force inteligente? Um sistema de reputation scoring para IPs? Ou talvez um TLS certificate monitor que avisa quando seus certificados estão expirando? Me conta nos comentários que o próximo post pode ser o seu pedido.

Posts Similares