Универсальный шаблон Grok для парсинга syslog Kaspersky Linux Mail Server
Наконец-то официальной антиспам и антивирус системой на моём предприятии стал Kaspersky Linux Mail Server (KLMS). Мы с коллегами прошли долгий путь, начиная с бесплатных решений Amavis + ClamAV + SpamAssassin, через платный DrWeb AV и вот после бета-теста официально купленный KLMS. Наш MTA Postfix по протоколу milter отдаёт письма для проверки и ожидает вердикта. За работой и результатами проверок писем можно наблюдать различными способами: кто-то, но не мы, может использовать отдельный единый Kaspersky Security Center (KSC) в виде оснастки на MS Windows Server; есть удобный веб-интерфейс; записи syslog, которые отправляются у нас на единый remote syslog сервер, созданный на базе ElasticSearch + Logstash + Kibana (ELK). В данной статье речь как раз и пойдёт о данном методе.
Исходные данные
Кто сражается с чудовищами, тому следует остерегаться, чтобы самому при этом не стать чудовищем. И если ты долго смотришь в бездну, то бездна тоже смотрит в тебя.
Фридрих Ницше
Если на сервере KLMS последить за событиями через tail -f /var/log/syslog, то можно наблюдать подобные строки:
-
Входящее письмо с нарушением и с вирусом.
not clean: message-id="<bad@site.my>": relay-ip="64.188.23.22": action="Rejected, Put in backup": backup-reason="AntiSpam": rules="1": size=270066: mail-from="evil@site.my": rcpt-to="info@firma.ru": kt-status="NotScanned, disabled by settings", av-status="Infected", ap-status="Clean", as-status="Spam", as-method="multiple", ma-status="NotScanned, policy", cf-status="Clean": part "Quote.xlsx", av-status="Infected", threats="HEUR:Exploit.MSOffice.CVE-2018-0802.gen" -
Входящее письмо с фишингом.
not clean: message-id="<82A909@ph.ishing.com>": relay-ip="45.137.2.7": action="Rejected, Put in backup": backup-reason="AntiSpam": rules="1": size=8217: mail-from="support@ph.ishing.com": rcpt-to="petroff@firma.ru": kt-status="NotScanned, disabled by settings", av-status="Clean", ap-status="Phishing (KSN)", as-status="Spam", as-method="Shikari Terms [main]", ma-status="NotScanned, policy", cf-status="Clean" -
Работа правила DenyList (BlackList).
not processed: message-id="<666@from.hell>": relay-ip="1.6.6.6": kt-reason="blacklist", av-reason="blacklist", ap-reason="blacklist", as-reason="blacklist", ma-reason="blacklist", cf-reason="blacklist": action="Rejected": rules="2": size=1111: mail-from="goat@from.hell": rcpt-to="angel@firma.ru" -
Помещение в карантин.
put to asp quarantine: message-id="<mass-7ee7@govno-site.ru>": relay-ip="1.33.111.7": action="Postponed": size=15269: mail-from="trash@govno-site.ru": rcpt-to="boss@firma.ru" -
Чистое входящее письмо.
clean: message-id="<b314@goodsite.ru>": relay-ip="19.64.16.18": action="Skipped": rules="1": size=18110: mail-from="orlova@goodsite.ru": rcpt-to="ivanoff@firma.ru": kt-status="NotScanned, disabled by settings", av-status="Clean", ap-status="Clean", as-status="Clean", ma-status="ViolationNotFound", dmarc="Pass", spf="Pass", dkim="Pass", cf-status="Clean" -
Чистое входящее письмо с проверкой лишь на вирусы от дружественного домена из списка AllowList (WhiteList).
clean: message-id="<d4@friend.ru>": relay-ip="21.12.98.14": action="Skipped": rules="3": size=777089: mail-from="manager@friend.ru": rcpt-to="sidoroff@firma.ru": kt-status="NotScanned, disabled by settings", av-status="Clean", ap-status="NotScanned, disabled by settings", as-status="NotScanned, disabled by settings", ma-status="NotScanned, disabled by settings", cf-status="NotScanned, disabled by settings"
Если отправить данные события в remote syslog, то они превратятся в документы с дефолтными полями syslog в виде timestamp, program, facility, priority, severity (и т.д) и ВСЕМ неразобранным текстом события в поле message. В таком режиме удобно лишь хранить и читать поступающую информацию глазами, но аналитику делать почти невозможно. Прибывающую информацию (поле message) необходимо разобрать-распарсить и вычленить части, превратив их в поля (fields), чтобы легко можно делать различные запросы на языке Kibana Query Language (KQL) или Lucene, рисовать графики, строить дашборды и т.д. Этим ELK'а сама не занимается, а жаль 
.
Удлиняем syslog
По умолчанию длина сообщения syslog ограничена 2048 байтами. Вы сможете на практике встретить ситуацию, когда (ваш|интернет) пользователь отправит письмо оооооооочень большому количеству адресатов и вся итоговая строка станет неприлично большой и потеряет часть полей (обычно впереди идущих). Подстрока message-id сама по себе часто длинная в символах и вносит свою лепту в итоговую длину.
Чтобы увеличить максимальную длину сообщения добавьте на сервере KLMS в файл sudo -e /etc/rsyslog.conf
строку $MaxMessageSize 64k
И перезапустите службу sudo systemctl restart rsyslog
Пляшущие человечки поля
Если наблюдать за строками syslog от KMLS, то видны вполне конкретные поля, которые то появляются то исчезают при различных событиях. Так как KLMS это система для проверки почты на различные угрозы, то прежде нужно отчётливо понимать какие именно компоненты-модули составляют его:
- AV - AntiVirus. Модуль проверки на вирусы.
- AS - AntiSpam. Модуль проверки на спам.
- AP - AntiPhishing. Модуль проверки на фишинг.
- MA - Mail Sender Authentication. Модуль проверки подлинности отправителя.
- CF - Content Filtering. Контентная фильтрация сообщений.
- KT - Kaspersky Anti Targeted Attack. Анализатор целевых атак. Это отдельный компонент, который необходимо приобрести за дополнительную плату и установить на отдельный сервер. Сразу признаюсь что данного модуля у нас нет.
Добавляя мысленно слова статус (status), метод (method), причина (reason), становится легко сходу читать и разбирать события своей головой. Осталось научить этому LogStash через его grok в разделе filter.
Регулярное выражение grok pattern
Если у вас есть проблема и вы захотели решить её с помощью регулярных выражений, то теперь у вас две проблемы.
А теперь приготовьтесь! Нас ждёт ооооочень длинное регулярное выражение, которое учитывает исчезающие поля.
Во-первых, для упрощения понимания регулярка будет разбита на логические части, хотя представляет собой непрерывную строку символов без единого пробела.
Во-вторых, необходимо использовать синтаксис регулярных выражений Онигурума, который позволяет найти нужное в искомой строке и сохранить подстроку в именованном поле. В общем виде это выглядит как (?<field_name>the pattern here). Весь синтаксис можете изучить на ГитХабе проекта.
В-третьих, вам стоит познакомиться с инструментом отладки в Kibana из раздела Dev Tools - Management - Grok Debugger, который поможет создать и проверить (ваше|моё) регулярное выражение (Grok Pattern) на разбор конкретной строки (Sample Data).
В-четвёртых, KLMS много чего пишет в результате своей работы, например про обновление своих баз. Нам же необходимы строки-вердикты проверки почты, поэтому в итоговом KLMS.conf для LogStash в конце статьи вы увидите конструкции, допускающие к grok только строки содержащие message-id. Этого достаточно, чтобы гарантировать разбор только нужных строк-событий-писем.
В-пятых, дефис (-) заменяется на подчёркивание (_), чтобы с одной стороны показать связь (подстрока av-status сохраняется в поле av_status), а с другой стороны - облегчить чтение регулярки.
Поехали!
(?<status_letter>%{DATA}):%{SPACE}?
Будут прибывать строки типа
- Помещение в карантин: put to asp quarantine: .....
- Чистое письмо: clean: .....
- С любым нарушением: not clean: .....
- Необработанное модулями, так как отброшено вашими DenyList правилами: not processed: .....
Благодаря вышеописанной части регулярки, всё до первого двоеточия попадёт и станет полем status_letter документа.
message-id="(?<message_id>%{DATA}?)":%{SPACE}?
message-id (идентификатор сообщения) обязательно присутствующая подстрока, она как путеводная звезда есть всегда, поэтому пользуемся этим и сохраняем значение (value) после знака равно в поле message_id. Пишем (?<message_id>%{DATA}?), вместо (?<message_id>%{DATA}), чтобы знаком вопроса ? в конце сказать что может быть пустота в значении. Бывает что от спамеров поступает строка вида message-id="":
relay-ip="(?<relay_ip>%{IP})":%{SPACE}?
Вторая обязательная подстрока relay-ip (IP адрес отправителя). Делаем по аналогии с message-id, но без знака вопроса ? так как маловероятно её отсутствие, учитывая сетевую природу почтового протокола SMTP. Используем менее жадный %{IP} по сравнению с %{DATA}, чтобы подсказать регулярке что искать. Более подробно на ГитХабе grok-patterns.
(?<t>(?:kt-reason="(?<kt_reason>%{DATA}?)",)?)%{SPACE}?
(?<t>(?:av-reason="(?<av_reason>%{DATA}?)",)?)%{SPACE}?
(?<t>(?:ap-reason="(?<ap_reason>%{DATA}?)",)?)%{SPACE}?
(?<t>(?:as-reason="(?<as_reason>%{DATA}?)",)?)%{SPACE}?
(?<t>(?:ma-reason="(?<ma_reason>%{DATA}?)",)?)%{SPACE}?
(?<t>(?:cf-reason="(?<cf_reason>%{DATA}?)":)?)%{SPACE}?
В зыбком болоте твёрдые кочки закончились и начались исчезающие поля, на которые нужно аккуратно наступать. Подстроки *-reason (причина от конкретного модуля) будут появляться в сообщениях при работе запрещающих правил DenyList.
Пример:
not processed: message-id="<666@from.hell>": relay-ip="1.6.6.6": kt-reason="blacklist", av-reason="blacklist", ap-reason="blacklist", as-reason="blacklist", ma-reason="blacklist", cf-reason="blacklist": action="Rejected": rules="2": size=1111: mail-from="goat@from.hell": rcpt-to="angel@firma.ru"
В других видах сообщений между подстрокой relay-ip и action нет ничего, поэтому нужно через синтаксис Онигурума описать что может отсутствовать целиком вся подстрока. Для этого вводим временное поле t и пишем конструкцию вида (?<t>(?:ma-reason="(?<ma_reason>%{DATA}?)",)?)
Поле t поможет не раз, но потом мы уберём его как лишнее через механизм
mutate {
remove_field => [ "t" ]
}
action="(?<action>%{DATA})":%{SPACE}?
Снова твёрдая земля. Есть возможность опереться на обязательную подстроку action, в которой KLMS сохранит нам свой вердикт.
(?<t>(?:backup-reason="(?<backup_reason>%{DATA}?)":)?)%{SPACE}?
Необязательная подстрока backup-reason появляется лишь тогда, когда вы в правилах просите дополнительно сохранить письмо в Хранилище.
Примерные части:
action="Skipped, Put in backup": backup-reason="AntiSpam":
action="Skipped, Put in backup": backup-reason="MessageAuthentication":
action="Rejected, Put in backup": backup-reason="AntiSpam":
(?<t>(?:rules="(?<rules>%{INT}?)":)?)%{SPACE}?
Необязательная подстрока rules (номер правила) исчезает при помещении письма в Карантин (put to asp quarantine). Правила работают сверху вниз, но номер правила возрастает снизу вверх. Дефолтное правило, которое нельзя удалить, имеет номер 1. Менее жадный %{INT} подскажет найти число.
size=(?<size>%{INT}?):%{SPACE}?
Обязательная подстрока size (размер письма) снова помогает своим присутствием. Менее жадный %{INT} подскажет найти число.
mail-from="(?<mail_from>%{DATA}?)":%{SPACE}?
Обязательная подстрока mail-from (от кого письмо) не проста как кажется. Приходится писать %{DATA} вместо желательной менее жадной подсказки EMAILADDRESS, так как проект LogStash ложно определяет регулярки вот так
EMAILLOCALPART [a-zA-Z][a-zA-Z0-9_.+-=:]+
EMAILADDRESS %{EMAILLOCALPART}@%{HOSTNAME}
по их мнению емайл обязан начинаться с буквы, хотя это не так. Так же стоит помнить что по стандарту RFC 5321 существует пустой емайл MAIL FROM:<> (NULL SENDER), который используется при уведомлении отправителя о доставке почты, избегая при этом петель (email loop).
В готовом конфигурационном файле вы найдёте конструкцию вида
if ![mail_from] {
mutate {
replace => {"mail_from" => "NULL_SENDER_RFC5321" }
}
}
rcpt-to="(?<rcpt_to>%{DATA}?)"
Обязательная подстрока rcpt-to (кому письма) может содержать несколько адресатов. Приходится пользоваться жадной %{DATA}, чтобы она съела нужное. В данном месте специалист по реляционным базам данных поймёт без слов что на лицо нарушение Первой Нормальной Формы. В одном поле находится не скалярное значение, а массив. Но, к сожалению, это обычная практика в мире ELK, так как тут нет классического множества таблиц, связанных с друг другом ключами. Для аналитики поиск нужного емайла в таком поле на языке KQL придётся делать так - rcpt_to:"*rezume@firma.ru*"
(?<t>$|:%{SPACE})
Очень важная конструкция-барьер. Выражение останавливает жадный захват %{DATA} из части rcpt-to, устанавливая "барьер" в виде конец строки ($) ИЛИ (|) есть подстрока двоеточие-пробел (:%{SPACE})
Некоторые строки заканчиваются подстрокой rcpt-to.
Работа правила DenyList (BlackList).
not processed: ... rcpt-to="angel@firma.ru"
Помещение в карантин.
put to asp quarantine: ... rcpt-to="boss@firma.ru"
Остальные строки продолжаются.
Чистое письмо.
clean: ... rcpt-to="ivanoff@firma.ru": kt-status="NotScanned, disabled by settings", ...
(?<t>(?:kt-status="(?<kt_status>%{DATA}?)",)?)%{SPACE}?
(?<t>(?:av-status="(?<av_status>%{DATA}?)",)?)%{SPACE}?
(?<t>(?:ap-status="(?<ap_status>%{DATA}?)",)?)%{SPACE}?
(?<t>(?:as-status="(?<as_status>%{DATA}?)",)?)%{SPACE}?
(?<t>(?:as-method="(?<as_method>%{DATA}?)",)?)%{SPACE}?
(?<t>(?:ma-status="(?<ma_status>%{DATA}?)",)?)%{SPACE}?
В штатном режиме письмо проходит череду проверок всеми модулями KLMS и каждый заполняет статус проверки. Подстрок может не быть, если письмо блокируется запрещающими правилами или попадает в карантин.
(?<t>(?:dmarc="(?<dmarc>%{DATA}?)",)?)%{SPACE}?
(?<t>(?:spf="(?<spf>%{DATA}?)",)?)%{SPACE}?
(?<t>(?:dkim="(?<dkim>%{DATA}?)",)?)%{SPACE}?
Три братца: DMARC (Domain-based Message Authentication, Reporting and Conformance), SPF (Sender Policy Framework) и DKIM (DomainKeys Identified Mail) появляются после работы модуля MA (Mail Sender Authentication, Модуль проверки подлинности отправителя).
(?<t>(?:cf-status="(?<cf_status>%{DATA}?)")?)%{SPACE}?
Статус контентной фильтрации попадает в поле cf_status из подстроки cf-status.
(?<part>%{GREEDYDATA}?)
Если письмо с зашифрованным вложением, которое естественно не распаковать, или с вирусом попадёт на проверку, то, возможно, появится подстрока part, которая сама по себе представляет массив значений на каждый элемент, так как вложений и вирусов может быть много.
Простые примеры:
: part "Смета2022.xlsx", av-status="Encrypted"
: part "PaymentDetails.xlsx", av-status="Infected", threats="HEUR:Exploit.MSOffice.CVE-2018-0802.gen, UDS:DangerousObject.Multi.Generic, UDS:DangerousObject.Multi.Generic"
Сложный пример:
: part "REQUEST_FOR_QUOTE_RFQ.doc", av-status="Infected", threats="HEUR:Exploit.MSOffice.Generic, HEUR:Exploit.MSOffice.Generic, UDS:DangerousObject.Multi.Generic": part "PO_467889999087746346.doc", av-status="Infected", threats="HEUR:Exploit.MSOffice.Generic, HEUR:Exploit.MSOffice.Generic, UDS:DangerousObject.Multi.Generic"
Мне не остаётся ничего другого как жадно съесть строку до конца всей строки с помощью GREEDYDATA (выражение .*), мысленно поблагодарив компьютерных богов что подстрока part последняя. В part, к сожалению, всё: и статусы отдельных файлов (av-status) и имена угроз (threats) и новый блок part. Ситуация - смешались в кучу кони, люди.
В оправдание скажу что уже ранее в поле rcpt-to пихали вместо скаляра массив. И я просто не знаю как следует поступать в рамках ElasticSearch, где документ (document) это SQL аналог понятия строки-кортежа (row). Где поле (field) это SQL аналог колонки (column) и есть по факту один лишь индекс (аналог SQL таблицы).
Конфигурационный файл для Logstash
Весь конфигурационный файл KLMS.conf доступен в виде архива. Разъясню лишь пару моментов:
- В разделе input специально используется несуществующий тип type => klmslog, чтобы в message получить чисто сообщение (извините за тавтологию). Если использовать тип = syslog, то парсить нужно будет всю приходящую строку, начиная с timestamp.
-
if [type] == "klmslog" and [program] == "KLMS" and [message] =~ "message-id"
Конструкция допускает внутрь ветки if только лишь сообщения от программы KLMS и строки, имеющие подстроку message-id. Её наличие гарантирует для grok'а что парсить придётся именно сообщение о почтовой SMTP сессии, а не рядовое событие самого KLMS.
- После парсинга и заполнения данными нужных полей, мне не нужна сама исходная строка в поле message и временное поле t. Удаляем remove_field => [ "t", "message" ]
- Если всё-таки что-то не учли и наш grok pattern не может распарсить строку, то автоматически будет добавлен тег grok parse failure. Скидываем такие события в отдельный файл. Мониторим появление и содержимое таких файлов и пытаемся проанализировать ситуацию.
if "_grokparsefailure" in [tags] {
file {
path => "/var/log/logstash/fail-KLMS-%{+YYYY.MM.dd}.log"
}
}
Итог
Если всё получилось настроить, то можно вывести необходимые вам поля для контроля ситуации. Теперь можно легко и просто наблюдать ход работы KLMS и оперативно реагировать на инциденты. Используя мощь KQL, разыскивать необходимое событие, строить аналитические отчёты, графики и т.д.
Изображение можно открыть в отдельной вкладке в оригинальном размере.
Представленное решение вполне рабочее и особо радует успешное создание универсального регулярного выражения, хоть и очень сложного. Самостоятельно, если есть желание, на базе relay_ip можете реализовать дополнительные хотелки-поля: получить FQDN (если есть), заполнить GeoIP и т.д.
Но меня заинтересовала возможность Kaspersky Linux Mail Server слать сообщения формата Common Event Format (CEF) в системы Security information and event management (SIEM). А Logstash тоже умеет играть в CEF. Но это уже будет другая история в другой статье. До встречи!
Дополнительные материалы:
LogStash и привилегированные порты.
Заметки про FileBeat из стека ELK.
Журналы Postfix в ElasticSearch.
