Python 룰 캐싱

개요

캐싱을 통해 이전 디택션 실행이 이후 실행에 직접 영향을 줄 수 있습니다. Panther의 실시간 분석 엔진은 이벤트를 하나씩 검사하며 때때로 호출 간 상태를 유지하는 것이 유용합니다. 룰은 내장 헬퍼 함수를 사용하여 값을 캐시할 수 있습니다. 이러한 헬퍼 함수는 Panther가 호스팅하는 DynamoDB 테이블과 인터페이스합니다. 이 기능은 때때로 "panther-kv-store"라고 불립니다.

디텍션은 임의의 키-값 쌍을 저장하고 검색할 수 있어 디텍션 실행 간 상태 보존이 가능합니다. 다음을 사용하는 대신 스케줄된 검색 및 스케줄된 룰, 디텍션은 대신 실시간으로 이벤트 메타데이터를 수집하고 분석할 수 있습니다.

Panther가 관리하는 DynamoDB 테이블에서 읽기를 원하면 Panther 지원팀에 문의하세요. DynamoDB에 대한 읽기 전용 권한이 있는 AWS 역할이 할당됩니다.

캐시 사용은 디텍션 처리에 상당한 지연을 추가하므로(이는 경보 지연과 같은 하류 영향이 있을 수 있음) 다음을 권장합니다:

• 데이터 양이 적게 수집되는 로그 유형의 디텍션에서만 캐시를 사용하세요

• 캐시는 절대적으로 필요한 경우에만 사용되도록 디텍션을 작성하세요(자세한 내용은 함수가 필요하기 전에 캐시를 사용하는 함정)

일반적인 사용 사례

• 원시 이벤트, 인리치먼트, 외부 소스 등에서 데이터 집계

  • 캐시를 활용하면 디텍션이 중복을 제거한 다음 나중 디텍션 실행 및/또는 경보 컨텍스트에서 사용할 수 있도록 데이터를 집계할 수 있습니다.

• 여러 이벤트 및/또는 로그 소스의 데이터 상관관계

  • 단일 이벤트는 고립된 상태에서는 많은 통찰을 제공하지 않을 수 있습니다. 그러나 일련의 이벤트는 보다 완전한 그림을 형성할 수 있어 매우 유용할 수 있습니다.

  • DynamoDB 캐시는 Panther 전반의 모든 디텍션 실행에서 참조될 수 있으므로 캐시를 사용하여 디텍션의 범위를 상당히 넓힐 수 있습니다.

• 리스크 기반 알림, 사용자 엔터티 및 행동 분석(UEBA)

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

  • 디텍션은 명시적인 필드 기반 로직 없이도 무작위 이벤트 조합에 점수를 통합할 수 있습니다.

캐싱 는 이벤트를 카운트하고 일정 이벤트 임계값이 충족되면 경보를 생성하는 데 사용할 수 있지만, 대신 내장된 중복 제거 기능을 사용하는 것이 권장됩니다.

DynamoDB의 키-값 쌍

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

DynamoDB 내의 모든 행은 키-값 쌍:

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

• 값: 주어진 키와 쌍을 이루는 모든 데이터

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

DynamoDB에 저장된 값은 최대 400KB까지 가능합니다.

키 생성

모든 Panther 디텍션은 캐시로 동일한 DynamoDB 테이블을 공유합니다. 이는 교차 디텍션 캐싱의 이점을 제공하지만 다음과 같은 키 선택을 요구합니다:

• 디텍션 런타임에 프로그래밍적으로 생성될 수 있어야 함

  • 키를 생성하는 데 사용되는 코드는 종종 함수에 배치됩니다.

  • 동일한 키를 여러 디텍션에서 구현하려면 키 생성기 함수를 글로벌 헬퍼 에 저장하는 것을 권장합니다.

• 이벤트 값을 활용

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

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

  • 캐시는 단일 디텍션 내에서 구현되거나 여러 디텍션 및 로그 소스에 대해 동시에 구현될 수 있습니다.

  • 여러 디텍션 및 로그 소스에서 동일한 캐시를 사용하려는 경우, 공통 필드 값 분류 체계를 만들기 위해 데이터 모델 를 활용해야 할 수 있습니다.

• 서로 충돌하지 않음

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

캐시된 값은 동일한 키를 사용하여 다른 디텍션에서 액세스할 수 있습니다.

캐시 헬퍼 함수는 panther_detection_helpers

Panther는 panther_detection_helpers라는 pip 패키지를 유지 관리하며 디텍션에서 사용할 수 있습니다.

디텍션 파일에서 panther_detection_helpers 를 참조하려면 다음 import 문을 추가하세요:

import panther_detection_helpers

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

from panther_detection_helpers.caching import get_dictionary

사전

이러한 Panther 제공 헬퍼 함수는 디텍션이 사전을 캐시할 수 있게 합니다:

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

