Mehrstufige Abfragen erstellen

Unterstützt in:

In diesem Dokument wird beschrieben, wie Sie mit mehrstufigen Abfragen in YARA-L die Ausgabe einer Abfragephase direkt in die Eingabe einer nachfolgenden Phase einfügen können. So haben Sie mehr Kontrolle über die Datentransformation als bei einer einzelnen, monolithischen Abfrage.

Mehrstufige Anfragen in bestehende Funktionen einbinden

Mehrstufige Abfragen funktionieren in Verbindung mit den folgenden vorhandenen Funktionen in Google Security Operations:

  • Zusammengesetzte Erkennungsregeln: Mehrstufige Abfragen ergänzen zusammengesetzte Erkennungsregeln, indem sie die Lücke zwischen automatisierter Erkennung und aktiver Untersuchung schließen. Während sich zusammengesetzte Regeln hervorragend eignen, um komplexe Korrelationen mit mehreren Ereignissen über längere Zeiträume hinweg zu erkennen, ermöglichen es Analysten, mit mehrstufigen Abfragen von diesen Erkennungen zu Echtzeit- und iterativen Suchvorgängen zu wechseln, um eine sich entwickelnde Bedrohung sofort zu validieren und einzugrenzen.

  • Zeiträume und Regeln für mehrere Ereignisse:Mit mehrstufigen Abfragen können Sie Anomalien erkennen, indem Sie verschiedene Zeiträume in Ihren Daten vergleichen. Sie können beispielsweise mit Ihren ersten Abfragen über einen längeren Zeitraum eine Baseline erstellen und dann mit späteren Abfragen die aktuelle Aktivität im Vergleich zu dieser Baseline auswerten. Sie können auch Regeln mit mehreren Ereignissen verwenden, um einen ähnlichen Vergleich zu erstellen.

Mehrstufige Abfragen in YARA-L werden sowohl in Dashboards als auch in Search unterstützt.

Mit Joins lassen sich Daten aus mehreren Quellen korrelieren, um mehr Kontext für eine Untersuchung zu erhalten. Durch das Verknüpfen von zusammengehörigen Ereignissen, Entitäten und anderen Daten können Sie komplexe Angriffsszenarien untersuchen. Weitere Informationen finden Sie unter Joins in Search verwenden.

Wichtige Überlegungen

Beachten Sie beim Konfigurieren einer mehrstufigen Abfrage Folgendes:

  • Limit-Stufe: Mehrstufige Abfragen müssen zusätzlich zur Root-Stufe zwischen einer und vier benannten Stufen enthalten.
  • Syntax für die Reihenfolge: Definieren Sie immer zuerst die Syntax für die benannte Phase und dann die Syntax für die Stammphase.

Wir empfehlen Ihnen, die folgenden bekannten Probleme und empfohlenen Problemumgehungen zu berücksichtigen, wenn Sie mehrstufige Abfragen implementieren:

  • Alle mehrstufigen Abfragen verhalten sich wie Statistik-Suchanfragen. Die Ausgabe besteht aus aggregierten Statistiken und nicht aus nicht aggregierten Ereignissen oder Datentabellenzeilen.

  • Die Leistung von Joins mit UDM- und Entitätsereignissen auf einer Seite kann aufgrund der Größe dieses Datasets gering sein. Wir empfehlen dringend, die UDM- und Entitätsereignisse auf der Join-Seite so weit wie möglich zu filtern (z. B. nach Ereignistyp).

Allgemeine Informationen zu empfohlenen Vorgehensweisen finden Sie unter Best Practices für YARA-L 2.0. Informationen speziell zu Joins finden Sie unter Best Practices.

Schlüsselterminologie

Im Kontext von Joins bezieht sich eine Fensterphase auf eine Phase mit einem match-Abschnitt, der ein Fenster enthält. Im Gegensatz dazu werden in einer Tabellenphase keine Fenster ausgegeben.

Mehrstufige YARA-L-Abfrage erstellen

So erstellen Sie eine mehrstufige YARA-L-Abfrage:

Struktur und Syntax von Phasen

Klicken Sie auf Prüfung > Suche. Halten Sie sich an diese Struktur, wenn Sie Ihre Abfragephasen definieren:

Syntax: Verwenden Sie die folgende Syntax, um jede Phase zu benennen und von anderen Phasen zu trennen:

