2. エラスティックスタック: セキュリティログの分析。 ログスタッシュ

過去に статье 私たちは会った ELKスタック、どのようなソフトウェア製品で構成されているか。 そして、エンジニアが ELK スタックを操作するときに直面する最初のタスクは、その後の分析のために Elasticsearch に保存されるログを送信することです。 ただし、これは単なるリップサービスであり、elasticsearch は特定のフィールドと値を含むドキュメントの形式でログを保存します。つまり、エンジニアはエンド システムから送信されるメッセージを解析するためにさまざまなツールを使用する必要があります。 これにはいくつかの方法があります。API を使用してデータベースにドキュメントを追加するプログラムを自分で作成するか、既製のソリューションを使用します。 このコースではその解決策を検討します ログスタッシュ、これは ELK スタックの一部です。 エンドポイント システムから Logstash にログを送信する方法を見てから、解析して Elasticsearch データベースにリダイレクトするための構成ファイルをセットアップします。 これを行うために、受信システムとして Check Point ファイアウォールからログを取得します。

このトピックに関する膨大な数の記事があるため、このコースでは ELK スタックのインストールについては説明しません。構成コンポーネントについて検討します。

Logstash 構成のアクション プランを作成しましょう。

1. elasticsearch がログを受け入れることを確認します (ポートの機能とオープン性を確認します)。
2. Logstash にイベントを送信する方法を検討し、方法を選択して実装します。
3. Logstash 構成ファイルで入力を構成します。
4. ログ メッセージがどのようなものかを理解するために、デバッグ モードで Logstash 構成ファイルの出力を構成します。
5. フィルターの設定。
6. ElasticSearch での正しい出力の設定。
7. Logstash が起動します。
8. Kibana でログを確認します。

各ポイントをさらに詳しく見てみましょう。

elasticsearch がログを受け入れることを確認する

これを行うには、curl コマンドを使用して、Logstash がデプロイされているシステムから Elasticsearch へのアクセスを確認します。 認証が設定されている場合は、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 管理サーバーから、log_exporter ユーティリティを使用して、syslog 経由でログを Logstash に送信できます。詳細については、こちらをご覧ください。 статье, ここでは、ストリームを作成するコマンドのみを残しておきます。

cp_log_export 追加名 check_point_syslog ターゲットサーバー < > ターゲットポート 5555 プロトコル TCP フォーマット汎用読み取りモード半統合

< > - Logstash が実行されるサーバーのアドレス、ターゲット ポート 5555 - ログの送信先ポート、tcp 経由でログを送信するとサーバーに負荷がかかる可能性があるため、場合によっては udp を使用する方が正しい場合があります。

Logstash 構成ファイルでの INPUT のセットアップ

デフォルトでは、構成ファイルは /etc/logstash/conf.d/ ディレクトリにあります。 設定ファイルは、INPUT、FILTER、OUTPUT という 3 つの意味のある部分で構成されます。 で 入力 システムがどこからログを取得するかを指定します。 フィルタ ログを解析します - メッセージをフィールドと値に分割する方法を設定します。 出力 解析されたログが送信される出力ストリームを構成します。

まず、INPUT を設定しましょう。ファイル、TCP、および exe などのタイプを検討してください。

TCP:

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"
    }
}

設定の説明:
パス => "/var/log/openvas_report/*"
ファイルを読み取る必要があるディレクトリを指定します。

「openvas」と入力します
イベントタイプ。

start_position => "始まり"
ファイルを変更する場合、ファイル全体が読み取られます。「end」を設定すると、システムはファイルの最後に新しいレコードが表示されるまで待機します。

実行:

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

この入力を使用して、(のみ!) シェル コマンドが起動され、その出力がログ メッセージに変換されます。

コマンド => "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="[email protected]" 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
	}
}
}

フィールドと値を分割するための記号「=」を選択します。 ログに同一のエントリがある場合は、データベースにインスタンスを XNUMX つだけ保存します。そうでない場合は、同一の値の配列が得られることになります。つまり、「foo = some foo=some」というメッセージがある場合、foo のみを書き込みます。 = いくつか。

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 での接続を受け入れます。 各ドキュメントは特定のインデックスに保存されます。この場合、インデックス「checkpoint-」 + 現在の日付に保存されます。 各インデックスには特定のフィールド セットを含めることができ、メッセージに新しいフィールドが表示されたときに自動的に作成されます。フィールド設定とそのタイプはマッピングで表示できます。

認証を構成している場合 (後で説明します)、特定のインデックスに書き込むための資格情報を指定する必要があります。この例では、パスワードが「cool」の「tssolution」です。 特定のインデックスにのみログを書き込み、それ以上は書き込まないようにユーザー権限を区別できます。

ログスタッシュを起動します。

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 start logstash

プロセスが開始されたことを確認します。
sudo systemctlステータスlogstash

ソケットが稼働しているかどうかを確認してみましょう。
netstat -nat |grep 5555

Kibana でログを確認します。

すべてが実行されたら、Kibana - Discover に移動し、すべてが正しく構成されていること、画像をクリックできることを確認します。

すべてのログが適切な場所にあり、すべてのフィールドとその値を確認できます。

まとめ

Logstash 構成ファイルの作成方法を検討した結果、すべてのフィールドと値のパーサーを取得しました。 これで、特定のフィールドの検索とプロットを行うことができます。 コースの次は、Kibana での視覚化を見て、簡単なダッシュボードを作成します。 Logstash 構成ファイルは、フィールドの値を数値から単語に置き換える場合など、特定の状況では常に更新する必要があることに注意してください。 今後の記事では、これを継続的に実行していきます。

乞うご期待Telegram, Facebook, VK, TSソリューションブログ), Yandex.Den.

出所： habr.com