YARA-L でマルチステージクエリを作成する

以下でサポートされています。

Google SecOps SIEM

このドキュメントでは、YARA-L のマルチステージクエリを使用して、あるクエリステージの出力を次のステージの入力に直接フィードする方法について説明します。このプロセスでは、単一のモノリシッククエリよりもデータ変換を細かく制御できます。

マルチステージクエリを既存の機能と統合する

マルチステージクエリは、Google Security Operations の次の既存の機能と連携して動作します。

複合検出ルール: マルチステージクエリは、複合検出ルールを補完します。複合ルールとは異なり、検索を使用するマルチステージクエリは結果をリアルタイムで返すことができます。
期間とマルチイベントルール: マルチステージクエリを使用して、データ内の異なる期間を比較することで異常を検出できます。たとえば、最初のクエリステージを使用して長期間にわたってベースラインを確立し、後のステージを使用してそのベースラインに対する最近のアクティビティを評価できます。マルチイベントルールを使用して、同様の比較を作成することもできます。

YARA-L のマルチステージクエリは、ダッシュボードと検索の両方でサポートされています。

結合は、複数のソースのデータを関連付けて、調査のコンテキストをより詳細に把握するのに役立ちます。関連するイベント、エンティティ、その他のデータをリンクすることで、複雑な攻撃シナリオを調査できます。詳細については、検索で結合を使用するをご覧ください。

マルチステージ YARA-L 構文を定義する

マルチステージクエリを構成する際は、次の点に注意してください。

制限ステージ: マルチステージクエリには、ルートステージに加えて、1 ～ 4 個の名前付きステージを含める必要があります。
順序の構文: 名前付きステージの構文は、常にルートステージの構文の前に定義します。

マルチステージ YARA-L クエリを作成する

マルチステージ YARA-L クエリを作成する手順は次のとおりです。

ステージの構造と構文

[調査] > [検索] に移動します。クエリステージを定義する場合は、次の構造要件に従ってください。

構文: 次の構文を使用して、各ステージに名前を付け、他のステージと区別します。

stage <stage name> { }

中かっこ: ステージ構文はすべて中かっこ {} で囲みます。
順序: ルートステージを定義する前に、名前付きのすべてのステージの構文を定義します。
参照: 各ステージは、クエリで前に定義されたステージを参照できます。
ルートステージ: クエリにはルートステージが必要です。ルートステージは、すべての名前付きステージの後に処理されます。

次のステージの例 daily_stats は、毎日のネットワーク統計情報を収集します。

stage daily_stats {
  metadata.event_type = "NETWORK_CONNECTION"
  $source = principal.hostname
  $target = target.ip
  $source != ""
  $target != ""
  $total_bytes = cast.as_int(network.sent_bytes + network.received_bytes)
  match:
    $source, $target by day
  outcome:
    $exchanged_bytes = sum($total_bytes)
}

アクセスステージの出力

名前付きステージの出力は、ステージフィールドを使用して後続のステージからアクセスできます。ステージフィールドは、ステージの match 変数と outcome 変数に対応しており、統合データモデル（UDM）フィールドと同様に使用できます。

ステージフィールドにアクセスするには、次の構文を使用します。

$<stage name>.<variable name>

アクセスウィンドウのタイムスタンプ（省略可）

名前付きステージでホップウィンドウ、スライディングウィンドウ、タンブリングウィンドウを使用する場合は、次の予約済みフィールドを使用して、各出力行のウィンドウの開始とウィンドウの終了にアクセスします。

$<stage name>.window_start
$<stage name>.window_end

window_start と window_end は、Unix エポックからの経過秒数で表される整数フィールドです。さまざまなステージのウィンドウのサイズは異なる場合があります。

制限事項

マルチステージクエリには、次の機能的および構造的な制約があります。

構造とステージの上限

ルートステージ: クエリごとに許可されるルートステージは 1 つだけです。
名前付きステージ: 最大 4 つの名前付きステージがサポートされています。
ステージ参照: ステージは、同じクエリ内で論理的に前に定義されたステージのみを参照できます。
結合: すべてのステージで、データテーブル以外の結合を最大 4 つ使用できます。
結果の要件: 名前付きの各ステージ（ルートステージを除く）には、match セクションまたは outcome セクションのいずれかを含める必要があります。outcome セクションには集計は必要ありません。

ウィンドウと互換性の上限