stage <stage name> { }

  • Geschweifte Klammern: Setzen Sie die gesamte Phasen-Syntax in geschweifte Klammern {}.

  • Reihenfolge: Definieren Sie die Syntax für alle benannten Phasen, bevor Sie die Root-Phase definieren.

  • Verweise: In jeder Phase kann auf Phasen verwiesen werden, die zuvor in der Abfrage definiert wurden.

  • Stammphase: Eine Abfrage muss eine Stammphase haben, die nach allen benannten Phasen verarbeitet wird.

In der folgenden Beispielphase daily_stats werden tägliche Netzwerkstatistiken erfasst:

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

Ausgabe der Zugriffsphase

Die Ausgabe einer benannten Phase ist für nachfolgende Phasen über Phasenfelder zugänglich. Die Staging-Felder entsprechen den Variablen match und outcome der Phase und können ähnlich wie Felder des einheitlichen Datenmodells (Unified Data Model, UDM) verwendet werden.

Verwenden Sie die folgende Syntax, um auf ein Phasenfeld zuzugreifen:

$<stage name>.<variable name>

Optional: Zeitstempel für den Zugriff

Wenn in einer benannten Phase ein Hop-, gleitendes oder rollierendes Fenster verwendet wird, können Sie mit diesen reservierten Feldern auf den Fensteranfang und das Fensterende für jede Ausgabezeile zugreifen:

  • $<stage name>.window_start

  • $<stage name>.window_end

window_start und window_end sind Ganzzahlfelder, die in Sekunden seit der Unix-Epoche angegeben werden. Fenster in verschiedenen Phasen können unterschiedlich groß sein.

Beispiele für mehrstufige Abfragen

Die Beispiele in diesem Abschnitt veranschaulichen, wie Sie eine vollständige mehrstufige YARA-L-Abfrage erstellen können.

Beispiel: Nach ungewöhnlich aktiven Netzwerkverbindungen suchen (Stunden)

In diesem mehrstufigen YARA-L-Beispiel werden IP-Adressenpaare mit überdurchschnittlicher Netzwerkaktivität identifiziert. Dabei werden Paare berücksichtigt, die über einen Zeitraum von mehr als drei Stunden eine hohe Aktivität aufweisen. Die Abfrage enthält zwei erforderliche Komponenten: die benannte Phase hourly_stats und die Phase root.

In der hourly_stats-Phase wird nach principal.ip- und target.ip-Paaren mit hoher Netzwerkaktivität gesucht.

In dieser Phase wird ein einzelner stündlicher Wert für die folgenden Felder zurückgegeben:

  • Statistiken für die Quell-IP-Adresse (String): $hourly_stats.src_ip

  • Statistiken für die Ziel-IP-Adresse (String): $hourly_stats.dst_ip

  • Statistiken für die Anzahl der Ereignisse (Ganzzahl): $hourly_stats.count

  • Standardabweichung der empfangenen Byte (Gleitkommazahl): $hourly_stats.std_recd_bytes

  • Durchschnittlich empfangene Byte (Gleitkommazahl): $hourly_stats.avg_recd_bytes

  • Startzeit des Stundenbereichs in Sekunden seit der Unix-Epoche (Ganzzahl): $hourly_stats.window_start

  • Endzeit des Stundenbereichs in Sekunden seit der Unix-Epoche (Ganzzahl): $hourly_stats.window_end

In der Root-Phase wird die Ausgabe der Phase hourly_stats verarbeitet. Es werden Statistiken für principal.ip- und target.ip-Paare mit Aktivitäten berechnet, die den von $hourly_stats angegebenen Grenzwert überschreiten. Anschließend wird nach Paaren mit mehr als drei Stunden hoher Aktivität gefiltert.


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

Wenn Sie die Abgleichsbedingung in der Root-Phase wie folgt ändern, können Sie eine fensterbasierte Aggregation nach Tag für die mehrstufige Abfrage einführen.

match:
 $src_ip, $dst_ip by day

Beispiel: Suche nach ungewöhnlich aktiven Netzwerkverbindungen (mit Z-Score)