• put_dictionary: 사전을 덮어씁니다

사전은 Python의 json 형으로 분류됩니다. 라이브러리를 사용하여 직렬화 및 역직렬화됩니다. 따라서 캐시된 사전은 다음을 포함할 수 없습니다:

• 셋(집합)

• 복소수 또는 수식

• 커스텀 객체

• 문자열이 아닌 키

예제

이벤트는 항상 디텍션에 사전으로 전달되므로 네이티브로 캐시할 수 있습니다:

from panther_detection_helpers.caching import get_dictionary, put_dictionary


def rule(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_detection_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_detection_helpers.caching import add_to_string_set, get_string_set


def rule(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_detection_helpers.caching import increment_counter, set_key_expiration, reset_counter


def rule(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)

수명( TTL )을 사용하면 캐시 항목에 만료 타임스탬프를 설정할 수 있습니다. 이 자동 삭제는 중복 제거 전략과 효율적인 데이터 정리에 유용합니다. 모든 캐시 항목의 기본 TTL은 90일이지만, 자체 TTL 값을 구성할 수 있습니다.

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

TTL 설정

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

• 사용자를 사용할 것이며, epoch_seconds put_string_set()와 같은 캐시에 쓰는 헬퍼 함수에서 사용할 수 있는 매개변수 put_string_set() 와 increment_counter()

• 사용자를 사용할 것이며, set_key_expiration() function

둘 다 epoch_seconds 와 set_key_expiration() 항목을 삭제해야 하는 타임스탬프를 정의합니다. 이러한 함수는 panther_detection_helpers.

에서 제공됩니다. epoch_seconds만약 당신이 set_key_expiration() 에 값을 전달하지 않는다면, 를 호출하는 것을 잊지 마세요모든 epoch_seconds를 받는 함수들 이후에. epoch_seconds 만약 epoch_seconds를 받는 함수가 set_key_expiration() 다음 호출 이후에 호출되고 epoch_seconds에 대한 값이 제공되지 않으면, TTL은 기본값인 90일로 재설정됩니다.

만료 타임스탬프를 생성하려면 이벤트 시간과 관련된 유닉스 타임스탬프를 가져와서 event.event_time_epoch()에 주어진 초 수를 더하세요. 결과 타임스탬프가 지나면 해당 행은 48시간 이내에 자동으로 삭제됩니다.

처리 지연을 고려하고 단위 테스트에서 발견되는 오래된 이벤트가 캐시를 어지럽히지 않도록 TTL의 기준으로 처리 시간(p_event_time) 대신 이벤트 시간(p_parse_time 이전에 생성한 Snowflake 사용자 이름, 예를 들면 datetime.datetime.now())을 사용하는 것이 권장됩니다.

예제

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로 송수신되는 데이터는 단위 테스트 결과에서 디버깅을 위해 alert_context() 에 커밋될 수 있습니다.

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

CLI 워크플로우로 테스트하기

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

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

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

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

  • 모킹에 대한 자세한 정보는 Panther 문서를 참조하세요 모킹에 대한 추가 정보.

캐시 사용 시의 일반적인 함정

함수가 필요하기 전에 캐시를 사용하는 함정

디텍션을 작성할 때는 kv-store를 필요할 때만 호출하고 그 이전에는 호출하지 않는 것이 중요합니다. 예를 들어, 동일한 악성 행위를 두 번 목격하는지를 확인하는 다음 디텍션을 고려해 보세요:

나쁜 예시

from panther_detection_helpers.caching import reset_string_set, get_string_set, add_to_string_set

def rule(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:
            # 반복되는 악성 행위자, 경고
            reset_string_set('BadGuys')
            return True

    return False

이 디텍션은 다음 두 가지 점에서 크게 개선될 수 있습니다:

1. 다음을 가져옵니다: BadGuys 문자열 집합을 BadGuyDetected 이벤트인지 확인하기 전에. 만약 이것이 Enterprise 조직 나쁜 행위자 이벤트라면, 문자열 집합을 가져올 필요가 없습니다. 이 호출은 항상 필요하지 않음에도 모든 디텍션 실행에 지연을 추가합니다.

2. 새로운 bad_guy 를 반복 악성 행위자인지 확인하기 전에 문자열 집합에 추가합니다. 만약 반복 악성 행위자라면 우리는 경고를 발생시키고 집합을 재설정할 것이므로 문자열 집합에 추가할 필요가 없습니다.

이 변경을 적용한 후 디텍션은 다음과 같이 보입니다:

from panther_detection_helpers.caching import reset_string_set, get_string_set, add_to_string_set

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

        if bad_guy in bad_guys:
            # 반복되는 악성 행위자, 경고
            reset_string_set('BadGuys')
            return True
            
        add_to_string_set('BadGuys', bad_guy) # <-- #2
        
    return False