機能のサポート: マルチステージクエリは、検索とダッシュボードでサポートされていますが、ルールではサポートされていません。
ウィンドウタイプ: 1 つのクエリ内で異なるウィンドウタイプを混在させないでください。
ウィンドウの依存関係: ホップウィンドウまたはスライディングウィンドウを使用するステージは、ホップウィンドウまたはスライディングウィンドウを使用する別のステージに依存できません。
タンブリングウィンドウサイズ: さまざまなステージのタンブリングウィンドウのサイズは異なる場合がありますが、サイズの差は 720 倍未満でなければなりません。

例: ステージの集計の差

次のウィンドウ構成の例は許可されていません。

stage monthly_stats {
  metadata.event_type = "NETWORK_CONNECTION"
    $source = principal.hostname
    $target = target.ip
    $source != ""
    $target != ""
    $total_bytes = cast.as_int(network.sent_bytes + network.received_bytes)

  match:
    $source, $target by month

  outcome:
    $exchanged_bytes = sum($total_bytes)
}

$source = $monthly_stats.source
$target = $monthly_stats.target

match:
    $source, $target by minute

monthly_stats ステージでデータを月単位で集計し、ルートステージで monthly_stats の出力を分単位で集計する場合、monthly_stats の各行はルートステージの 43,200 行にマッピングされます（1 か月は 43,200 分であるため）。

ステージとクエリの制限事項

マルチステージクエリ内の各ステージには、次の制約があります。

単一ステージクエリに適用される制限のほとんどは、個々のステージにも適用されます。
- 出力要件: 各ステージで、少なくとも 1 つの一致変数または結果変数（ステージフィールド）を出力する必要があります。
- 結合のウィンドウ: 結合で使用される最大ウィンドウサイズ（ホップ、タンブリング、スライディング）は 2 日です。
- 結果変数の最大数:
  - 結果変数の上限を増やすことを選択していないお客様の場合は 20
  - 結果変数の上限を増やすことを選択したお客様は 50 個
- ホップウィンドウの最小サイズと最大サイズ。
- 配列値の結果変数内の要素の最大数。
マルチステージクエリには、統計情報クエリと同じ制限が適用されます。
- 統計クエリ: 120 QPH（API と UI）
- Google SecOps からのビューの検索: 1 分あたり 100 ビュー
- マルチステージ結合は、ユーザーインターフェースと EventService.UDMSearch API でサポートされていますが、SearchService.UDMSearch API ではサポートされていません。結合のないマルチステージクエリもユーザーインターフェースでサポートされています。

イベントとグローバルの制限事項

最大イベント数:

マルチステージクエリで同時に処理できるイベントの数には厳しい制限があります。

UDM イベント: 最大 2 つの UDM イベントを指定できます。
エンティティコンテキストグラフ（ECG）イベント: ECG イベントは 1 つまで許可されます。

グローバルクエリの制限事項:

これらの上限は、マルチステージクエリで返されるデータの範囲と量を制御するプラットフォーム全体の制約です。

クエリの期間の場合、標準クエリの最大期間は 30 日です。
結果セットの最大合計サイズは 10,000 件です。

マルチステージクエリの例

このセクションの例は、完全なマルチステージ YARA-L クエリを作成する方法を説明するのに役立ちます。

例: 異常にアクティブなネットワーク接続を検索する（時間）

このマルチステージ YARA-L の例では、ネットワークアクティビティが通常よりも高い IP アドレスのペアを特定します。3 時間以上高いアクティビティを維持しているペアをターゲットにします。クエリには、名前付きステージ hourly_stats と root ステージの 2 つの必須コンポーネントが含まれています。

hourly_stats ステージは、ネットワークアクティビティのレベルが高い principal.ip と target.ip のペアを検索します。

このステージでは、次のフィールドの 1 時間ごとの単一の値が返されます。

送信元 IP の統計情報（文字列）: $hourly_stats.src_ip
宛先 IP の統計情報（文字列）: $hourly_stats.dst_ip
イベント数の統計情報（整数）: $hourly_stats.count
受信バイト数の標準偏差（浮動小数点数）: $hourly_stats.std_recd_bytes
受信バイト数の平均（浮動小数点数）: $hourly_stats.avg_recd_bytes
Unix エポックからの秒単位の 1 時間単位の開始時間（整数）: $hourly_stats.window_start
Unix エポックからの秒単位の 1 時間単位の終了時間（整数）: $hourly_stats.window_end