Bei dieser mehrstufigen Abfrage wird die durchschnittliche tägliche Netzwerkaktivität mit der heutigen Aktivität verglichen. Dazu wird ein Z-Score berechnet, der die Anzahl der Standardabweichungen vom Mittelwert angibt. Mit dieser Abfrage wird effektiv nach ungewöhnlich hoher Netzwerkaktivität zwischen internen Assets und externen Systemen gesucht.

Voraussetzung: Der Abfragezeitraum muss mindestens zwei Tage umfassen und den aktuellen Tag einschließen, damit der berechnete Z-Score wirksam ist.

Diese mehrstufige Abfrage umfasst die Phase daily_stats und die Phase root, die zusammen den Z-Score für die Netzwerkaktivität berechnen:

  • In der Phase daily_stats wird die erste tägliche Zusammenfassung durchgeführt. Es werden die insgesamt ausgetauschten Byte für jedes IP-Paar (source und target) pro Tag berechnet und die folgenden Stufenfelder zurückgegeben (entsprechend den Spalten in den Ausgaberow):

    • $daily_stats.source: Singular, String
    • $daily_stats.target: Singular, String
    • $daily_stats.exchanged_bytes: Singular, Ganzzahl
    • $daily_stats.window_start: Singular, Ganzzahl
    • $daily_stats.window_end: Singular, Ganzzahl

  • In der Root-Phase wird die Ausgabe der daily_stats-Phase für jedes IP-Paar zusammengefasst. Es werden der Durchschnitt und die Standardabweichung der täglich ausgetauschten Bytes für den gesamten Suchbereich sowie die heute ausgetauschten Bytes berechnet. Anhand dieser drei berechneten Werte wird der Z-Score ermittelt.

  • In der Ausgabe werden die Z-Scores für alle IP-Paare des aktuellen Tages in absteigender Reihenfolge aufgeführt.

// 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 != ""

  match:
    $source, $target by day

  outcome:
    $exchanged_bytes = sum(network.sent_bytes + network.received_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()), cast.as_int($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

Nicht aggregierte Variablen aus Phasen exportieren

Bei einer typischen mehrstufigen Abfrage werden Daten häufig durch einen Aggregationsprozess zwischen den Phasen übergeben, wodurch mehrere Ereignisse in einer einzigen Zusammenfassung zusammengefasst werden können. Es gibt jedoch Szenarien, in denen Sie die spezifischen Details jedes Ereignisses beibehalten müssen, z. B. eine eindeutige Prozess-ID oder eine bestimmte Befehlszeile, ohne die Granularität zu verlieren. Dazu können Sie Variablen direkt ohne Gruppierungsfunktion exportieren.

Benannte Phasen können einen nicht aggregierten outcome-Abschnitt enthalten. Das bedeutet, dass Variablen, die in diesem outcome-Abschnitt definiert sind, direkt aus der Phase ausgegeben werden. Nachfolgende Phasen können als Phasenfelder darauf zugreifen, ohne dass eine gruppierte Aggregation erforderlich ist.

Diese Detailinformationen sind aus folgenden Gründen nützlich:

  • Datenintegrität wird beibehalten: Sie können die genauen Attribute eines Ereignisses (z. B. einen bestimmten Dateipfad) an die nächste Phase übergeben, ohne eine künstliche „Platzhalter“-Aggregation wie max() oder array_distinct() verwenden zu müssen.
  • Weniger komplexe Abfragen: Die YARA-L-Logik wird vereinfacht, da kein match-Abschnitt oder Gruppieren von Anweisungen erforderlich ist, um einen Wert von Phase 1 zu Phase 2 zu übertragen.
  • Leistung optimieren: Wenn die Aggregations-Engine umgangen wird, kann das System Daten effizienter zwischen den Phasen streamen. Das führt zu kürzeren Ausführungszeiten für komplexe, umfangreiche Suchvorgänge.

Beispiel: Nicht aggregierte Variable exportieren

In diesem Beispiel wird gezeigt, wie nicht aggregierte Variablen exportiert werden. Beachten Sie die folgende Logik:

  • In der Phase top_5_bytes_sent wird nach den fünf Ereignissen mit der höchsten Netzwerkaktivität gesucht.

  • In der Phase top_5_bytes_sent werden die folgenden Phasenfelder ausgegeben, die den Spalten in den Ausgabezeilen entsprechen:

    • $top_5_bytes_sent.bytes_sent: Singular, Ganzzahl
    • $top_5_bytes_sent.timestamp_seconds: Singular, Ganzzahl

  • In der Phase root werden die neuesten und frühesten Zeitstempel für die fünf Ereignisse mit der höchsten Netzwerkaktivität berechnet.

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

  outcome:
    $bytes_sent = 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))

