Pipelines (дамжуулах шугам) — Security Data Lake-ийн лог боловсруулалт — 8-р хэсэг: Examples

← Үндсэн хуудас

Функцийн жишээнүүд

Доорх жагсаалтад pipeline функц бүрийн жишээ rule-ийг үзүүллээ.

• array_contains — rule "array_contains" when true then set_field("contains_number", array_contains([1, 2, 3, 4, 5], 1)); set_field("does_not_contain_number", array_contains([1, 2, 3, 4, 5], 7)); set_field("contains_string", array_contains(["test", "test2"], "test")); set_field("contains_string_case_insensitive", array_contains(["test", "test2"], "TEST")); set_field("contains_string_case_sensitive", array_contains(["test", "test2"], "TEST", true)); end
• array_remove — rule "array_remove" when true then set_field("remove_number", array_remove([1, 2, 3], 2)); set_field("remove_string", array_remove(["one", "two", "three"], "two")); set_field("remove_missing", array_remove([1, 2, 3], 4)); set_field("remove_only_one", array_remove([1, 2, 2], 2)); set_field("remove_all", array_remove([1, 2, 2], 2, true)); end
• concat — let build_message_0 = concat(to_string($message.protocol), " connect from "); let build_message_1 = concat(build_message_0, to_string($message.src_ip)); let build_message_2 = concat(build_message_1, " to "); let build_message_3 = concat(build_message_2, to_string($message.dst_ip)); let build_message_4 = concat(build_message_3, " Port "); let build_message_5 = concat(build_message_4, to_string($message.dst_port)); set_field("message", build_message_5);
• contains — contains(to_string($message.hostname), "example.org", true)
• debug — Dropped message from <source>"let debug_message = concat("Dropped message from ", to_string($message.source));debug(debug_message);
• drop_message — rule "drop messages over 16383 characters" when has_field("message") AND regex(pattern: "^.{16383,}$", value: to_string($message.message)).matches == true then drop_message(); // added debug message to be notified about the dropped message debug( concat("dropped oversized message from ", to_string($message.source))); end
• ends_with — true буцаана: ends_with ( "Foobar Baz Quux" , "quux" , true );; false буцаана: ends_with ( "Foobar Baz Quux" , "Baz" );
• grok_exists — when grok_exists("USERNAME") then let parsed = grok("%{USERNAME:username}", to_string($message.message)); set_field("parsed_username", parsed.username); end
• hex_to_decimal_byte_list — hex_to_decimal_byte_list(value: "0x17B90004"); нь [23, 185, 0, 4] буцаана; hex_to_decimal_byte_list(value: "0x117B90004"); нь [1, 23, 185, 0, 4] буцаана; hex_to_decimal_byte_list(value: "17B90004"); нь [23, 185, 0, 4] буцаана; hex_to_decimal_byte_list(value: "117B90004"); нь [1, 23, 185, 0, 4] буцаана; hex_to_decimal_byte_list(value: "not_hex"); нь null буцаана
• is_not_null — is_null(src_addr)
• lookup — rule "dst_ip geoip lookup" when has_field("dst_ip") then let geo = lookup("geoip-lookup", to_string($message.dst_ip)); set_field("dst_ip_geolocation", geo["coordinates"]); set_field("dst_ip_geo_country_code", geo["country"].iso_code); set_field("dst_ip_geo_country_name", geo["country"].names.en); set_field("dst_ip_geo_city_name", geo["city"].names.en); end
• lookup_all — rule "function lookup all" when true then let values = lookup_all("lut_name", ["key1", "key2", "key3"]); set_field("values", values); end
• lookup_value — ("ip_lookup", to_string($message.src_addr));
• multi_grok — when true then set_fields( fields: multi_grok( patterns: [ "^ABC %{IPORHOST:msg_ip}: %{GREEDYDATA:abc_message}", "^123 %{IPORHOST:msg_ip}: %{GREEDYDATA:123_message}", "^ABC2 %{IPORHOST:abc_ip}: %{GREEDYDATA:abc_message}" ], value: to_string($message.message), only_named_captures: true ) ); end
• otx_lookup_domain — rule "PARSE IP to DNS" when has_field("source_ip") && regex( pattern: "^\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}$", value: to_string($message.source_ip) ).matches == true then let rs = lookup_value("dns_lookups", to_string($message.source_ip)); set_field("source_ip_dns", to_string(rs)); end
• otx_lookup_ip — rule "PARSE source_ip - otx-api-ip" when // validate message has a source_ip field has_field("source_ip") // validate that soruce IP is IPv4 format && regex( pattern: "^\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}$", value: to_string($message.source_ip) ).matches == true then let rs = otx_lookup_ip(to_string($message.source_ip)); set_fields(rs); end
• parse_unix_milliseconds — set_field ( "timestamp" , timestamp );
• regex_replace —

```
let username = regex_replace(".user: (.)", to_string($message.message), "$1");
```