ルートステージは hourly_stats ステージの出力を処理します。$hourly_stats で指定されたしきい値を超えるアクティビティを含む principal.ip と target.ip のペアの統計情報を計算します。次に、アクティビティの高い時間が 3 時間を超えるペアをフィルタします。


stage hourly_stats {
  metadata.event_type = "NETWORK_CONNECTION"
  $src_ip = principal.ip
  $dst_ip = target.ip
  $src_ip != ""
  $dst_ip != ""

  match:
    $src_ip, $dst_ip by hour

  outcome:
    $count = count(metadata.id)
    $avg_recd_bytes = avg(network.received_bytes)
    $std_recd_bytes = stddev(network.received_bytes)

  condition:
    $avg_recd_bytes > 100 and $std_recd_bytes > 50
}

$src_ip = $hourly_stats.src_ip
$dst_ip = $hourly_stats.dst_ip
$time_bucket_count = strings.concat(timestamp.get_timestamp($hourly_stats.window_start), "|", $hourly_stats.count)

match:
 $src_ip, $dst_ip

outcome:
 $list = array_distinct($time_bucket_count)
 $count = count_distinct($hourly_stats.window_start)

condition:
 $count > 3

ルートステージの一致条件を次のように変更すると、マルチステージクエリに日単位のウィンドウ集計を導入できます。

match:
 $src_ip, $dst_ip by day

例: 異常にアクティブなネットワーク接続を検索する（Z スコアを使用）

このマルチステージクエリは、Z スコアの計算（平均からの標準偏差の数を測定）を使用して、1 日の平均ネットワークアクティビティと今日のアクティビティを比較します。このクエリは、内部アセットと外部システム間の異常に高いネットワークアクティビティを効果的に検索します。

前提条件: 計算された Z スコアを有効にするには、クエリの時間枠が 2 日以上で、当日が含まれている必要があります。

このマルチステージクエリには、ネットワークアクティビティの Z スコアを計算するために連携して動作する daily_stats ステージと root ステージが含まれています。

daily_stats ステージは、毎日の初期集計を実行します。各 IP ペア（source と target）について、1 日に交換された合計バイト数を計算し、次のステージフィールド（出力行の列に対応）を返します。
- $daily_stats.source: 単数、文字列
- $daily_stats.target: 単数、文字列
- $daily_stats.exchanged_bytes: 単数、整数
- $daily_stats.window_start: 単数、整数
- $daily_stats.window_end: 単数、整数
ルートステージは、各 IP ペアの daily_stats ステージの出力を集計します。検索範囲全体で交換された 1 日あたりのバイト数の平均と標準偏差、および本日交換されたバイト数を計算します。これらの 3 つの計算値を使用して Z スコアを決定します。
出力には、今日すべての IP ペアの Z スコアが降順で表示されます。

// Calculate the total bytes exchanged per day by source and target

stage daily_stats {
  metadata.event_type = "NETWORK_CONNECTION"
  $source = principal.hostname
  $target = target.ip
  $source != ""
  $target != ""
  $total_bytes = cast.as_int(network.sent_bytes + network.received_bytes)
  match:
    $source, $target by day
  outcome:
    $exchanged_bytes = sum($total_bytes)
}

// Calculate the average per day over the time window and compare with the bytes
   exchanged today

$source = $daily_stats.source
$target = $daily_stats.target
$date = timestamp.get_date($daily_stats.window_start)

match:
  $source, $target

outcome:
  $today_bytes = sum(if($date = timestamp.get_date(timestamp.current_seconds()), $daily_stats.exchanged_bytes, 0))
  $average_bytes = window.avg($daily_stats.exchanged_bytes)
  $stddev_bytes = window.stddev($daily_stats.exchanged_bytes)
  $zscore = ($today_bytes - $average_bytes) / $stddev_bytes

order:
  $zscore desc

ステージから集計されていない変数をエクスポートする

名前付きステージには、集計されていない outcome セクションを含めることができます。つまり、その outcome セクション内で定義された変数はステージから直接出力されるため、後続のステージはグループ化された集計を必要とせずに、ステージフィールドとしてアクセスできます。

例: 集計されていない変数をエクスポートする

この例では、集計されていない変数をエクスポートする方法を示します。次のロジックに注意してください。