Fensterfunktionen in mehrstufigen Abfragen implementieren

Bei mehrstufigen Erkennungen können Sie mit der Fensterung bestimmte Zeitgrenzen für die Ereigniskorrelation innerhalb einer Phase definieren. Durch das Partitionieren von Daten in diskrete Zeiträume, z. B. ein gleitendes 5‑Minuten-Fenster oder ein 1‑Stunden-Fenster, können Sie Muster wie Brute-Force-Angriffe oder Beaconing-Verhalten erkennen, die erst bei der Analyse als zeitgebundene Sequenz sichtbar werden.

Mehrstufige Abfragen unterstützen alle Arten von Windowing (Hop, Sliding und Tumbling) in benannten Stufen. So können Sie zeitbezogene Daten zwischen Phasen übergeben, z. B. Phase 1 verwenden, um ein Fenster mit Ereignissen mit hoher Häufigkeit zu identifizieren, und Phase 2, um dieses bestimmte Fenster mit nachfolgenden administrativen Maßnahmen in Beziehung zu setzen.

Wenn eine benannte Phase ein Fenster enthält, sind der Fensterstart und das Fensterende für jede Ausgaberow über die folgenden reservierten Felder verfügbar:

  • $stage_window_start: Der Unix-Zeitstempel, der den Beginn des Zeitraums markiert.
  • $stage_window_end: Der Unix-Zeitstempel, der das Ende des Zeitraums markiert.

Weitere Informationen zum Windowing finden Sie unter YARA-L 2.0-Windowing-Logik.

Gängige Anwendungsfälle

  • Sequenzielle Erkennung: Nach Ablauf des bestimmten Zeitraums eines erkannten Anstiegs fehlgeschlagener Anmeldungen wird in einer zweiten Phase nach einer kurz darauf erfolgten erfolgreichen Anmeldung gesucht.
  • Daueranalyse: Berechnung der „Time-to-Compromise“ (Zeit bis zum Eindringen) durch Vergleich der $stage_window_start einer ersten Exploiting-Phase mit den Zeitstempeln von Ereignissen in einer späteren Phase.
  • Historische Baseline: Hier werden Fenster verwendet, um die aktuellen Ereigniszahlen mit den Ausgabevariablen eines vorherigen Fensters zu vergleichen.

Beispiel: Hop-Window

Das folgende Beispiel veranschaulicht, wie Sie Hop-Fenster in einer mehrstufigen Abfrage verwenden können:

  • In der hourly_stats-Phase wird nach IP-Paaren gesucht, die innerhalb derselben Stunde eine hohe Netzwerkaktivität aufweisen.

  • hourly_stats gibt die folgenden Phasenfelder aus, die den Spalten in den Ausgabezeilen entsprechen:

    • $hourly_stats.src_ip: Singular, String
    • $hourly_stats.dst_ip: Singular, String
    • $hourly_stats.count: Singular, Ganzzahl
    • $hourly_stats.std_recd_bytes: Singular, Gleitkommazahl
    • $hourly_stats.avg_recd_bytes: Singular, Gleitkommazahl
    • $hourly_stats.window_start: Singular, Ganzzahl
    • $hourly_stats.window_end: Singular, Ganzzahl

  • Im Root-Abschnitt werden IP-Paare mit mehr als 3 Stunden hoher Aktivität herausgefiltert. Die Stunden können sich aufgrund der Verwendung eines Hop-Fensters in der hourly_stats-Phase überschneiden.

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

Inner Joins in mehrstufigen Abfragen

Mit Inner Joins können Sie Daten aus verschiedenen Phasen oder Quelltypen in Beziehung setzen. So lassen sich komplexe Analyse-Workflows erstellen, z. B. um Echtzeitereignisse mit vorab berechneten statistischen Baselines zu vergleichen. Durch das Verknüpfen von Phasen können Sie Rohdaten aus der Telemetrie mit zustandsbezogenen Daten (z. B. Medianen oder Nachschlagetabellen) anreichern, um Ausreißer oder Multi-Vektor-Bedrohungen zu erkennen, die bei einem Eventfilter mit nur einer Phase nicht erkannt würden.