• replace — let new_field = replace(to_string($message.message), "oo", "u"); // "fu ruft uta" let new_field = replace(to_string($message.message), "oo", "u", 1); // "fu rooft oota"
• route_to_stream — route_to_stream(id: "512bad1a535b43bd6f3f5e86");
• starts_with — true буцаана: starts_with ( "Foobar Baz Quux" , "foo" , true );; false буцаана: starts_with ( "Foobar Baz Quux" , "Quux" );
• string_array_add — rule "string_array_add" when true then set_field("add_number_to_string_array_converted", string_array_add(["1", "2"], 3)); set_field("add_number_array_to_string_array_converted", string_array_add(["1", "2"], [3, 4])); set_field("add_string", string_array_add(["one", "two"], "three")); set_field("add_string_again", string_array_add(["one", "two"], "two")); set_field("add_string_again_unique", string_array_add(["one", "two"], "two", true)); set_field("add_array_to_array", string_array_add(["one", "two"], ["three", "four"])); end
• substring — = substring(to_string($message.message), 0, 20);
• to_map — let json = parse_json(to_string($message.json_payload)); let map = to_map(json); set_fields(map);
• traffic_accounting_size — set_field( field: "license_usage", value: traffic_accounting_size() // size in bytes //value: traffic_accounting_size() / 1024 // size in kb );

Логийг баяжуулах (Log enrichment)

Lookup Tables (хайлтын хүснэгт)

Lookup tables нь мессежийн талбарын утгыг шинэ утгаар солих, эсвэл огт шинэ талбар үүсгэх замаар лог өгөгдлийг харгалзуулах, хөрвүүлэх, баяжуулах боломж олгоно. Жишээлбэл, статик CSV файлаар IP хаягийг хост нэртэй холбох, эсвэл гадаад өгөгдлийн эх сурвалжаар мессежийг аюулын мэдээлэл, газарзүйн байршил, активын мэдээллээр баяжуулж болно.

Энэ боломжоор түүхий лог өгөгдлийг дотоод систем болон гуравдагч талын интеграцийн нөхцөл байдлын мэдээллээр баяжуулж, илүү баялаг, хэрэгцээтэй мэдээлэл болгон хувиргана.

Бүрэлдэхүүн хэсгүүд

Lookup table систем дөрвөн бүрэлдэхүүн хэсгээс тогтоно.

Data Adapters (өгөгдлийн адаптер)

Data adapter нь утгыг бодитоор хайх үүрэгтэй. Тэр CSV файлаас унших, өгөгдлийн сантай холбогдох, эсвэл хүсэлт илгээж хайлтын үр дүнг авч болно.

Data adapter-ийн хэрэгжүүлэлтүүд залгагдах (pluggable) бөгөөд шинийг нэмэлт (plugin)-өөр нэмж болно.

Caches (кэш)

Кэш нь хайлтын үр дүнг хадгалж, хайлтын хурдыг сайжруулах болон өгөгдлийн сан, API-г ачаалалд оруулахгүй байх үүрэгтэй. Нэг кэшийн хэрэгжүүлэлтийг өөр өөр data adapter-д дахин ашиглахын тулд кэшийг тусдаа нэгж болгосон. Ингэснээр data adapter кэшийн талаар санаа зовох, өөрсдөө хэрэгжүүлэх шаардлагагүй.

Кэшийн хэрэгжүүлэлтүүд залгагдах бөгөөд шинийг нэмэлт (plugin)-өөр нэмж болно.

Lookup Tables

Lookup table нь data adapter болон кэшийн нэг нэг үүсгэлийг холбоно. Үүнийг converters, pipeline функцүүд, decorators-т ашиглахад шаардлагатай.

Lookup Results (хайлтын үр дүн)

Хайлтын үр дүнг lookup table нь data adapter-аар дамжуулан буцаах бөгөөд хоёр төрлийн өгөгдөл агуулж болно: single value (нэг утга) ба multi value (олон утга).

Single value нь тэмдэгт мөр, тоо эсвэл boolean байж болох бөгөөд converters, decorators болон pipeline rules-д ашиглагдана. Манай CSV жишээнд IP хаягаар хост нэр хайвал энэ нь хост нэрийн тэмдэгт мөр байна.

Multi value нь map буюу толь маягийн өгөгдлийн бүтэц бөгөөд хэд хэдэн өөр утга агуулж болно. Data adapter нэг түлхүүрт олон утга өгөх боломжтой үед энэ нь хэрэгтэй. Сайн жишээ нь geo-ip data adapter: тэр IP хаягийн өргөрөг, уртрагаас гадна байршлын хот, улсын мэдээллийг бас өгдөг. Одоогоор multi value-г зөвхөн pipeline rule дотор lookup() pipeline функцтэй хамт ашиглаж болно.

Жишээ 1: Нэг утга ба олон утга агуулсан CSV data adapter-ийн гаралт.

Жишээ 2: Нэг утга ба олон утга агуулсан geo-ip data adapter-ийн гаралт.

Setup (тохируулах)

Lookup tables-ийг System > Lookup Tables цонхонд тохируулна.

Lookup table-д дор хаяж нэг data adapter, нэг кэш шаардлагатай.

1. 1Data Adapter үүсгэнэ:

1. System > Lookup Tables > Data Adapters руу очно.
2. Create Adapter-ийг сонгоод data adapter-ийн төрлийг сонгоно.
3. Adapter-ийн тохиргооны маягтыг бөглөнө. Маягтад төрөл бүрийн суурилуулсан баримт бичиг орсон байдаг.

CIDR lookup идэвхтэй үед IP хаягийг яг таарах түлхүүртэй биш, CIDR блоктой харьцуулна.

Жишээлбэл:

```
192.168.100.0/24 → "Finance Department Subnet"
192.168.101.0/24 → "IT Department Subnet"
192.168.102.0/24 → "HR Department Subnet"
```

192.168.101.117 хаягийн хайлт "IT Department Subnet" буцаана.

TTL сонголт нь MongoDB дээр суурилсан lookup table эсвэл watchlist-ийн бичлэгийг тогтоосон хугацааны дараа автоматаар хүчингүй болгодог.

Энэ нь байнга өөрчлөгддөг өгөгдөлд хэрэгтэй, жишээлбэл:

- Цаг хугацаа өнгөрөх тусам хуучирдаг Indicators of Compromise (IoC, эвдэрсний шинж).
- Хэрэглэгчийн түгжээг хянах тохиолдол: нууц үг солигдсоныг тусгахын тулд бичлэг 72 цагийн дараа хүчингүй болж болно.

2. 2Кэш үүсгэнэ:

1. System > Lookup Tables > Caches руу очно.

2. Create Cache-ийг дарж кэшийн төрлийг сонгоно.

3. Кэшийн тохиргооны маягтыг бөглөнө. Маягтад орсон кэшийн төрлийн баримтыг уншиж танилцана.

> Тэмдэглэл: Тохиргооны үед Ignore empty results-ийг сонгоогүй бол null үр дүн кэшлэгдэнэ.

3. 3Lookup Table үүсгэнэ:

1. System > Lookup Tables руу очно.

2. Create Lookup Table-ийг сонгоно.

3. Өөрийн data adapter болон кэшийн үүсгэлийг сонгож, хүсвэл анхдагч утга тодорхойлно.

> Тэмдэглэл: Хайлт үр дүн буцаахгүй үед анхдагч утгыг ашиглана. Түлхүүр lookup table-д олдохгүй бол Security Data Lake тодорхойлсон анхдагч утгыг автоматаар буцаана.

Үүсгэсний дараа lookup table-ийг extractors, decorators, pipeline rules-д ашиглаж болно.

Usage (хэрэглээ)

Lookup tables-ийг Security Data Lake-ийн хэд хэдэн хэсэгт өгөгдлийг баяжуулж, нөхцөл байдлыг нэмэхэд ашиглаж болно:

• Converters – Мессеж хүлээн авах үед гаргаж авсан утгууд дээр хайлт хийнэ.
• Decorators – Хадгалсан өгөгдлийг өөрчлөхгүйгээр хайлтын үед мессежийг баяжуулна.
• Pipeline Rules – lookup() эсвэл lookup_value() функцээр логикийг динамикаар хэрэгжүүлнэ.

Built-in Data Adapters (суурилуулсан adapter-ууд)

Security Data Lake нь хэд хэдэн бэлэн data adapter-тэй ирдэг. Төрөл бүрийн баримт бичиг Edit Data Adapter маягт дээр байдаг.

• CSV File Adapter — статик CSV файлаас түлхүүр/утгын хайлт хийнэ.
• DNS Lookup Adapter — хост нэр болон IP хаягийг тайлна (A, AAAA, PTR, TXT бичлэг).
• DSV File Adapter — CSV-тэй төстэй боловч өөрийн тусгаарлагч, тохируулж болох түлхүүр/утгын багана дэмжинэ.
• HTTPS JSONPath Adapter — GET хүсэлт илгээж, JSONPath илэрхийллээр өгөгдөл гаргаж авна.
• Geo IP – MaxMind — MaxMind өгөгдлийн сангаар IP хаягийн газарзүйн мэдээлэл өгнө.

MongoDB

Security Data Lake нь MongoDB Data Adapter-ийг дэмждэг бөгөөд хайлтын өгөгдлийг Security Data Lake-ийн тохиргооны өгөгдлийн санд шууд хадгална. Бичлэгийг API, GUI эсвэл pipeline функцээр нэмэх, шинэчлэх, устгаж болно.

MongoDB Data Adapter-ийг API-аар удирдах

Түлхүүр нэмэх curl хүсэлтийн жишээ:

curl -u <token>:token \
-H 'X-Requested-By: cli' \
-H 'Accept: application/json' \
-H 'Content-Type: application/json' \
-X POST 'http://127.0.0.1:9000/api/plugins/org.graylog.plugins.lookup/lookup/adapters/mongodb/mongodb-data-name' \
--data-binary '{
    "key": "myIP",
    "values": ["12.34.42.99"],
    "data_adapter_id": "5e578606cdda4779dd9f2611"
  }'