2. كومة مرنة: تحليل سجلات الأمن. Logstash

في المقالة الأخيرة ، التقينا بكومة ELK عن منتجات البرمجيات التي تتكون منها. وتتمثل المهمة الأولى التي يواجهها المهندس عند العمل مع مكدس ELK في إرسال سجلات للتخزين في elasticsearch لمزيد من التحليل. ومع ذلك ، هذا فقط بالكلمات ، يقوم elasticsearch بتخزين السجلات في شكل مستندات تحتوي على حقول وقيم معينة ، مما يعني أن المهندس يجب أن يستخدم أدوات متنوعة لتحليل الرسالة التي يتم إرسالها من الأنظمة النهائية. يمكن القيام بذلك بعدة طرق - عن طريق كتابة البرنامج بنفسك ، والذي سيضيف مستندات إلى قاعدة البيانات باستخدام واجهة برمجة التطبيقات أو استخدام حلول جاهزة. في هذه الدورة التدريبية ، سننظر إلى حل Logstash ، وهو جزء من مكدس ELK. سنرى كيف يمكنك إرسال السجلات من الأنظمة الطرفية إلى Logstash ، ثم سنقوم بتكوين ملف التكوين للتحليل وإعادة التوجيه إلى قاعدة بيانات Elasticsearch. للقيام بذلك ، خذ السجلات من جدار الحماية Check Point كنظام وارد.

كجزء من الدورة التدريبية ، لا يتم اعتبار تثبيت مكدس ELK ، نظرًا لوجود عدد كبير من المقالات حول هذا الموضوع ، سننظر في مكون التكوين.

فلنضع خطة عمل لتكوين Logstash:

1. تحقق من أن elasticsearch سيقبل السجلات (التحقق من صحة الميناء والانفتاح).
2. نحن نفكر في كيفية إرسال الأحداث إلى Logstash واختيار الطريقة وتنفيذها.
3. تكوين الإدخال في ملف التكوين Logstash.
4. نقوم بتكوين الإخراج في ملف التكوين Logstash في وضع التصحيح لفهم كيف تبدو رسالة السجل.
5. تكوين مرشح.
6. تكوين الإخراج الصحيح في ElasticSearch.
7. تطلق Logstash.
8. تحقق من سجلات في Kibana.

لننظر في كل عنصر بمزيد من التفصيل:

التحقق من أن elasticsearch سيقبل السجلات

للقيام بذلك ، يمكنك استخدام الأمر curl للتحقق من الوصول إلى Elasticsearch من النظام الذي تم نشر Logstash عليه. إذا قمت بتكوين المصادقة ، ثم مرر المستخدم / كلمة المرور من خلال curl ، فحدد المنفذ 9200 إذا لم تقم بتغييره. إذا ظهرت إجابة مثل الإجابة أدناه ، فسيتم ترتيب كل شيء.

[elastic@elasticsearch ~]$ curl -u <<user_name>> : <<password>> -sS -XGET "<<ip_address_elasticsearch>>:9200" { "name" : "elastic-1", "cluster_name" : "project", "cluster_uuid" : "sQzjTTuCR8q4ZO6DrEis0A", "version" : { "number" : "7.4.1", "build_flavor" : "default", "build_type" : "rpm", "build_hash" : "fc0eeb6e2c25915d63d871d344e3d0b45ea0ea1e", "build_date" : "2019-10-22T17:16:35.176724Z", "build_snapshot" : false, "lucene_version" : "8.2.0", "minimum_wire_compatibility_version" : "6.8.0", "minimum_index_compatibility_version" : "6.0.0-beta1" }, "tagline" : "You Know, for Search" } [elastic@elasticsearch ~]$ 

إذا لم تتم الإجابة ، فيمكن أن يكون هناك عدة أنواع من الأخطاء: عملية elasticsearch لا تعمل ، أو المنفذ الخاطئ محدد ، أو المنفذ محظور بجدار حماية على الخادم حيث تم تثبيت elasticsearch.

فكر في كيفية إرسال السجلات إلى Logstash من جدار الحماية لنقطة التفتيش

باستخدام خادم إدارة Check Point ، يمكنك إرسال السجلات إلى Logstash عبر syslog باستخدام الأداة المساعدة log_exporter ، ويمكنك التعرف عليها في هذه المقالة بمزيد من التفصيل ، وهنا سنترك الأمر الذي ينشئ الدفق فقط:

cp_log_export إضافة اسم check_point_syslog الخادم المستهدف <<ip_address_logstash>> بروتوكول 5555 لتنسيق بروتوكول TCP المستهدف للمنافذ العامة وضع شبه موحد

<<ip_address_logstash>> - عنوان الخادم حيث يتم تشغيل Logstash ، الهدف 5555 - المنفذ الذي سنرسل إليه السجلات ، وإرسال السجلات عبر tcp يمكنه تحميل الخادم ، لذلك في بعض الحالات يكون ضبط udp أكثر صحة.

