Proof#

Prova de conceito do click_watcher.

Criação do ambiente#

vamos utilizar uv para gerenciar o projeto python.

1. Inicializar o projeto

1
uv init

2. Adicionar dependencias

1
uv add watchdog websockets

3. Iniciar o virtual environment

1
source .venv/bin/activate

Monitorando edição dos logs#

Para monitorar qualquer nova atualização e ou modificação do arquivo de alertas, iremos utilizar a biblioteca watchdog.

Watchdog funciona em multiplas plataformas, não apenas linux

A biblioteca watchdog utiliza no linux o inotify para monitorar qualquer mudança evento de filesystem em um diretório.

Vamos para um exemplo antes de trazer o código final.

Código para identificar alterações no arquivo

1
from watchdog.observers import Observer
2
from watchdog.events import FileSystemEventHandler
3
import time
4
import os
5

6
file = './test.log'
7

8
class LogWatcher(FileSystemEventHandler):
9
    def on_modified(self, event):
10
        global file
11
        if os.path.abspath(event.src_path) == os.path.abspath(file):
12
            print('arquivo modificado')
13

14
watcher = LogWatcher()
15
observer = Observer()
16
observer.schedule(watcher, os.path.dirname(file) or '.', recursive=False)
17
observer.start() # inicia thread em background
18
while True:
19
    time.sleep(1)

Com esse código, conseguimos identificar eventos de mudanças no arquivo, porém ainda não conseguimos ver o conteúdo que foi alterado, então, como iremos fazer?

Para monitorar também alterações no arquivo, nós iremos:

1. Abrir o arquivo
2. Mudar a nossa posição do arquivo para o final
3. Para cada evento de on_modified, realizar a leitura e atualizar a nossa posição no arquivo.

1
...
2
from io import SEEK_END, SEEK_SET
3

4
class LogWatcher(FileSystemEventHandler):
5
    def __init__(self):
6
        self.fd = open(file, 'r') # abre o arquivo
7
        self.fd.seek(0, SEEK_END) # muda a posição para o final
8

9
    def on_modified(self, event):
10
        ...
11
        for log in self.fd: # lê novas entradas do arquivo e atualiza a posição
12
            log = log.strip()
13
            self.process_log(log)
14
    def process_log(self, log):
15
        ...

Veja que tem algumas variaveis que não foram utilizadas ainda, iremos utilizar nos proximos posts da série.

Depois disso você pode ter se perdido, mas calma que eu vou te explicar.

1
if len(self.buffer) >= self.max_batch:
2
    flush = True

1
def on_deleted(self, event):
2
    ...
3
    self._close_file() # fecha o descritor do arquivo
4
def on_created(self, event):
5
    ...
6
    self.seek_end() # acessa novamente o arquivo
7
    self.file.seek(0, SEEK_SET) # posiciona o arquivo para o inicio para não perder logs do novo arquivo

Testando o projeto file watcher#

Para testar o projeto, vamos utilizar o click_server.

O click_server é um servidor NodeJS que implementa um servidor websocket usando o framework NestJS!

Vamos testar o ambiente simulando como seria com o Wazuh.

1
npm run start

1
uv run main.py -u 'ws://localhost:3000/' -f ./alerts.json

1
uv run test/random_wazuh_events.py 5 -o alerts.json

1
uv run main.py -u 'ws://localhost:3000/' -f /var/ossec/logs/alerts/alerts.json

Proximos passos#

No proximo post irei desenvolver e publicar o servidor completo, realizando o envio dos alertas dinamicamente para o ClickHouse e ja funcionando com os Webhooks