Inner Joins werden innerhalb und zwischen den Phasen von mehrstufigen Abfragen unterstützt. Die Inner Join-Funktion unterstützt die folgenden Typen:

  • UDM und UDM: Zwei verschiedene Gruppen von Sicherheitsereignissen korrelieren, z. B. ein Anmeldeereignis mit einem nachfolgenden Dateizugriff.
  • UDM und ECG: Ereignisdaten mit Informationen aus dem Entity Context Graph zusammenführen, um Identitäten oder Assets anzureichern.
  • UDM und Datentabelle: Live-Ereignisse mit statischen oder hochgeladenen Referenzlisten verknüpfen, z. B. einer Liste mit hochwertigen Assets oder abteilungsspezifischen IP-Bereichen.

Im folgenden Beispiel wird gezeigt, wie ein Join ohne Entsprechung (ein Join, der im Abschnitt outcome oder events anstelle des Abschnitts match ausgeführt wird) zwischen UDM-Ereignissen und einer berechneten Tabellenphase konfiguriert wird. Mit diesem Muster können Sie die statistische Anomalieerkennung durchführen, wie in der folgenden Berechnung der mittleren absoluten Abweichung (Mean Absolute Deviation, MAD) gezeigt wird:

  • median-Phase: Berechnet die mediane Anzahl der gesendeten Byte für jedes Paar aus Quellhost und Ziel-IP-Adresse.
    • $median.host: Singular, String
    • $median.target: Singular, String
    • $median.median: Singular, Gleitkommazahl
  • absolute_deviations-Phase: Hier wird jedes UDM-Ereignis mit der entsprechenden Zeile aus der Median-Phase verknüpft. So können Sie die absolute Abweichung der gesendeten Byte für jedes einzelne Ereignis im Vergleich zur entsprechenden Gruppe berechnen.
    • $absolute_deviations.host: Singular, String
    • $absolute_deviations.target: Singular, String
    • $absolute_deviations.absolute_deviation: Singular, Gleitkommazahl
  • Stammphase: Der Mittelwert dieser absoluten Abweichungen wird für alle UDM-Ereignisse berechnet, um einen Anomalieschwellenwert festzulegen.

Beispiel: Join ohne Übereinstimmung konfigurieren

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)

Beispiel: Join ohne Übereinstimmung zwischen Fenster- und Tabellenphase

Im folgenden Beispiel wird veranschaulicht, wie Sie einen Join ohne Entsprechung zwischen einer Window-Phase und einer Tabellenphase in einer mehrphasigen Abfrage konfigurieren.

  • In der Phase hourly_stats werden die insgesamt gesendeten Byte für jedes Quell- und Zielhostpaar und jeden Stunden-Bucket berechnet.
  • In der Phase hourly_stats werden die folgenden Phasenfelder ausgegeben, die den Spalten in den Ausgabezeilen entsprechen:
    • $hourly_stats.source_host: Singular, String
    • $hourly_stats.dst_host: Singular, String
    • $hourly_stats.total_bytes_sent: Singular, Gleitkommazahl
    • $hourly_stats.window_start: Singular, Ganzzahl
    • $hourly_stats.window_end: Singular, Ganzzahl
  • In der Phase agg_stats werden der Durchschnitt und die Standardabweichung der Byte pro Stunde für jedes Quell- und Zielhostpaar berechnet.

  • agg_stats gibt die folgenden Phasenfelder aus, die den Spalten in den Ausgabezeilen entsprechen:

    • $agg_stats.source_host: Singular, String
    • $agg_stats.dst_host: Singular, String
    • $agg_stats.avg_bytes_sent: Singular, Gleitkommazahl
    • $agg_stats.stddev_bytes_sent: Singular, Gleitkommazahl

  • In der Root-Phase wird jede Zeile aus hourly_stats mit der Zeile aus agg_stats für dasselbe Quell- und Zielhostpaar zusammengeführt. Für jedes Quell- und Zielhostpaar wird der Z-Score anhand der insgesamt gesendeten Byte für diesen Hostpaar-Bucket und der aggregierten Statistiken berechnet.

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

Cross-Joins in mehrstufigen Abfragen