top_5_bytes_sent ステージでは、ネットワークアクティビティが最も高い 5 つのイベントを検索します。
top_5_bytes_sent ステージは、出力行の列に対応する次のステージフィールドを出力します。
- $top_5_bytes_sent.bytes_sent: 単数、整数
- $top_5_bytes_sent.timestamp_seconds: 単数、整数
root ステージは、ネットワークアクティビティが最も高い 5 つのイベントの最新のタイムスタンプと最も古いタイムスタンプを計算します。

stage top_5_bytes_sent {
  metadata.event_type = "NETWORK_CONNECTION"
  network.sent_bytes > 0

  outcome:
    $bytes_sent = cast.as_int(network.sent_bytes)
    $timestamp_seconds = metadata.event_timestamp.seconds

  order:
    $bytes_sent desc 
  
  limit:
    5
}

outcome:
  $latest_timestamp = timestamp.get_timestamp(max($top_5_bytes_sent.timestamp_seconds))
  $earliest_timestamp = timestamp.get_timestamp(min($top_5_bytes_sent.timestamp_seconds))

マルチステージクエリでウィンドウ処理を実装する

マルチステージクエリは、名前付きステージのすべてのタイプのウィンドウ処理（ホップ、スライディング、タンブリング）をサポートしています。名前付きステージにウィンドウが含まれている場合、次の予約済みフィールドを使用して、各出力行のウィンドウの開始と終了にアクセスできます。

$<stage name>.window_start
$<stage name>.window_end

例: ホップウィンドウ

次の例は、マルチステージクエリでホップウィンドウを使用する方法を示しています。

hourly_stats ステージでは、同じ時間内にネットワークアクティビティが高い IP ペアを検索します。
hourly_stats は、出力行の列に対応する次のステージフィールドを出力します。
- $hourly_stats.src_ip: 単数、文字列
- $hourly_stats.dst_ip: 単数、文字列
- $hourly_stats.count: 単数、整数
- $hourly_stats.std_recd_bytes: 単数、浮動小数点数
- $hourly_stats.avg_recd_bytes: 単数、浮動小数点数
- $hourly_stats.window_start: 単数、整数
- $hourly_stats.window_end: 単数、整数
ルートステージでは、アクティビティが 3 時間以上高い IP ペアが除外されます。hourly_stats ステージでホップウィンドウが使用されているため、時間が重複している可能性があります。

stage hourly_stats {
  metadata.event_type = "NETWORK_CONNECTION"
  $src_ip = principal.ip
  $dst_ip = target.ip
  $src_ip != ""
  $dst_ip != ""

  match:
    $src_ip, $dst_ip over 1h

  outcome:
    $count = count(metadata.id)
    $avg_recd_bytes = avg(network.received_bytes)
    $std_recd_bytes = stddev(network.received_bytes)

  condition:
    $avg_recd_bytes > 100 and $std_recd_bytes > 50
}

$src_ip = $hourly_stats.src_ip
$dst_ip = $hourly_stats.dst_ip
$time_bucket_count = strings.concat(timestamp.get_timestamp($hourly_stats.window_start), "|", $hourly_stats.count)

match:
 $src_ip, $dst_ip

outcome:
 $list = array_distinct($time_bucket_count)
 $count = count_distinct($hourly_stats.window_start)

condition:
 $count > 3

マルチステージクエリの結合

内部結合は、マルチステージクエリのステージ内およびステージ間でサポートされています。内部結合機能は、次の型をサポートしています。

UDM と UDM
UDM と ECG
UDM と DataTable

結合のコンテキストでは、ウィンドウ化されたステージは、ウィンドウを含む一致セクションがあるステージを指します。一方、テーブルステージはウィンドウを出力しません。

次の例は、マルチステージクエリで UDM イベントとテーブルステージ間のマッチレス結合を構成する方法を示しています。

median ステージでは、送信元ホストとターゲット IP のペアごとに送信されたバイト数の中央値が計算されます。
median ステージは、出力行の列に対応する次のステージフィールドを出力します。
- $median.host: 単数、文字列
- $median.target: 単数、文字列
- $median.median: 単数、浮動小数点数
absolute_deviations ステージは、同じ送信元ホストと宛先 IP のペアの median の行と各 UDM イベントを結合します。UDM イベントごとに、送信されたバイト数の絶対値を計算します。
absolute_deviations は、出力行の列に対応する次のステージフィールドを出力します。
- $absolute_deviations.host: 単数、文字列
- $absolute_deviations.target: 単数、文字列
- $absolute_deviations.absolute_deviation: 単数、浮動小数点数
ルートステージでは、すべての UDM イベントで送信されたバイト数の絶対偏差の平均を計算します。

