스크립트 로그 파서

Starlark 구성 언어로 정의된 스크립트로 들어오는 로그를 파싱

개요

script 의 가능한 값 중 하나입니다 파서 키 사용자 지정 로그 스키마에서. 이 파서는 Panther가 들어오는 각 로그 이벤트에 수행해야 하는 변환을 다음을 사용하여 지정할 수 있게 합니다 Starlark 구성 언어로, Python과 많은 문법적 유사점을 공유합니다. Panther의 script 파서는 구조화된(JSON) 이벤트와 비구조화된 이벤트를 모두 처리할 수 있습니다.

다음을 사용하면 도움이 될 수 있습니다 script 파서를 사용해 다음을 하고 싶을 때:

• 비구조화된 로그를 파싱하지만, 다른 파서 옵션(csv, fastmatch, regex)으로는 부족할 때

• 데이터에 변환을 수행하지만, Panther가 제공하는 스키마 변환이 부족할 때

다음을 이해하기 script 파서

다음을 정의하기 함수

다음을 사용할 때 script 파서를 사용하려면 Starlark 함수함수를 구현해야 합니다. 이 함수는 string 를 입력받아 비어 있지 않은 딕셔너리를 반환해야 합니다. 반환된 딕셔너리는 출력 이벤트의 형식을 정의합니다.

사용 가능한 함수

다음 script 파서는 Starlark 사양에 설명된 기본 요소를 사용할 수 있습니다. 또한 다음 함수를 사용할 수 있습니다:

함수 이름	설명
json.decode	JSON 문자열을 딕셔너리로 디코딩합니다
json.encode	딕셔너리를 JSON 문자열로 인코딩합니다
base64.decode	base64로 인코딩된 문자열을 디코딩합니다
base64.encode	문자열에 base64 인코딩을 수행합니다

제한 사항

스크립트에는 다음 제한 사항이 적용됩니다:

• 예외를 발생시키는 것은 허용되지 않습니다.

• import는 허용되지 않습니다.

JSON 처리

반면 script 은 주로 텍스트 로그에 사용하도록 되어 있지만, Panther에서 기본적으로 지원하는 것 이외의 변환을 수행하고 싶을 때는 JSON 로그에도 사용할 수 있습니다. 이러한 이유로 script 파서에는 json 모듈이 미리 로드되어 있어, JSON을 문자열 유형에서 딕셔너리로 변환할 수 있습니다.

예를 들어, 다음 구성은 is_panther_employee 라는 새 필드를 생성하며, 이는 true actor 이메일에 panther.com 도메인이 있으면 false 그렇지 않으면 false가 됩니다.

파서:
  스크립트:
    함수: |
      def parse(log):
        event = json.decode(log)
        if event['actor']['email'].endswith('@panther.com'):
          event['is_panther_employee'] = True
        else:
          event['is_panther_employee'] = False
        return event

이해를 돕기 위해 위의 parse 함수는 아래에 Python 구문 강조와 함께 표시됩니다:

def parse(log):
  event = json.decode(log)
  if event['actor']['email'].endswith('@panther.com'):
    event['is_panther_employee'] = True
  else:
    event['is_panther_employee'] = False
  return event

다음을 사용하는 예 script

다음 Apache Common Log 형식의 로그 라인이 Panther로 전송된다고 가정해 보겠습니다:

127.0.0.1 - frank [10/Oct/2000:13:55:36 -0700] "GET /apache_pb.gif HTTP/1.0" 200 2326

이 로그 유형을 script를 사용해 파싱하기 위해, 다음 함수를 정의합니다:

def parse(log):
  fields = log.split(" ")
  return {
    'remote_ip': fields[0],
    'identity': fields[1],
    'user': fields[2],
    'timestamp': ' '.join(fields[3:5]).strip('[]'),
    'request_uri': ' '.join(fields[5:8]).strip('"'),
    "status": int(fields[8]),
    "bytes_sent": int(fields[9])
  }

그리고 다음 스키마 필드를 사용합니다:

fields:
  - 이름: remote_ip
    type: string
    지표:
      - ip
  - 이름: identity
    type: string
  - 이름: user
    type: string
  - name: timestamp
    type: timestamp
    isEventTime: true
    timeFormats:
     - '%d/%b/%Y:%H:%M:%S %z'
  - 이름: method
    type: string
  - name: request_uri
    type: string
  - name: protocol
    type: string
  - name: status
    type: int
  - name: bytes_sent
    type: bigint

위 로그가 이 파서로 정규화되면 다음과 같이 됩니다:

{
    "bytes_sent":2326,
    "identity": "-",
    "method":"GET",
    "protocol":"HTTP/1.0",
    "remote_ip":"127.0.0.1",
    "request_uri":"/apache_pb.gif",
    "status":200,
    "timestamp":"2000-10-10 20:55:36.000000000",
    "user":"frank"
}