Wenn Sie Google SecOps Search oder Dashboards verwenden, können Sie mit Cross-Joins in mehrstufigen Abfragen einzelne UDM-Ereignisdaten mit aggregierten Statistiken vergleichen, die in anderen YARA-L-Phasen berechnet wurden.

In YARA-L funktioniert das Schlüsselwort cross join mit einer Phase, die nur eine Zeile zurückgibt.

Wenn ein Cross Join zwischen einer Phase mit einem Limit von 1 und einem anderen Dataset (z. B. UDM-Ereignisse) verwendet wird, wird die einzelne Zeile der Ausgabe der Phase an jede Zeile des anderen Datasets angehängt. Dadurch werden die Ereignisdaten mit den Gesamtstatistiken angereichert.

Beispiel: Ungewöhnliche Anmeldeaktivitäten finden

Im folgenden Beispiel werden die Nutzer identifiziert, die sich häufiger als normal anmelden. Dazu wird die Anzahl der Logins jedes Nutzers (mit der user_login_counts-Phase) mit der durchschnittlichen Anzahl der Logins aller Nutzer (mit der total_users-Phase) verglichen. Nutzer, die sich ungewöhnlich oft anmelden, können in den Suchergebnissen sortiert werden.

Anschließend verwenden Sie das Keyword „cross join“, um die Ergebnisse aus der total_users-Phase mit den Ergebnissen aus der user_login_counts-Phase zu verknüpfen.

stage user_login_counts {
    $user = principal.user.userid
    metadata.event_type = "USER_LOGIN"
    security_result.action = "ALLOW"

    match:
        $user

    outcome:
        $login_count = count(metadata.id)
}

stage total_users {
    outcome:
        $count = count($user_login_counts.user)
    limit:
        1
}

cross join $total_users, $user_login_counts

$login_count = $user_login_counts.login_count
$user = $user_login_counts.user
$tot_users = $total_users.count

// all users who logged in the same number of times are grouped together.
match:
    $login_count
outcome:
    $num_users = count($user)
    $frequency_percent = (count($user) / max($tot_users) ) * 100

Beschränkungen

Für mehrstufige Abfragen gelten die folgenden funktionalen und strukturellen Einschränkungen:

Strukturelle Anforderungen

Beim Erstellen von Abfragen müssen Sie die folgenden strukturellen Anforderungen beachten:

  • Root-Phase: Pro Abfrage ist nur eine Root-Phase zulässig.
  • Benannte Phasen: Es werden maximal vier benannte Phasen unterstützt.
  • Referenzierung von Phasen: Eine Phase kann nur auf Phasen verweisen, die in derselben Abfrage logisch vor ihr definiert sind.
  • Cross Joins: Ein Cross Join kann nur auf eine Phase verweisen, die eine einzelne Zeile zurückgibt. Sie müssen in der referenzierten Phase ein Limit (1) angeben, um diese Anforderung zu erfüllen. Das ist nützlich, weil Sie so jeder einzelnen Ereigniszeile zum Vergleich eine globale Statistik (z. B. ein Maximum oder einen Durchschnitt) hinzufügen können.
  • Joins: In allen Phasen sind maximal vier Joins ohne Datentabelle zulässig.
  • Anforderung an das Ergebnis: Jede benannte Phase (mit Ausnahme der Stammphase) muss entweder einen match-Abschnitt oder einen outcome-Abschnitt enthalten, wobei für den outcome-Abschnitt keine Aggregation erforderlich ist.

Fenster- und Kompatibilitätslimits

Für die Verwendung von Fenstern und die Ausführung von Abfragen gelten die folgenden Einschränkungen:

  • Unterstützte Funktionen: Mehrstufige Abfragen funktionieren in Search und Dashboards, werden aber in Regeln nicht unterstützt.
  • Fenstertypen: Vermeiden Sie es, verschiedene Fenstertypen in einer einzelnen Abfrage zu mischen.
  • Fensterabhängigkeit: Eine Phase, in der ein Hop oder ein gleitendes Fenster verwendet wird, kann nicht von einer anderen Phase abhängen, in der ebenfalls ein Hop oder ein gleitendes Fenster verwendet wird.
  • Größe des rollierenden Fensters: Die Größe der rollierenden Fenster in verschiedenen Phasen kann variieren, der Unterschied darf jedoch nicht mehr als 720-mal so groß sein.