تكوين INPUT في ملف التكوين Logstash

بشكل افتراضي ، يوجد ملف التكوين في الدليل /etc/logstash/conf.d/. يتكون ملف التكوين من 3 أجزاء ذات معنى: INPUT ، FILTER ، OUTPUT. في INPUT ، نشير إلى أين سيأخذ النظام السجلات منه ، في FILTER نقوم بتحليل السجل - تكوين كيفية تقسيم الرسالة إلى حقول وقيم ، في OUTPUT نقوم بتكوين دفق الإخراج - حيث سيتم إرسال السجلات التي تم تحليلها.

أولاً ، قم بتكوين INPUT ، فكر في بعض الأنواع التي قد تكون - ملف و tcp و exe.

برنامج التعاون الفني:

 input { tcp { port => 5555 host => “10.10.1.205” type => "checkpoint" mode => "server" } } 

الوضع => "الخادم"
يقول أن Logstash يقبل الاتصالات.

ميناء => 5555
المضيف => "10.10.1.205"
نحن نقبل الاتصالات إلى عنوان IP 10.10.1.205 (Logstash) ، المنفذ 5555 - يجب السماح بالمنفذ بواسطة سياسة جدار الحماية.

اكتب => "نقطة تفتيش"
نقوم بتمييز المستند ، وهو مناسب جدًا إذا كان لديك العديد من الاتصالات الواردة. في ما يلي ، لكل اتصال ، يمكنك كتابة الفلتر الخاص بك باستخدام البناء المنطقي if.

ملف:

 input { file { path => "/var/log/openvas_report/*" type => "openvas" start_position => "beginning" } } 

وصف الإعدادات:
path => "/ var / log / openvas_report / *"
حدد الدليل الذي يجب أن تقرأ فيه الملفات.

اكتب => "openvas"
نوع الحدث.

start_position => "بداية"
عند تغيير الملف ، يقوم بقراءة الملف بالكامل ؛ إذا قمت بتعيين "النهاية" ، فإن النظام ينتظر ظهور إدخالات جديدة في نهاية الملف.

إكسيك:

 input { exec { command => "ls -alh" interval => 30 } } 

في هذا الإدخال ، (فقط!) يتم تشغيل أمر shell ويتم التفاف الإخراج الخاص به في رسالة سجل.

command => "ls -alh"
الفريق الذي يهمنا الانتاج.

الفاصل => 30
استدعاء الأوامر الفاصل في ثوان.

من أجل الحصول على سجلات من جدار الحماية ، نحن نصف مرشح tcp أو udp ، اعتمادًا على كيفية إرسال السجلات إلى Logstash.

نقوم بتكوين الإخراج في ملف التكوين Logstash في وضع التصحيح لفهم كيف تبدو رسالة السجل

بعد أن قمنا بتكوين INPUT ، نحتاج إلى فهم شكل رسالة السجل ، والأساليب التي يجب استخدامها لتكوين مرشح (محلل) السجلات.

للقيام بذلك ، سوف نستخدم مرشحًا يعرض النتيجة في stdout لعرض الرسالة الأصلية ، سيبدو ملف التكوين بالكامل في الوقت الحالي كما يلي:

 input { tcp { port => 5555 type => "checkpoint" mode => "server" host => “10.10.1.205” } } output { if [type] == "checkpoint" { stdout { codec=> json } } } 

يقوم بتشغيل الأمر للتحقق:
sudo / usr / share / logstash / bin // logstash -f /etc/logstash/conf.d/checkpoint.conf
نرى النتيجة ، الصورة قابلة للنقر:



