Python 룰 캐싱

개요

캐싱은 이전 디택션 실행이 이후 실행에 직접적인 영향을 미치도록 합니다. Panther의 실시간 분석 엔진은 이벤트를 하나씩 검사하며, 때로는 호출 간에 상태를 유지하는 것이 도움이 됩니다. 룰은 기본 제공 헬퍼 함수를 사용하여 값을 캐시할 수 있습니다. 이러한 헬퍼 함수는 Panther에서 호스팅되는 DynamoDB 테이블과 연동됩니다. 이 기능은 때때로 "panther-kv-store"라고도 불립니다.

디택션은 임의의 키-값 쌍을 저장하고 검색할 수 있어 디택션 실행 간 상태를 보존할 수 있습니다. 대신에 a를 사용하는 예약 검색 그리고 예약 룰을 사용하는 대신, 디택션은 이벤트 메타데이터를 실시간으로 수집하고 분석할 수 있습니다.

Panther가 관리하는 DynamoDB 테이블에서 읽어오고 싶다면 Panther 지원 팀에 문의하세요. 그러면 DynamoDB에 대한 읽기 전용 권한이 있는 Amazon Web Services (AWS) 역할이 프로비저닝됩니다.

캐시를 사용하면 디택션 처리에 상당한 지연이 추가되므로(이로 인해 알림 지연과 같은 후속 영향이 발생할 수 있음), 다음을 권장합니다:

• 소량의 데이터를 수집하는 로그 유형에 대해서만 디택션에서 캐시를 사용하세요

• 캐시가 정말로 필요한 경우에만 사용되도록 디택션을 작성하세요(참고 함정: 필요하기 전에 캐시를 사용하는 것)

일반적인 사용 사례

• 원시 이벤트, 보강, 외부 소스 등에서 데이터를 집계하는

  • 캐시를 활용하면 디택션이 데이터를 중복 제거한 다음 이후 디택션 실행 및/또는 알러트 컨텍스트에서 사용하기 위해 데이터를 집계할 수 있습니다.

• 여러 이벤트 및/또는 로그 소스의 데이터를 상호 연관시키기

  • 단일 이벤트만으로는 그다지 많은 통찰을 제공하지 못할 수 있습니다. 하지만 일련의 이벤트는 훨씬 더 유용할 수 있는 더 완전한 그림을 형성할 수 있습니다.

  • DynamoDB 캐시는 Panther 전반의 어떤 디택션 실행에서도 참조할 수 있으므로, 캐시를 사용해 디택션의 범위를 상당히 넓힐 수 있습니다.

• 위험 기반 알러트, 사용자 엔티티 및 행동 분석(UEBA)

  • DynamoDB 캐시는 Panther로 들어온 이벤트를 기반으로 엔티티를 모니터링하고 점수를 매기는 데 사용할 수 있습니다. 이는 서로 다른 이벤트를 추상화하는 계층을 제공하여 디택션이 위험한 행동을 추적, 점수화 및 분류할 수 있게 합니다.

  • 디택션은 명시적인 필드 기반 로직을 사용하지 않고도 그렇지 않으면 무작위인 이벤트 조합에 점수를 반영할 수 있습니다.

캐싱하는 동안 할 수 있습니다 이벤트를 집계하고 어떤 이벤트 임계값에 도달한 후 알러트를 생성하는 데 사용될 수 있으며, 이를 달성하려면 내장된 중복 제거 대신 기능을 사용하세요.

DynamoDB의 키-값 쌍

Panther의 디택션 캐시를 구동하는 DynamoDB는 빠르고 가벼운 NoSQL 키-값 데이터베이스입니다. Panther는 디택션 캐싱을 구동하는 단일 DynamoDB 테이블을 구현했습니다.

DynamoDB의 모든 행은 키-값 쌍:

• 키: 행의 고유 식별자(테이블 내에서 중복될 수 없음)

• Value: 주어진 키와 연결된 모든 데이터

키와 값 모두 디택션 코드에서 생성될 수 있습니다.

DynamoDB에 저장된 값은 최대 400 KB까지 클 수 있습니다.

키 생성

모든 Panther 디택션은 캐시로 동일한 DynamoDB 테이블을 공유합니다. 이는 디택션 간 캐싱에 이점을 제공하지만, 다음과 같은 키를 선택해야 합니다:

• 디택션 런타임에 프로그래밍 방식으로 생성할 수 있습니다

  • 키를 생성하는 데 사용되는 코드는 종종 함수로 분리됩니다.

  • 키 생성기 함수를 저장하는 것을 권장합니다 전역 헬퍼 여러 디택션에서 동일한 키를 구현하기 위해.