Beispiel: Differenz bei der Zusammenfassung von Phasen (ungültig)

Die folgende Konfiguration ist ungültig, da ein Monat 44.640 Minuten enthält (44.640 / 1 > 720):

Phase: monthly_stats { ... match: by month }

Root: match: by minute

Beispiel: Differenz bei der Aggregation von Phasen (gültig)

Um dieses Problem zu beheben, muss das Verhältnis zwischen den Phasen kleiner sein. Beispiel: Stündliche Daten in einem Tagesbericht zusammenfassen:

Phase: daily_stats { ... match: by day }

Root: match: by hour

Da 24 (Stunden pro Tag) kleiner als 720 ist, kann das System die Stufendaten effizient der Stammstufe zuordnen.

Einschränkungen für Phasen und Abfragen

Für jede einzelne Phase einer mehrphasigen Abfrage gelten bestimmte Einschränkungen. Die meisten Einschränkungen, die für eine einphasige Abfrage gelten, gelten auch für jede einzelne Phase:

  • Ausgabeanforderung: In jeder Phase muss mindestens eine Abgleichs- oder Ergebnisvariable (Phasenfeld) ausgegeben werden.

  • Abfragezeitraum:

    • Standardabfragen: Maximal 30 Tage.
    • Mehrstufige Abfragen mit Join-Vorgängen ohne Entsprechung: Der Zeitraum ist auf maximal 14 Tage begrenzt.

  • Beschränkungen der Fenstergröße: Die maximale Fenstergröße (Hop, Sliding oder Tumbling) hängt davon ab, ob Ihre Abfrage einen Join enthält:

    • Mit Joins: Die maximale Fenstergröße für jeden Typ (Hop, Sliding oder Tumbling) beträgt 2 Tage. Weitere Informationen finden Sie unter Einschränkungen für Suchvorgänge.

    • Ohne Joins (einzelnes Ereignis):

    • Hop- und gleitende Fenster: Das Maximum beträgt 2 Tage.

    • Gleitende Zeiträume: Die maximale Dauer wird auf 30 Tage erhöht.

  • Maximale Anzahl von Ergebnisvariablen:

    • Standardmäßig 20
    • 50 für Kunden, die das höhere Limit aktiviert haben
    • Array-Limits: Eine Outcome-Variable mit Array-Werten darf maximal 10.000 Elemente enthalten.

  • Ereigniseinschränkungen pro Abfrage:

    • Maximal zwei UDM-Ereignisse
    • Maximal ein EKG-Ereignis
    • Maximal zwei Datentabellen

Dienst- und Leistungslimits

Für mehrstufige Abfragen gelten dieselben Einschränkungen wie für Statistikabfragen:

  • Statistikabfragen: 120 Abfragen pro Stunde (API und Benutzeroberfläche).
  • Suchaufrufe: 100 Aufrufe pro Minute.
  • API-Unterstützung: Das Google SecOps-System und die EventService.UDMSearch API unterstützen mehrstufige Joins, die SearchService.UDMSearch API jedoch nicht. Das System ermöglicht auch die Ausführung mehrstufiger Abfragen ohne Joins.

Einschränkungen für Events und global

Sie müssen die folgenden ereignis- und plattformweiten Grenzen einhalten:

Maximale Anzahl von Ereignissen

Bei mehrstufigen Abfragen ist die Anzahl der Ereignisse, die gleichzeitig verarbeitet werden können, stark begrenzt:

  • UDM-Ereignisse: Es sind maximal zwei UDM-Ereignisse zulässig.
  • ECG-Ereignisse (Entity Context Graph): Es ist maximal ein ECG-Ereignis zulässig.

Globale Abfragebeschränkungen

Diese plattformweiten Einschränkungen legen fest, wie weit zurückliegende Daten und wie viele Daten von einer mehrstufigen Abfrage zurückgegeben werden:

  • Zeitraum für Abfrage: Der maximale Zeitraum für eine Standardabfrage beträgt 30 Tage.
  • Gesamtergebnismenge: Die maximale Größe der Gesamtergebnismenge beträgt 10.000 Ergebnisse.

Benötigen Sie weitere Hilfe? Antworten von Community-Mitgliedern und Google SecOps-Experten erhalten