Pular para o conteúdo

Detector em log

Ler um log de acesso, correlacionar por IP e alertar força bruta com os eventos que a causaram.

Gravar log é quase inútil sozinho: ninguém lê dez milhões de linhas. O que transforma log em segurança é a regra — cinco falhas do mesmo IP em um minuto — e o alerta que traz os eventos, para que quem investiga não precise voltar ao log para descobrir o que aconteceu.

PeçaO que ela exercita
D.motoras regras e a correlação
por := "ip"cinco falhas de cinco IPs não são força bruta
janela deslizante4 falhas às 23h59 e 4 às 00h01 contam juntas
o alerta com os eventosinvestigável sem voltar ao log

Estrutura#

text
vigia-log/
  src/
    ler.df         linha de log -> evento
    regras.df      o catalogo
    main.df        segue o arquivo e alerta
  tests/
forge.toml
[project]
name = "vigia-log"
version = "0.1.0"
description = "Detector de força bruta"
entry = "src/main.df"
dataforge = ">=1.1"

[dependencies]

[scripts]
start = "run src/main.df"
test = "test tests/"

O núcleo#

Este bloco roda sozinho — copie para um arquivo e rode dataforge run. Ele termina com assert, e é assim que esta página é conferida a cada build.

src/main.df
adopt Arcane.Deteccao as D

LOG := """203.0.113.7 POST /login 401
198.51.100.2 POST /login 200
203.0.113.7 POST /login 401
203.0.113.7 POST /login 401
192.0.2.10 POST /login 401
203.0.113.7 POST /login 401
203.0.113.7 POST /login 401"""

action evento_de(linha):
    ip, metodo, caminho, status := linha.split(" ")
    given caminho is "/login" and status is "401":
        yield {"nome": "login.falhou", "ip": ip}
    yield void

m := D.motor()
m.regra("forca-bruta", quando := "login.falhou", vezes := 5,
    janela := 60.0, gravidade := "alto", attack := "T1110", por := "ip")

alertas := []
cycle linha in LOG.lines():
    e := evento_de(linha)
    given e is not void:
        alertas := [...alertas, ...m.evento(e["nome"], {"ip": e["ip"]})]

cycle a in alertas:
    out $"{a['gravidade']}: {a['regra']} ({a['attack']}) com {len(a['eventos'])} eventos"

assert len(alertas) is 1
assert len(alertas[0]["eventos"]) is 5

O teste#

No projeto, a regra mora em src/ e o teste a importa pelo caminho relativo — dataforge test tests/ descobre o arquivo sozinho.

tests/nucleo_test.df
adopt ../src/main as L

crucible "leitura":
    trial "login bem-sucedido nao e evento":
        expect L.evento_de("1.2.3.4 POST /login 200") is void

As decisões#

DecisãoSem ela
correlação por IPcinco usuários errando a senha viram um ataque
a regra fala de evento, não de linhatrocar o formato do log reescreve as regras
o alerta carrega os eventosquem investiga volta ao log de 10 GB
o código ATT&CK (T1110)o alerta não se liga ao catálogo que o time de segurança usa

Para ir além#

  • Indicadores de comprometimento com prazo: Detecção.
  • Supressão, para mil alertas por minuto não virarem ruído: Arcane.Deteccao.
  • Ler o log enquanto ele cresce: Streams.

Volte para todos os tipos de projeto.