Enrichment VirusTotal dla alertów Cloudflare

Do SIEM trafiają alerty z Cloudflare. Alert pokazuje, która reguła zadziałała i z jakiego adresu IP przyszło żądanie, ale nic nie mówi o reputacji tego adresu z publicznych źródeł.

Analitycy mieli w zwyczaju ręcznie sprawdzać adresy z alertów Cloudflare w VirusTotal, co przy każdym alercie oznaczało powtarzalną pracę.

Cały skrypt napisałem samodzielnie, w Pythonie, bez AI.

Wejście

nowe alerty z reguł Cloudflare w SIEM; adres IP klienta z treści alertu

Źródło kontekstu

VirusTotal API v3 - /ip_addresses/{ip}

Wyjście

pola z wynikiem VirusTotal dopisane do dokumentu alertu

Biblioteki

requests, python-dotenv, ipaddress, logging

1. 1

   Pobranie nowych alertów

   Co minutę (cron) skrypt pyta SIEM zapytaniem Query DSL o najnowsze alerty z ostatniego okna czasowego, które nie mają jeszcze wyniku z VirusTotal. Limit alertów na jedno uruchomienie jest dopasowany do faktycznej liczby alertów w SIEM.

2. 2

   Wybór alertów Cloudflare

   Zostają tylko alerty z reguł Cloudflare, a z ich treści brany jest adres klienta.

3. 3

   Tylko adresy publiczne

   Do VirusTotal trafiają wyłącznie adresy publiczne (ipaddress.ip_address(ip).is_global).

4. 4

   Zapytanie do VirusTotal

   Z odpowiedzi skrypt bierze kraj, właściciela AS, ASN, zakres sieci, reputację i statystyki ostatniej analizy, a gdy adres jest oznaczony jako złośliwy, także listę silników, które go oznaczyły.

5. 5

   Zapis do alertu

   Wynik trafia do alertu w SIEM przez _update: operator, kraj, reputacja, liczba oznaczeń jako złośliwy i, jeśli są, silniki, które je dały.

6. 6

   Logi etapów i błędów

   Każdy etap zapisuje w logu swój czas i czas łączny, a błędy trafiają do osobnego, szczegółowego logu.

{
  "size": <limit>,
  "query": {
    "bool": {
      "must": [{ "range": { "@timestamp": { "gte": "now-<okno>" } } }],
      "must_not": [{ "exists": { "field": "<pole_wyniku>" } }]
    }
  },
  "sort": [{ "@timestamp": { "order": "desc" } }]
}

if ipaddress.ip_address(ip).is_global:
    response = requests.get(url, headers={"X-Apikey": VT_key})
    attributes = response.json()["data"]["attributes"]
    stats = attributes["last_analysis_stats"]
    result = {
        "country": attributes.get("country"),
        "isp_name": attributes.get("as_owner"),
        "reputation": attributes.get("reputation"),
        "malicious": stats.get("malicious"),
    }

{
  "virustotal": {
    "ISP_name": "LAB-EXAMPLE-AS",
    "Country": "ZZ",
    "Reputation": -12,
    "Malicious": 7,
    "FlaggedBy": "EngineA,EngineB,EngineC"
  }
}

Każdy alert sprawdzany raz

Zapytanie pomija alerty, które mają już wynik z VirusTotal, więc żaden alert nie jest sprawdzany dwa razy.

Tylko adresy publiczne

Adresy prywatne i wewnętrzne nigdy nie wychodzą poza organizację.

Limit dopasowany do ruchu

Liczba alertów pobieranych w jednym uruchomieniu jest skalibrowana do faktycznej liczby alertów Cloudflare w SIEM przy uruchamianiu co minutę.

Sekrety poza kodem

Klucz VirusTotal i dane dostępowe do SIEM są w .env, nie w kodzie.

Wynik w samym alercie

Analityk widzi reputację adresu od razu w alercie, bez otwierania VirusTotal.

Skrypt działa produkcyjnie: cron uruchamia go co minutę, a wyniki trafiają prosto do alertów w SIEM, z których korzysta cały zespół SOC.

• Enrichment obejmuje tylko alerty z reguł Cloudflare i tylko publiczne adresy IP.
• Reputacja z VirusTotal to kontekst dla analityka, a nie werdykt: decyzja o alercie zostaje po stronie człowieka.