• 이벤트 값을 활용하세요

  • 예: IP 주소, 사용자 이름, 해시, ID, ARN.

• 의도된 범위 내에서 충분한 엔트로피와 고유성을 제공하세요

  • 캐시는 단일 디택션 내에서 또는 여러 디택션과 Log Sources에 동시에 구현할 수 있습니다.

  • 여러 디택션과 Log Sources에 걸쳐 동일한 캐시를 사용하려는 경우, 다음을 활용해야 할 수 있습니다 데이터 모델 공통 필드 값 분류 체계를 만들기 위해.

• 서로 충돌하지 않도록 하세요

  • 키-값 쌍을 실수로 덮어쓸 수 있으므로, 이를 방지하기 위해 키를 신중하게 구성해야 합니다.

같은 키를 사용하면 서로 다른 디택션에서 캐시된 값에 접근할 수 있습니다.

의 캐시 헬퍼 함수 panther_디택션_helpers

Panther는 유지합니다 panther_디택션_helpers, 탐지에 사용할 수 있는 pip 패키지입니다.

참조하기 panther_디택션_helpers 디택션 파일에 다음 import 문을 추가하세요:

import panther_디택션_helpers

다음과 같은 문으로 특정 함수만 가져올 수도 있습니다:

from panther_디택션_helpers.caching import get_dictionary

사전

Panther에서 제공하는 이러한 헬퍼 함수는 탐지가 사전을 캐시할 수 있도록 합니다:

• 사전 가져오기: 사전의 현재 값을 가져옵니다

• put_dictionary: 사전을 덮어씁니다

사전은 Python 라이브러리를 사용하여 직렬화 및 역직렬화됩니다. 따라서 캐시되는 사전에는 다음을 포함할 수 없습니다: json 라이브러리. 따라서 캐시되는 사전에는 다음을 포함할 수 없습니다:

• 집합

• 복소수 또는 수식

• 사용자 정의 객체

• 문자열이 아닌 키

예시

이벤트는 항상 사전으로 디택션에 전달되므로, 기본적으로 캐시할 수 있습니다:

from panther_디택션_helpers.caching import get_dictionary, put_dictionary


def 룰(event):
    key = __name__ + ":" + event.get("username")

    # 이전 이벤트 가져오기
    previous_event_data = get_dictionary(key)

    # 현재 이벤트 저장
    put_dictionary(key, event)

    # 이전 이벤트 데이터가 없으면 종료
    if not previous_event_data:
        return False

    # 이전 이벤트와 현재 이벤트 간의 IP 비교
    if event.get("ipAddress") != previous_event_data.get("ipAddress"):
        return True

    return False

코드에서 딕셔너리를 구성하고 이를 캐시하는 것도 가능합니다:

from panther_base_helpers import deep_get
from panther_디택션_helpers.caching import get_dictionary, put_dictionary


def store_login_info(key, event):
    # 사용자를 가장 최근 로그인의 경도/위도 및 시간에 매핑
    put_dictionary(
        key,
        {
            "city": deep_get(event, "client", "geographicalContext", "city"),
            "lon": deep_get(event, "client", "geographicalContext", "geolocation", "lon"),
            "lat": deep_get(event, "client", "geographicalContext", "geolocation", "lat"),
            "time": event.get("p_event_time")
        }
    )

이 방법론은 매우 복잡한 데이터 세트를 DynamoDB에 저장하도록 확장할 수 있습니다.

문자열 집합

Panther에서 제공하는 이 도우미 함수들을 사용하면 탐지가 문자열 집합을 캐시할 수 있습니다:

• get_string_set: 문자열 집합의 현재 값을 가져옵니다

• put_string_set: 문자열 집합을 덮어씁니다

• add_to_string_set: 집합에 하나 이상의 문자열을 추가합니다

• remove_from_string_set: 하나 이상의 문자열을 집합에서 제거

• reset_string_set: 집합을 비움

• set_key_expiration: 문자열 집합의 수명을 설정

예시

아래 룰은 문자열 집합을 캐싱하는 방법을 보여줍니다.

from panther_디택션_helpers.caching import add_to_string_set, get_string_set


def 룰(event):
    if event['eventName'] != 'AssumeRole':
        return False

    role_arn = event['requestParameters'].get('roleArn')
    if not role_arn:
        return False

    role_arn_key = '{}-UniqueSourceIPs'.format(role_arn)
    ip_addr = event['sourceIPAddress']

    previously_seen_ips = get_string_set(role_arn_key)

    # 이것이 유일한 값이라면, 처음 사용 시 신뢰
    if len(previously_seen_ips) == 0:
        add_to_string_set(role_arn_key, ip_addr)
        return False

    if ip_addr not in previously_seen_ips:
        return True

    return False