stage median {
  metadata.event_type = "NETWORK_CONNECTION"
  $host = principal.hostname
  $target = target.ip

  match:
    $host, $target

  outcome:
    $median = window.median(network.sent_bytes, true)
}

stage absolute_deviations {
  metadata.event_type = "NETWORK_CONNECTION"
  $join_host = principal.hostname
  $join_host = $median.host
  $join_target = target.ip[0]
  $join_target = $median.target

  outcome:
    $host = $join_host
    $target = $join_target
    $absolute_deviation = math.abs(network.sent_bytes - $median.median)
}

$host = $absolute_deviations.host
$target = $absolute_deviations.target

match:
  $host, $target

outcome:
  $mean_absolute_deviation = avg($absolute_deviations.absolute_deviation)

例: ウィンドウステージとテーブルステージ間のマッチレス結合

次の例は、マルチステージクエリでウィンドウステージとテーブルステージ間のマッチレス結合を構成する方法を示しています。

hourly_stats ステージは、送信された合計バイト数を移行元と移行先のホストペアと時間バケットごとに計算します。
hourly_stats ステージは、出力行の列に対応する次のステージフィールドを出力します。
- $hourly_stats.source_host: 単数、文字列
- $hourly_stats.dst_host: 単数、文字列
- $hourly_stats.total_bytes_sent: 単数、浮動小数点数
- $hourly_stats.window_start: 単数、整数
- $hourly_stats.window_end: 単数、整数
agg_stats ステージは、各送信元ホストと宛先ホストのペアについて、1 時間あたりのバイト数の平均と標準偏差を計算します。
agg_stats は、出力行の列に対応する次のステージフィールドを出力します。
- $agg_stats.source_host: 単数、文字列
- $agg_stats.dst_host: 単数、文字列
- $agg_stats.avg_bytes_sent: 単数、浮動小数点数
- $agg_stats.stddev_bytes_sent: 単数、浮動小数点数
ルートステージは、hourly_stats の各行を、同じソースホストとターゲットホストのペアの agg_stats の行と結合します。ソースホストとターゲットホストのペアごとに、そのホストペアバケットの合計送信バイト数と集計統計情報を使用して z スコアを計算します。

stage hourly_stats {
 $source_host = principal.hostname
 $dst_host = target.hostname
 principal.hostname != ""
 target.hostname != ""
 match:
   $source_host, $dst_host by hour
 outcome:
   $total_bytes_sent = sum(network.sent_bytes)
}

stage agg_stats {
  $source_host = $hourly_stats.source_host
  $dst_host = $hourly_stats.dst_host
  match:
    $source_host, $dst_host
  outcome:
   $avg_bytes_sent = avg($hourly_stats.total_bytes_sent)
   $stddev_bytes_sent = stddev($hourly_stats.total_bytes_sent)
}

$source_host = $agg_stats.source_host
$source_host = $hourly_stats.source_host

$dst_host = $agg_stats.dst_host
$dst_host = $hourly_stats.dst_host

outcome:
  $hour_bucket = timestamp.get_timestamp($hourly_stats.window_start)
  $z_score = ($hourly_stats.total_bytes_sent - $agg_stats.avg_bytes_sent)/$agg_stats.stddev_bytes_sent

既知の問題

マルチステージクエリを実装する場合は、次の制限事項と推奨される回避策を確認することをおすすめします。

すべてのマルチステージクエリは、統計情報検索クエリのように動作します（出力は、集計されていないイベントやデータテーブルの行ではなく、集計された統計情報で構成されます）。
一方の側で UDM イベントとエンティティイベントを結合すると、データセットのサイズが原因でパフォーマンスが低下する可能性があります。結合の UDM イベントとエンティティイベントの側をできるだけフィルタすること（イベントタイプでフィルタするなど）を強くおすすめします。

推奨されるプラクティスの一般的なガイダンスについては、Yara-L のベストプラクティスをご覧ください。結合に固有の情報については、ベストプラクティスをご覧ください。

さらにサポートが必要な場合 コミュニティメンバーや Google SecOps のプロフェッショナルから回答を得ることができます。