Indeksowania danych do klastra Elastic Stack to dobry moment na wyłuskanie wartości z danych. W przypadku Threat Hunting‘u, dobrze przygotowane dane mogą znacznie przyśpieszyć proces wyszukiwania zagrożeń i anomalii. W artykule uprościmy wyszukiwanie procesów LOLBAS.
Co to jest LOLBAS / LOLBIN?
LOLBAS (Living Off The Land Binaries, Scripts and Libraries) to niezłośliwe skrypty, binarki i biblioteki, które wykorzystywane są adwersarze do robienia złośliwych rzeczy. Pliki te są elementami rozwiązań Microsoftu i nierzadko są podpisane przez Microsoft.
Wykorzystywane są do eksploitacji, persystencji, eskalacji uprawnień, dostarczania złośliwego oprogramowania i obchodzenia detekcji. Na stronie projektu znajdziesz opis, funkcje, mapping do technik MITRE ATT&CK i przykłady użycia.
SITREP
Dostałeś/aś zadanie przeprowadzenia Threat Huntingu w oparciu o projekt LOLBAS 🤠.
Setup:
- Elastic Stack – możesz użyć zorteran/elastic-stack-docker-boilerplate
- Windows 10 + Winlogbeat + Sysmon z konfiguracją od Floriana
Repo:
https://github.com/zorteran/threat-hunting-lolbas-elastic-stack
Opcja 1. Łatwiej ale wolniej (ale trudniej)
Chcemy ograniczyć analizę do Event 1 (process creation) z Sysmon’a, więc odpalamy filtr event.provider: Microsoft-Windows-Sysmon AND event.code: 1 i przeglądamy… przeglądamy… i nagle jest!
Sam wykonałem mshta.exe aby ściągnąć i wyświetlić plik robots.txt z mojego bloga, więc wiedziałem czego szukam.
Problem z szukaniem
Co w przypadku gdy nie wiemy czego szukamy? Robimy nowe query!process.name: ("mshta.exe" or "certutil.exe" or "certos.exe" or "replace.exe" or ...) Sporo tych binarek. Dużo pisania. Ciężko wypisać wszystkie, a co dopiero pokatalogować je po funkcjonalności. Może w pierwszej kolejności najlepiej się skupić np. na ściąganiu plików za pomocą binarek lolbas?
Czy jest to problem? Zależy. Nie raz generowałem zapytania KQL za pomocą Excela lub Pythona. Ad-hoc zapytanie wykona się o ile mieścimy się w maksymalnej liczbie filtrów. Gorzej jeśli chcemy je powtarzać co 5 minut w regule detekcji….
Problem z wydajnością
Zastanawiałeś/aś się co dzieje się pod spodem gdy wykonujemy zapytanie? Na pomoc przychodzi Search Profiler znajdujący się w Dev Tools w Kibanie.
Na screenach powyżej widać ponad 5 krotną różnicę w czasie zapytań. Nie jest to precyzyjny pomiar (wykonałem go tylko raz), ale widać mniejszą lub większą różnicę. Na produkcyjnym klastrze i danych różnice będą bardziej widoczne.
Czy to źle?
Nie. Analityk w działaniach ad-hoc poradzi sobie z żonglowaniem zapytaniami, a czas wykonania zapytania będzie i tak wystarczający (Elasticsearch jest szybki i potrafi sporo wybaczyć). Jednak w przypadku reguł detekcji i powtarzalnych obserwacji takie podejście może być najwygodniejsze.
Opcja 2. Trudniej ale szybciej (ale łatwiej)
Będzie to opcja dłuższa, ale ciekawsza. Przygodę zaczniemy od odpalenia Jupytera 😎. Używałem go już z sukcesem we wpisie o 10 najważniejszych źródeł MITRE ATT&CK. LOLBAS to też projekt na Github’ie. Każda binarka to osobny plik YAML. Spróbujemy wyciągnąć listę binarek oraz ich funkcje. Będzie to słownik pod wzbogacanie danych z winlogbeat.
Przekształcenie danych z LOLBAS w CSV
Pierwszym krokiem jest ściągnięcie repozytorium.
!git clone https://github.com/LOLBAS-Project/LOLBAS.git
Następnie import bibliotek i załadowanie wszystkich YAML do listy.
import os
import sys
import yaml
from yaml.loader import SafeLoader
import pandas as pd
lolbas_list = []
for root, subdirs, files in os.walk(".\LOLBAS\yml"):
for file in files:
with open(f'{root}\\{file}') as f:
lolbas_entry = yaml.load(f, Loader=SafeLoader)
lolbas_list.append(lolbas_entry)
Ograniczymy się do nazwy i kategorii, a rezultat zapiszemy w pliku CSV. Tagi z funkcjami będą miały przedrostek lolbas_. Ograniczam się do binarek *.exe.
def unique(list1):
list_set = set(list1)
unique_list = (list(list_set))
return unique_list
def process_lolbas_entry(entry):
return {
'name': entry['Name'].lower(),
'tags': ",".join(unique(['lolbas'] + [ 'lolbas_' + x["Category"].lower() for x in entry["Commands"] ]))
}
df = pd.DataFrame(list(map(process_lolbas_entry, lolbas_list)))
df[df['name'].str.contains(".exe")].to_csv("lolbas_dict.csv",index=False,header=False)
Przygotowanie słownika i ingest pipeline
Wygenerowany plik CSV jest mały i można go zapisać w Elasticsearch za pomocą Kibany – Data Visualizer’a w module Machine Learning (jest to darmowa funkcjonalność). Utworzyłem indeks dict-lolbas. W Ingest pipeline wykorzystałem processor split, aby otrzymać tablicę tagów.
Dodany indeks wykorzystamy do wzbogacania danych za pomocą Enrich Policy i Enrich Processor. Wspominałem o tej metodzie w artykule gdzie wykrywaliśmy phishing na podstawie feedu z CERT Polska. W momencie wykonania _execute tworzony jest dedykowany indeks na bazie wskazanych indeksów źródłowych. Jest zoptymalizowany pod kątem liczby segmentów (1) i replik (odnoszę wrażenie, że shardów łącznie jest tyle ile hot/content node’ów).
PUT /_enrich/policy/lolbas-binaries
{
"match": {
"indices": "dict-lolbas",
"match_field": "process_name",
"enrich_fields": ["tags"]
}
}
PUT _enrich/policy/lolbas-binaries/_execute
Finalnie mamy pipeline który:
- Wzbogaca dokument o pole
lolbas_enrichment - Dodaje tagi z pola
lolbas_enrichment.tags - Usuwa pole
lolbas_enrichment
PUT _ingest/pipeline/winlogbeat-lolbas-enrichment
{
"description": "Adding tags to lolbas processes",
"processors": [
{
"enrich": {
"policy_name": "lolbas-binaries",
"field": "process.name",
"target_field": "lolbas_enrichment",
"if": "ctx.process.name != null"
}
},
{
"foreach": {
"field": "lolbas_enrichment.tags",
"processor": {
"append": {
"field": "tags",
"value": [
"{{{_ingest._value}}}"
],
"allow_duplicates": false
}
},
"if": "ctx.lolbas_enrichment != null"
},
"remove": {
"field": "lolbas_enrichment",
"if": "ctx.lolbas_enrichment != null"
}
}
]
}
Ostatni element układanki to dodanie Pipeline Processor do głównego ingest pipeline wykorzystywanego przez winlogbeat’a, czyli w tym przypadku winlogbeat-8.5.2-routing. Pipeline ten składa się praktycznie z samych pod-pipeline’ów.
[
{
"set": {
"field": "event.ingested",
"value": "{{_ingest.timestamp}}"
}
},
{
"pipeline": {
"name": "winlogbeat-8.5.2-security",
"if": "ctx?.winlog?.channel == 'Security' && ['Microsoft-Windows-Eventlog', 'Microsoft-Windows-Security-Auditing'].contains(ctx?.winlog?.provider_name)"
}
},
{
"pipeline": {
"name": "winlogbeat-8.5.2-sysmon",
"if": "ctx?.winlog?.channel == 'Microsoft-Windows-Sysmon/Operational'"
}
},
{
"pipeline": {
"name": "winlogbeat-8.5.2-powershell",
"if": "ctx?.winlog?.channel == 'Windows PowerShell'"
}
},
{
"pipeline": {
"if": "ctx?.winlog?.channel == 'Microsoft-Windows-PowerShell/Operational'",
"name": "winlogbeat-8.5.2-powershell_operational"
}
},
{
"pipeline": {
"name": "winlogbeat-lolbas-enrichment",
"if": "ctx?.process?.name != null",
"ignore_failure": true
}
}
]
Efekt
Chwila działania nowego pipeline’u i już widać efekty. Mamy do dyspozycji kilka tagów.
Za pomocą prostego filtra tags: "lolbas_download" widzimy nie tylko Event 1, ale również Event 3 (a docelowo wszystkie zdarzenia które mają pole process.name.
Zróbmy więc regułę! LOLBAS z funkcją Download odnoszą się do zasobów takich jak web (http) czy zasoby sieciowe (ftp, smb). Skupmy się na http. Wykorzystamy do tego EQL.
process where tags == "lolbas_download" and process.command_line: "*http*"
Wildcard query również można zoptymalizować tagując wcześniej dokumenty z frazą http na poziomie indeksowania do klastra Elasticsearch.
Czy to rozwiązanie jest kompletne?
Nie jest, ale nie ma sensu się tym przejmować. Zasada Pareto: 20 % wysiłku, 80 % efektu. Np. pierwsza binarka z listy LOLBAS “AppInstaller.exe” ma inną nazwę procesu. Baza rozwiązania jest. Trzeba trochę dopracować CSV’kę 😇.
Jaki jest morał tej historii?
- Wzbogacanie danych potrafi znacznie zwiększyć efektywność pracy analityka..
- Jupyter to fajne narzędzie, które minimalizuje potrzebę wykorzystania interfejsu białkowego (robienie czegoś “z łapy”)
