Apache Cassandra のログを収集する
このドキュメントでは、Bindplane を使用して Apache Cassandra ログを Google Security Operations に取り込む方法について説明します。パーサーはフィールドを抽出し、統合データモデル(UDM)に変換します。Grok パターンを使用して初期メッセージを解析し、ネストされたデータに JSON フィルタを使用して、条件付き変換を実行してさまざまなフィールドを UDM の同等のフィールドにマッピングし、さまざまなログレベルを処理して、メタデータで出力を拡充します。
始める前に
次の前提条件を満たしていることを確認します。
- Google SecOps インスタンス
- Windows 2016 以降、または systemd を使用する Linux ホスト
- プロキシの背後で実行されている場合、ファイアウォール ポートが開いている
- Apache Cassandra インスタンスへの特権アクセス
Google SecOps の取り込み認証ファイルを取得する
- Google SecOps コンソールにログインします。
- [SIEM 設定] > [コレクション エージェント] に移動します。
- Ingestion Authentication File をダウンロードします。Bindplane をインストールするシステムにファイルを安全に保存します。
Google SecOps のお客様 ID を取得する
- Google SecOps コンソールにログインします。
- [SIEM 設定] > [プロファイル] に移動します。
- [組織の詳細情報] セクションから [お客様 ID] をコピーして保存します。
Bindplane エージェントをインストールする
Windows のインストール
- 管理者として コマンド プロンプトまたは PowerShell を開きます。
次のコマンドを実行します。
msiexec /i `https://github.com/observIQ/bindplane-agent/releases/latest/download/observiq-otel-collector.msi` /quiet
Linux のインストール
- root 権限または sudo 権限でターミナルを開きます。
次のコマンドを実行します。
sudo sh -c `$(curl -fsSlL https://github.com/observiq/bindplane-agent/releases/latest/download/install_unix.sh)` install_unix.sh
その他のインストール リソース
その他のインストール オプションについては、インストール ガイドをご覧ください。
Syslog を取り込んで Google SecOps に送信するように Bindplane エージェントを構成する
- 構成ファイルにアクセスします。
config.yaml
ファイルを見つけます。通常、Linux では/etc/bindplane-agent/
ディレクトリに、Windows ではインストール ディレクトリにあります。- テキスト エディタ(
nano
、vi
、メモ帳など)を使用してファイルを開きます。
config.yaml
ファイルを次のように編集します。receivers: udplog: # Replace the port and IP address as required listen_address: `0.0.0.0:514` exporters: chronicle/chronicle_w_labels: compression: gzip # Adjust the path to the credentials file you downloaded in Step 1 creds: '/path/to/ingestion-authentication-file.json' # Replace with your actual customer ID from Step 2 customer_id: <customer_id> endpoint: malachiteingestion-pa.googleapis.com # Add optional ingestion labels for better organization ingestion_labels: log_type: 'CASSANDRA' raw_log_field: body service: pipelines: logs/source0__chronicle_w_labels-0: receivers: - udplog exporters: - chronicle/chronicle_w_labels
自社のインフラストラクチャでの必要性に応じて、ポートと IP アドレスを置き換えます。
<customer_id>
は、実際の顧客 ID に置き換えます。/path/to/ingestion-authentication-file.json
の値を、Google SecOps の取り込み認証ファイルを取得するで認証ファイルを保存したパスに更新します。
Bindplane エージェントを再起動して変更を適用する
Linux で Bindplane エージェントを再起動するには、次のコマンドを実行します。
sudo systemctl restart bindplane-agent
Windows で Bindplane エージェントを再起動するには、Services コンソールを使用するか、次のコマンドを入力します。
net stop BindPlaneAgent && net start BindPlaneAgent
Apache Cassandra で Syslog エクスポートを構成する
- SSH を使用して Apache Cassandra ホストにログインします。
- 構成ファイル
logback.xml
を開き、28 行目に次のコードを挿入します。- ほとんどのバージョンの Apache Cassandra では、場所は
$(CASSANDRA_HOME)/conf
になります。 - Datastax Enterprise のパッケージ インストールの場合、場所は
/etc/dse
になります。 - DSE の tar ファイル インストールの場合、場所は
$(TARBALL_ROOT)/resources/cassandra/conf
になります。
- ほとんどのバージョンの Apache Cassandra では、場所は
次の Appender 定義を
logback.xml
ファイルの 28 行目に追加します。<appender name="SYSLOG" class="ch.qos.logback.classic.net.SyslogAppender"> <syslogHost>bindplane-ip</syslogHost> <port>bindplane-port</port> <facility>LOCAL7</facility> <throwableExcluded>true</throwableExcluded> <suffixPattern>%thread:%level:%logger{36}:%msg</suffixPattern> </appender>
bindplane-ip
とbindplane-port
は、実際の Bindplane エージェントの IP アドレスとポートに置き換えます。logback.xml
ファイルのルート ロガー ブロック<root level=
INFO>
に次のコードを追加します。この行が挿入される場所は、Apache Cassandra のバージョンによって異なります。
- Apache Cassandra 5.0.x、123 行目。
- Apache Cassandra 4.0.x および 4.1.x、115 行目。
- Apache Cassandra 3.11.x および 3.0.x、92 行目。
- Datastax Enterprise(すべてのバージョン)、121 行目。
<appender-ref ref=`SYSLOG` />
UDM マッピング テーブル
ログフィールド | UDM マッピング | ロジック |
---|---|---|
agent.ephemeral_id |
observer.labels.value |
内部 JSON メッセージの agent.ephemeral_id の値。 |
agent.hostname |
observer.hostname |
内部 JSON メッセージの agent.hostname の値。 |
agent.id |
observer.asset_id |
filebeat: と内部 JSON メッセージの agent.id の値の連結。 |
agent.name |
observer.user.userid |
内部 JSON メッセージの agent.name の値。 |
agent.type |
observer.application |
内部 JSON メッセージの agent.type の値。 |
agent.version |
observer.platform_version |
内部 JSON メッセージの agent.version の値。 |
cloud.availability_zone |
principal.cloud.availability_zone |
内部 JSON メッセージの cloud.availability_zone の値。 |
cloud.instance.id |
principal.resource.product_object_id |
内部 JSON メッセージの cloud.instance.id の値。 |
cloud.instance.name |
principal.resource.name |
内部 JSON メッセージの cloud.instance.name の値。 |
cloud.machine.type |
principal.resource.attribute.labels.value |
内部 JSON メッセージの cloud.machine.type の値。対応する key は machine_type です。 |
cloud.provider |
principal.resource.attribute.labels.value |
内部 JSON メッセージの cloud.provider の値。対応する key は provider です。 |
event_metadata._id |
metadata.product_log_id |
内部 JSON メッセージの event_metadata._id の値。 |
event_metadata.version |
metadata.product_version |
内部 JSON メッセージの event_metadata.version の値。 |
host.architecture |
target.asset.hardware.cpu_platform |
内部 JSON メッセージの host.architecture の値。 |
host.fqdn |
target.administrative_domain |
内部 JSON メッセージの host.fqdn の値。 |
host.hostname |
target.hostname |
内部 JSON メッセージの host.hostname の値。 |
host.id |
target.asset.asset_id |
Host Id: と内部 JSON メッセージの host.id の値の連結。 |
host.ip |
target.asset.ip |
内部 JSON メッセージの host.ip からの IP アドレスの配列。 |
host.mac |
target.mac |
内部 JSON メッセージの host.mac からの MAC アドレスの配列。 |
host.os.kernel |
target.platform_patch_level |
内部 JSON メッセージの host.os.kernel の値。 |
host.os.platform |
target.platform |
host.os.platform が debian の場合、LINUX に設定します。 |
host.os.version |
target.platform_version |
内部 JSON メッセージの host.os.version の値。 |
hostname |
principal.hostname |
grok を使用して message フィールドから抽出された hostname の値。 |
key |
security_result.detection_fields.value |
grok を使用して message フィールドから抽出された key の値。対応する key は key です。 |
log.file.path |
principal.process.file.full_path |
内部 JSON メッセージの log.file.path の値。 |
log_level |
security_result.severity |
log_level の値に基づいてマッピングされます。DEBUG 、INFO 、AUDIT は INFORMATIONAL にマッピングされ、ERROR は ERROR にマッピングされ、WARNING は MEDIUM にマッピングされます。 |
log_level |
security_result.severity_details |
grok を使用して message フィールドから抽出された log_level の値。 |
log_type |
metadata.log_type |
未加工ログの log_type の値。 |
message |
security_result.description |
grok を使用して message フィールドから抽出された説明。 |
message |
target.process.command_line |
コマンドラインは、grok を使用して message フィールドから抽出されます。 |
now |
security_result.detection_fields.value |
grok を使用して message フィールドから抽出された now の値。対応する key は now です。Grok を使用して message フィールドから抽出された event_time フィールドから解析されます。hostname と host.hostname の両方が存在する場合は USER_RESOURCE_ACCESS に設定し、それ以外の場合は GENERIC_EVENT に設定します。CASSANDRA に設定します。CASSANDRA に設定します。ephemeral_id に設定します。cloud.instance.name が存在する場合は VIRTUAL_MACHINE に設定します。対応する検出フィールドに対して key と now に設定します。 |
timestamp |
timestamp |
未加工ログの create_time フィールドから取得されます。 |
さらにサポートが必要な場合 コミュニティ メンバーや Google SecOps のプロフェッショナルから回答を得ることができます。