إذا قمت بنسخه ، فسيبدو هكذا:

 action=\"Accept\" conn_direction=\"Internal\" contextnum=\"1\" ifdir=\"outbound\" ifname=\"bond1.101\" logid=\"0\" loguid=\"{0x5dfb8c13,0x5,0xfe0a0a0a,0xc0000000}\" origin=\"10.10.10.254\" originsicname=\"CN=ts-spb-cpgw-01,O=cp-spb-mgmt-01.tssolution.local.kncafb\" sequencenum=\"8\" time=\"1576766483\" version=\"5\" context_num=\"1\" dst=\"10.10.10.10\" dst_machine_name=\"ts-spb-dc-01@tssolution.local\" layer_name=\"TSS-Standard Security\" layer_name=\"TSS-Standard Application\" layer_uuid=\"dae7f01c-4c98-4c3a-a643-bfbb8fcf40f0\" layer_uuid=\"dbee3718-cf2f-4de0-8681-529cb75be9a6\" match_id=\"8\" match_id=\"33554431\" parent_rule=\"0\" parent_rule=\"0\" rule_action=\"Accept\" rule_action=\"Accept\" rule_name=\"Implicit Cleanup\" rule_uid=\"6dc2396f-9644-4546-8f32-95d98a3344e6\" product=\"VPN-1 & FireWall-1\" proto=\"17\" s_port=\"37317\" service=\"53\" service_id=\"domain-udp\" src=\"10.10.1.180\" ","type":"qqqqq","host":"10.10.10.250","@version":"1","port":50620}{"@timestamp":"2019-12-19T14:50:12.153Z","message":"time=\"1576766483\" action=\"Accept\" conn_direction=\"Internal\" contextnum=\"1\" ifdir=\"outbound\" ifname=\"bond1.101\" logid=\"0\" loguid=\"{0x5dfb8c13, 

بالنظر إلى هذه الرسائل ، نفهم أن السجلات لها النموذج: field = value أو key = value ، مما يعني أن المرشح الذي يسمى kv مناسب. من أجل اختيار المرشح المناسب لكل حالة محددة ، سيكون من الجيد أن تتعرف عليها في الوثائق التقنية ، أو تسأل صديقًا.

تخصيص مرشح

في المرحلة الأخيرة ، اختاروا kv ، ثم يتم تقديم تكوين هذا الفلتر:

 filter { if [type] == "checkpoint"{ kv { value_split => "=" allow_duplicate_values => false } } } 

نختار الرمز الذي نقسم به الحقل والقيمة - "=". إذا كان لدينا نفس الإدخالات في السجل ، فإننا نحفظ مثيلًا واحدًا فقط في قاعدة البيانات ، وإلا فسوف تحصل على مجموعة من نفس القيم ، أي إذا تلقينا الرسالة "foo = some foo = some" ، فاكتب فقط foo = some.

تكوين الإخراج الصحيح في ElasticSearch

بعد تكوين الفلتر ، يمكنك تحميل السجلات إلى قاعدة بيانات elasticsearch :

 output { if [type] == "checkpoint" { elasticsearch { hosts => ["10.10.1.200:9200"] index => "checkpoint-%{+YYYY.MM.dd}" user => "tssolution" password => "cool" } } } 

إذا تم توقيع المستند بنوع نقطة التفتيش ، احفظ الحدث في قاعدة بيانات elasticsearch ، التي تقبل الاتصالات في 10.10.1.200 إلى المنفذ 9200 بشكل افتراضي. يتم حفظ كل وثيقة في فهرس معين ، في هذه الحالة نقوم بحفظ في "نقطة تفتيش" الفهرس + + التاريخ الحالي. يمكن أن يحتوي كل فهرس على مجموعة محددة من الحقول ، أو يتم إنشاؤه تلقائيًا عندما يظهر حقل جديد في الرسالة ، ويمكن عرض إعدادات الحقل ونوعها في التعيينات.

إذا كنت قد قمت بتكوين المصادقة (سننظر لاحقًا) ، فيجب تحديد أرصدة الكتابة إلى فهرس معين ، في هذا المثال تكون "tssolution" مع كلمة المرور "cool". يمكنك التمييز بين حقوق المستخدم في كتابة السجلات إلى فهرس معين فقط وليس أكثر.

إطلاق Logstash.

ملف تكوين Logstash:

 input { tcp { port => 5555 type => "checkpoint" mode => "server" host => “10.10.1.205” } } filter { if [type] == "checkpoint"{ kv { value_split => "=" allow_duplicate_values => false } } } output { if [type] == "checkpoint" { elasticsearch { hosts => ["10.10.1.200:9200"] index => "checkpoint-%{+YYYY.MM.dd}" user => "tssolution" password => "cool" } } } 

تحقق من ملف التكوين للتجميع الصحيح:
/ usr / share / logstash / bin // logstash -f checkpoint.conf


نبدأ عملية Logstash:
sudo systemctl بدء logstash

تحقق من أن العملية قد بدأت:
sudo systemctl حالة السجل



تحقق مما إذا كان المقبس قد ارتفع:
netstat-nat | grep 5555

تحقق من سجلات في Kibana.

بعد بدء كل شيء ، انتقل إلى Kibana - اكتشف ، تأكد من أن كل شيء تم تكوينه بشكل صحيح ، فإن الصورة قابلة للنقر!



جميع السجلات في مكانها ويمكننا أن نرى كل الحقول وقيمها!

استنتاج

درسنا كيفية كتابة ملف تكوين Logstash ، ونتيجة لذلك حصلنا على محلل لجميع الحقول والقيم. الآن يمكننا العمل مع البحث والتخطيط لحقول معينة. علاوة على ذلك ، سننظر في التصور في Kibana ، وإنشاء لوحة معلومات بسيطة. تجدر الإشارة إلى أنه يجب إضافة ملف تكوين Logstash باستمرار في بعض المواقف ، على سبيل المثال ، عندما نريد استبدال قيمة الحقل من رقم إلى كلمة. في المقالات اللاحقة ، سنفعل ذلك طوال الوقت.

لذا ترقبوا ( Telegram ، Facebook ، VK ، TS Solution Blog )، Yandex.Zen .