Arbiter¶

Arbiter는 SIEM의 데이터 분석 엔진입니다.

SIEM 탐지 규칙을 생성 및 편집할 때 Arbiter는 작성된 스크립트를 실행하여 데이터를 처리하고 이벤트를 생성합니다. Arbiter는 일련의 내장 함수를 제공하며, 스크립트의 입출력 관련 함수로는 dql 함수(Guance에서 데이터 조회), trigger 함수(이벤트 트리거), printf 함수(정보를 표준 출력으로 출력) 등이 있습니다.

시작하기¶

오늘 기준 어제보다 새로 추가된 방문 IP를 통계하는 예를 들어, 스크립트는 다음과 같습니다.

DQL 문 R::resource:(distinct(ip) as ip) [2d:1d]를 사용하여 하루 전 사용자 방문의 중복 제거된 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 목록은 일반적으로 하나의 요소만 포함합니다(즉, 하나의 시계열만 포함).

원래 결과를 처리하여 IP 목록을 얻어야 합니다. 구체적으로는 다음 두 가지 방법으로 구현할 수 있습니다.

dql_series_get 함수를 통해 for 루프를 통한 순회

이 함수는 모든 시계열을 가져와 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 함수는 여러 번 트리거를 지원합니다. 한 번만 실행되었으므로 현재 결과에는 하나의 요소만 있습니다. 그 결과는 웹 페이지에서 템플릿 변수로 사용할 수 있습니다. 예: {{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( # 이벤트 한 번 트리거
            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("이 세션(%s)에서 login을 방문하지 않고 send_msg를 방문했습니다", session_id)
            )
        }

        # 오름차순이므로 첫 번째 요소가 가장 이른 방문 시간입니다
        if pos_login_api > pos_send_msg_api {
            # 이벤트 트리거
            trigger(
                result= strfmt("이 세션(%s)에서 login이 send_msg보다 나중에 방문되었습니다", session_id)
            )
        }
    }
}

특정 토큰이 여러 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("토큰 %s이(가) 여러 IP에서 방문되었습니다. IPs: %s", token, str_join(ip_list[0], ", "))
        )
    }
}