コンテンツにスキップ

Arbiter

Arbiter は SIEM のデータ分析エンジンです。

SIEM 検出ルールの作成と編集において、Arbiter は記述されたスクリプトを実行し、データを処理してイベントを生成します。Arbiter は一連の組み込み関数を提供します。スクリプトの入出力に関連する関数としては、dql 関数（Guance からデータをクエリする）、trigger 関数（イベントをトリガーする）、printf 関数（情報を標準出力に出力する）などがあります。

クイックスタート

今日の新規アクセス IP を昨日と比較する例を以下に示します。スクリプトは次のとおりです。

DQL ステートメント R::`resource`:(distinct(`ip`) as ip) [2d:1d] を使用して、1 日前のユーザーアクセスにおける重複排除された IP データをクエリします。

v = dql("R::`resource`:(distinct(`ip`) as ip) [2d:1d]")
# 結果を stdout に出力
printf("%v", v)

スクリプトを実行した結果は次のとおりです。

{"series":[[{"columns":{"ip":"120.20.000.79","time":1747041737200},"tags":null},{"columns":{"ip":"120.130.000.85","time":1747031791143},"tags":null},{"columns":{"ip":"153.30.000.2","time":1747030318384},"tags":null}]],"status_code":200}

by ステートメントを使用してグループ化していない場合、DQL クエリ結果の series リストには通常 1 つの要素のみが含まれます（つまり、1 つの時系列のみを含みます）。

IP リストを取得するには、元の結果を処理する必要があります。具体的には、次の 2 つの方法で実現できます。

この関数はすべての時系列を取得し、2 次元リストを返します。フィールドが存在しない場合は、nil がプレースホルダーとして使用されます。

  • スクリプト：
result_dql = dql("R::`resource`:(distinct(`ip`) as ip) [2d:1d]")
ips = dql_series_get(result_dql, "ip")

printf("%v", ips)
  • 標準出力：
["120.20.000.79","120.130.000.85","153.30.000.2"]
  • スクリプト：
result_dql = dql("R::`resource`:(distinct(`ip`) as ip) [2d:1d]")
ips = []
for series in result_dql["series"] {
    for elem in series {
        # columns には ip フィールドがあり、tags にはないことがわかっている
        if "columns" in elem && "ip" in elem["columns"] {
            ips = append(ips, elem["columns"]["ip"])
        } else {
            # dql_series_get 関数では、該当フィールドがない場合、nil がプレースホルダーとして追加される
            ips = append(ips, nil)
        }
    }
}
printf("%v", ips)
  • 標準出力：
["120.250.000.179","120.130.000.185","153.30.000.42"]

IP リストを取得したら、今日のデータと昨日の IP リストデータの比較を開始できます。スクリプトの参考例は次のとおりです。

# 昨日の
result_dql = dql("R::`resource`:(distinct(`ip`) as ip) [2d:1d]")

ip_yesterday = dql_series_get(result_dql, "ip")

# 今日の
result_dql = dql("R::`resource`:(distinct(`ip`) as ip) [1d]")
ip_today = dql_series_get(result_dql, "ip")


# len(ip_today) > 0 のチェックを推奨。インデックスが範囲外の場合、実行時にエラーが発生する
new_ips = []
for s in ip_today[0] {
    if s == nil {
        continue
    }
    if !(s in ip_yesterday[0]) {
        new_ips = append(new_ips, s)
    }
}

# 新規 IP 検出イベントをトリガー
trigger(
    result=new_ips,
    status="info",
    dimension_tags={
        "user_cron_job_": "new_ips_check",
        "data_category": "rum"
    },
    related_data={
        "IPs": new_ips
    }
)

trigger 関数は複数回のトリガーをサポートしています。今回は 1 回だけ実行されたため、現在の結果には 1 つの要素のみが含まれています。その結果は、Web ページ上でテンプレート変数として使用できます。たとえば、{{Result}} は以下の result に対応します。

[
  {
    "result": [
        "116.60.000.211",
        "14.150.000.99",
        "58.240.000.245"
    ],
    "status": "info",
    "dimension_tags": {
      "data_category": "rum",
      "user_cron_job_": "new_ips_check"
    },
    "related_data": {
      "IPs": [
        "116.60.000.211",
        "14.150.000.99",
        "58.240.000.245"
      ]
    }
  }
]

ユースケース例

異常ログによる地理的位置の変更の特定

# クエリ時間範囲は自動的に設定されます。定期トリガーの場合
user_id_data = dql("L::access_log:(distict(user_id) as user_id)")

user_ids = dql_series_get(user_id_data, "user_id")
# user_ids の値は [["user_id_1", "user_id_2", "user_id_3"]] のようになる

user_ids = user_ids[0]

for uid in user_ids {
    # ユーザーアクセス国の情報をクエリし、重複を排除
    country_data = dql(strfmt(
        "L::access_log:(distict(country) as country) {user_id = \"%s\"}", uid)) 

    country_list = dql_series_get(country_data, "country")

    # 複数の国がある場合
    if len(country_list) == 1 && len(country_list[0]) > 1 {
        trigger( # イベントを 1 回トリガー
            result=uid,
            related_data = country_list[0]
        )
    }
}

URL アクセス順序の検出

# 検出期間内のすべてのセッション ID を取得
session_id_data = dql("R::view:(distinct(session_id) as session_id)")

session_id_list = dql_series_get(session_id_data, "session_id")
session_id_list = session_id_list[0]

for session_id in session_id_list {
    # 指定された session_id に対するすべての view_path を取得
    view_data = dql(strfmt("R::view:(view_path) {session_id = \"%s\"} order by time asc", session_id))

    view_path = dql_series_get(view_data, "view_path")
    view_path = view_path[0]

    pos_login_api = -1
    pos_send_msg_api = -1

    for i = 0; i < len(view_path); i++ {
        if pos_login_api < 0 && view_path[i] == "/login" {
            pos_login_api = i
        }
        if pos_send_msg_api < 0 && view_path[i] == "/send_message" {
            pos_send_msg_api = i
        }

        if pos_login_api == -1 && pos_send_msg_api >= 0 {
            trigger(
                result= strfmt("この session(%s) では、login にアクセスする前に send_msg にアクセスしました", session_id)
            )
        }

        # 昇順のため、最初の要素が最も古いアクセス時間
        if pos_login_api > pos_send_msg_api {
            # イベントをトリガー
            trigger(
                result= strfmt("この session(%s) では、login が send_msg より後にアクセスされました", session_id)
            )
        }
    }
}

特定の Token が複数の IP からアクセスされている

token_data = dql("L::access_log:(distict(token) as token)")

token_list = dql_series_get(token_data, "token")
token_list = token_list[0]

for token in token_list {
    token_ip_data = dql(strfmt(
        "L::access_log:(distinc(ip) as ip) {token = \"%s\"}", token)
    ip_list = dql_series_get(token_ip_data, "ip")

    if len(ip_list) == 0 && len(ip_list[0]) > 1  {
        # イベントをトリガー
        trigger(
            result = strfmt("token %s が複数の ip からアクセスされました, IPs: %s", token, str_join(ip_list[0], ", "))
        )
    }
}

フィードバック

このページは役に立ちましたか？