카운터

카운터 기반 룰을 구현하려면 다음 함수 중 하나 이상을 사용하십시오:

• get_counter: 최신 카운터 값을 가져옴

• increment_counter: 카운터에 추가(기본값 1)

• reset_counter: 카운터를 0으로 초기화

• set_key_expiration: 카운터의 수명을 설정

예시

아래 룰은 카운터 사용 방법을 보여줍니다.

from panther_디택션_helpers.caching import increment_counter, set_key_expiration, reset_counter


def 룰(event):
    # AccessDenied 호출만 분석하도록 필터링
    if event.get('errorCode') != 'AccessDenied':
        return False

    # 비교적 고유해야 하는 카운터 키를 생성
    key = '{}-AccessDeniedCounter'.format(event['userIdentity'].get('arn'))

    # 카운터를 증가시킨 다음 현재 값을 확인
    hourly_error_count = increment_counter(key)
    if hourly_error_count == 1:
        set_key_expiration(key, time.time() + 3600)
    elif failure_hourly_count >= 10:
    # 임계값을 초과하면 초기화한 다음 알러트를 반환
        reset_counter(key)
        return True
    return False

타임스탬프를 사용하여 상태 추적

DynamoDB 캐시의 일반적인 사용 사례는 특정 기간의 이벤트 그룹을 추적하는 것입니다. 모든 키-값 쌍은 코드에서 생성되어야 하므로, 값에 포함되어 제공되지 않는 한 타임스탬프 추적은 제공되지 않습니다.

디택션 작성자는 저장하는 것을 고려해야 합니다 p_event_time 이벤트를 집계할 때.

타임스탬프는 예측할 수 없는 일련의 이벤트 로그 전체에서 재현되기 매우 드물기 때문에 키에 사용해서는 안 됩니다.

Time to Live

Time to Live(TTL)는 캐시 항목에 만료 타임스탬프를 설정할 수 있게 해줍니다. 이 자동 삭제는 중복 제거 전략으로도 유용하며, 효율적인 데이터 정리에도 도움이 됩니다. 모든 캐시 항목의 기본 TTL은 90일이지만, 직접 TTL 값을 구성할 수 있습니다.

TTL은 연결된 값의 데이터 유형과 관계없이 단일 캐시 키에 연결됩니다. 예를 들어, 만약 add_to_string_set() 가 호출되면 전체 문자열 세트의 TTL이 전달된 값으로 재설정됩니다 epoch_seconds (또는 값이 전달되지 않으면 기본값인 90일)

TTL 설정

다음 중 하나를 사용하여 90일 기본 TTL을 재정의할 수 있습니다:

• 다음 epoch_seconds 캐시에 쓰는 캐싱 헬퍼 함수에서 사용할 수 있는 매개변수로, 예를 들면 put_string_set() 및 increment_counter()

• 다음 set_key_expiration() 함수

둘 다 epoch_seconds 및 set_key_expiration() 항목이 삭제되어야 하는 타임스탬프를 정의합니다. 이러한 함수는 다음에서 사용할 수 있습니다 panther_디택션_helpers.

에 대한 값을 전달하지 않는 경우 epoch_seconds, 반드시 호출하세요 set_key_expiration() 다음을 받는 모든 함수 뒤에 epoch_seconds.다음을 받는 함수가 epoch_seconds 다음 이후에 호출되고 set_key_expiration() 그리고 다음에 대한 값이 제공되지 않으면 epoch_seconds, TTL은 기본값인 90일로 재설정됩니다.

만료 타임스탬프를 생성하려면, 다음을 통해 이벤트 시간과 관련된 Unix 타임스탬프를 가져와 event.event_time_epoch(), 그리고 주어진 초를 더합니다. 결과 타임스탬프가 지나면 행은 48시간 이내에 자동으로 삭제됩니다.

이벤트 시간(p_event_time) 처리 시간(p_parse_time 또는 datetime.datetime.now())을 TTL의 기준으로 사용하는 것이 좋습니다. 이는 이벤트 처리 지연을 반영하고, 단위 테스트에서 발견되는 것과 같은 오래된 이벤트가 캐시를 어지럽히지 않도록 하기 위함입니다.

예시

Panther의 예시 지리적으로 불가능한 Okta 로그인 디택션을 사용하는 epoch_seconds:

# 테이블이 이전 사용자로 가득 차지 않도록 일주일 후 항목 만료
put_string_set(
    key,
    [
        dumps(
            {
                "city": deep_get(event, "client", "geographicalContext", "city"),
                "lon": deep_get(event, "client", "geographicalContext", "geolocation", "lon"),
                "lat": deep_get(event, "client", "geographicalContext", "geolocation", "lat"),
                "time": event.get("p_event_time"),
            }
        )
    ],
    epoch_seconds=event.event_time_epoch() + timedelta(days=7).total_seconds(),
)

다음은 동일한 예시로, 다음을 사용한 것입니다 set_key_expiration():

# 테이블이 이전 사용자로 가득 차지 않도록 일주일 후 항목 만료
put_string_set(
    key,
    [
        dumps(
            {
                "city": deep_get(event, "client", "geographicalContext", "city"),
                "lon": deep_get(event, "client", "geographicalContext", "geolocation", "lon"),
                "lat": deep_get(event, "client", "geographicalContext", "geolocation", "lat"),
                "time": event.get("p_event_time"),
            }
        )
    ]
)
set_key_expiration(key, event.event_time_epoch() + timedelta(days=7).total_seconds())

테스팅

캐시가 DynamoDB에 의존한다는 것은 디택션 코드를 테스트하고 검증하기 위해 특별한 고려가 필요하다는 뜻입니다:

Panther 콘솔에서 테스트

• 적용 가능한 함수가 모의 객체로 재정의되지 않는 한, 단위 테스트 호출은 DynamoDB와 통신합니다.

• DynamoDB로 전송되거나 수신된 데이터는 다음에 커밋할 수 있습니다 알러트_context() 단위 테스트 결과에서 디버깅하기 위해.

• DynamoDB의 원시 내용을 탐색하는 것은 불가능합니다.

CLI 워크플로로 테스트

• Panther의 디택션은 DynamoDB와 통신하기 위해 AWS IAM 역할을 활용합니다.

  • 다음을 사용할 때 panther_analysis_tool 로컬에서 또는 CI/CD 워크플로의 일부로 단위 테스트를 실행할 때 이 IAM 역할에는 접근할 수 없습니다.

  • Panther 콘솔의 맥락 밖에서는 DynamoDB 캐시와 상호작용할 수 없으므로, 테스트는 입력과 출력을 시뮬레이션해야 합니다.

• CI/CD 워크플로를 지원하기 위해, DynamoDB와 상호작용하는 모든 함수를 모의 처리하여 예상 출력을 시뮬레이션할 것을 권장합니다.

  • Panther의 문서를 참조하세요 Mocks에 대한 추가 정보.

캐시 사용 시 흔한 함정

함정: 필요하기 전에 캐시를 사용하는 것

디택션을 작성할 때는 kv-store를 필요할 때에만 호출하고, 그 전에는 호출하지 않는 것이 중요합니다. 예를 들어, 악성 행위자를 두 번 보는지 확인하는 다음 디택션을 살펴보겠습니다:

잘못된 예

panther_디택션_helpers.caching에서 reset_string_set, get_string_set, add_to_string_set를 가져오기

def 룰(event):
    bad_guys = get_string_set('BadGuys') # <-- #1
    bad_guy = event.get('BadGuyName')
    
    if event.get('eventType') == 'BadGuyDetected':
        add_to_string_set('BadGuys', bad_guy) # <-- #2
        
        if bad_guy in bad_guys:
            # bad guy 반복, 알러트
            reset_string_set('BadGuys')
            return True

    return False

이 디택션을 크게 개선할 수 있는 두 곳이 있습니다:

1. 가져옵니다 BadGuys 문자열 집합을 확인하기 전에 이것이 하나의 BadGuyDetected 이벤트입니다. 만약 그렇다면 이 아니라 bad guy 이벤트라면 문자열 집합을 가져올 필요가 없습니다. 이 호출은 항상 필요한 것은 아니지만 모든 디택션 실행에 지연을 추가합니다.

2. 새로운 것을 추가합니다 bad_guy 이것이 반복되는 bad_guy인지 확인하기 전에 문자열 집합에 추가합니다. 반복되는 bad_guy라면 알러트하고 집합을 초기화하므로 문자열 집합에 추가할 필요가 없습니다.

이러한 변경을 적용한 후 디택션은 다음과 같습니다:

panther_디택션_helpers.caching에서 reset_string_set, get_string_set, add_to_string_set를 가져오기

def 룰(event):   
    if event.get('eventType') == 'BadGuyDetected':
        bad_guy = event.get('BadGuyName')
        bad_guys = get_string_set('BadGuys') # <-- #1

        if bad_guy in bad_guys:
            # bad guy 반복, 알러트
            reset_string_set('BadGuys')
            return True
            
        add_to_string_set('BadGuys', bad_guy) # <-- #2
        
    return False