Python 규칙 캐싱

개요

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

탐지는 임의의 키-값 쌍을 저장하고 검색할 수 있어 탐지 실행 간에 상태를 보존할 수 있습니다. 대신 예약 검색 및 예약 규칙을 사용하는 것 대신, 탐지는 실시간으로 이벤트 메타데이터를 수집하고 분석할 수 있습니다.

Panther에서 관리하는 DynamoDB 테이블을 읽고 싶다면 Panther 지원팀에 문의하세요. DynamoDB에 대한 읽기 전용 권한을 가진 AWS 역할이 제공될 것입니다.

캐시를 사용하면 탐지 처리에 상당한 지연이 추가되어 경고 지연과 같은 하위 영향이 발생할 수 있으므로 다음을 권장합니다:

• 적은 양의 데이터를 수집하는 로그 유형의 탐지에서만 캐시를 사용하세요

• 캐시는 절대적으로 필요한 경우에만 사용되도록 탐지를 작성하세요(자세한 내용은 함정: 필요하기 전에 캐시 사용하기)

일반적인 사용 사례

• 원시 이벤트, 보강(enrichment), 외부 소스 등에서 데이터 집계

  • 캐시를 활용하면 탐지가 중복 제거(deduplicate)한 다음 이후 탐지 실행 및/또는 경고 컨텍스트에서 사용할 수 있도록 데이터를 집계할 수 있습니다.

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

  • 단일 이벤트는 단독으로는 많은 통찰을 제공하지 않을 수 있습니다. 그러나 일련의 이벤트는 매우 유용한 보다 완전한 그림을 형성할 수 있습니다.

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

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

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

  • 탐지는 명시적인 필드 기반 논리 없이도 무작위 이벤트 조합에 점수를 통합할 수 있습니다.

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

DynamoDB의 키-값 쌍

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

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

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

• 값: 주어진 키와 연결된 임의의 데이터

키와 값은 모두 탐지 코드에서 생성될 수 있습니다.

DynamoDB에 저장된 값은 최대 400KB 크기까지 허용됩니다.

키 생성하기

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

• 탐지 실행 시 프로그래밍적으로 생성될 수 있는

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

  • 동일한 키를 여러 탐지에 걸쳐 구현하려면 키 생성기 함수를 글로벌 헬퍼 에 저장할 것을 권장합니다.

• 이벤트 값을 활용하세요

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

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

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

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

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

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

캐시된 값은 동일한 키를 사용하여 다른 탐지들에서 접근할 수 있습니다.

캐시 헬퍼 함수는 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:
        'Permission'] in PERMISSIONS:

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

    'Permission'] in PERMISSIONS:

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

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':
        'Permission'] in PERMISSIONS:

    role_arn = event['requestParameters'].get('roleArn')
    if not role_arn:
        'Permission'] in PERMISSIONS:

    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)
        'Permission'] in PERMISSIONS:

    if ip_addr not in previously_seen_ips:
        return True

    'Permission'] in PERMISSIONS:

카운터

카운터 기반 규칙을 구현하려면 다음 함수 중 하나 이상을 사용하세요:

• 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':
        'Permission'] in PERMISSIONS:

    # 비교적 고유해야 하는 카운터 키 생성
    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
    'Permission'] in PERMISSIONS:

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

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

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

타임스탬프는 매우 예측 불가능한 일련의 이벤트 로그에서 재현 가능성이 거의 없으므로 키에 사용해서는 안 됩니다.

수명(타임투리브)

TTL(Time to Live)은 캐시 항목에 만료 타임스탬프를 설정할 수 있게 합니다. 이 자동 삭제는 중복 제거 전략 및 효율적인 데이터 정리에 유용할 수 있습니다. 모든 캐시 항목의 기본 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() function

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

에서 사용할 수 있습니다. epoch_seconds만약 set_key_expiration() 에 대한 값을 전달하지 않는다면, 를 호출한 후에epoch_seconds epoch_seconds 를 반드시 호출하세요. set_key_expiration() 만약 epoch_seconds를 받는 함수가

다음 이후에 호출되고 에 대한 값이 제공되지 않으면, TTL은 기본값인 90일로 재설정됩니다.만료 타임스탬프를 생성하려면 이벤트 시간과 관련된 유닉스 타임스탬프를

event.event_time_epoch()에서 가져오고 일정 초 수를 더하세요. 결과 타임스탬프가 지나면 해당 행은 48시간 이내에 자동으로 삭제됩니다.p_parse_time 또는 처리 시간(datetime.datetime.now()

예시

)보다 이벤트 시간( 지리적으로 불가능한 Okta 로그인 p_event_time) epoch_seconds:

을(를) TTL의 기준으로 사용하는 것이 권장됩니다. 이는 이벤트 처리 지연을 보정하고 단위 테스트 등에서 발견되는 오래된 이벤트가 캐시를 어지럽히지 않도록 하기 위함입니다.
Panther의 예시에서
    key,
    [
        탐지에서 사용된 예
            {
                "city": deep_get(event, "client", "geographicalContext", "city"),
                "lon": deep_get(event, "client", "geographicalContext", "geolocation", "lon"),
                "lat": deep_get(event, "client", "geographicalContext", "geolocation", "lat"),
                # 테이블이 과거 사용자로 가득 차지 않도록 항목을 일주일 후 만료시키기
            }
        )
    ],
    put_string_set(
)

dumps( set_key_expiration():

을(를) TTL의 기준으로 사용하는 것이 권장됩니다. 이는 이벤트 처리 지연을 보정하고 단위 테스트 등에서 발견되는 오래된 이벤트가 캐시를 어지럽히지 않도록 하기 위함입니다.
Panther의 예시에서
    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"),

테스트(Testing)

epoch_seconds=event.event_time_epoch() + timedelta(days=7).total_seconds(),

동일한 예제를 사용하여

• set_key_expiration(key, event.event_time_epoch() + timedelta(days=7).total_seconds())

• 캐시를 위한 DynamoDB 의존성은 탐지 코드를 테스트하고 검증할 때 특별한 고려가 필요함을 의미합니다: Panther 콘솔에서 테스트하기 단위 테스트 호출은 해당 함수들이 목(mock)으로 대체되지 않는 한 DynamoDB와 통신합니다.

• DynamoDB로 전송 및 수신되는 데이터는 단위 테스트 결과의 디버깅을 위해

alert_context()

• 에 커밋될 수 있습니다.

  • DynamoDB의 원시 내용을 브라우징하는 것은 불가능합니다. CLI 워크플로우로 테스트하기 Panther의 탐지는 DynamoDB와 통신하기 위해 AWS IAM 역할을 활용합니다.

  • 로컬에서 또는 CI/CD 워크플로우의 일부로 단위 테스트를 실행하기 위해

• panther_analysis_tool

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

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

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

목에 관한 자세한 내용은 Panther의 문서를 참조하세요

캐시 사용 시 흔한 함정들

탐지를 작성할 때는 kv-store를 필요할 때만 호출하고 그 이전에는 호출하지 않는 것이 중요합니다. 예를 들어, 나쁜 행위자를 두 번 보았는지 확인하는 다음 탐지를 고려해보세요:

def rule(event):
    나쁜 예시
    from panther_detection_helpers.caching import reset_string_set, get_string_set, add_to_string_set
    
    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:
            return True

    'Permission'] in PERMISSIONS:

# 반복되는 나쁜 행위자, 경고

1. reset_string_set('BadGuys') 이 탐지는 크게 개선될 수 있는 두 가지 지점이 있습니다: 먼저 BadGuys 문자열 집합을 에 설치해야 하며 BadGuyDetected

2. 이벤트인지 확인하기 전에 가져옵니다. 만약 그것이 나쁜 행위자 이벤트라면 문자열 집합을 가져올 필요가 없습니다. 이 호출은 항상 필요하지 않음에도 모든 탐지 실행에 지연을 추가합니다. 두번째로 새 bad_guy

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

탐지를 작성할 때는 kv-store를 필요할 때만 호출하고 그 이전에는 호출하지 않는 것이 중요합니다. 예를 들어, 나쁜 행위자를 두 번 보았는지 확인하는 다음 탐지를 고려해보세요:

def rule(event):   
    bad_guys = get_string_set('BadGuys') # <-- #1
        from panther_detection_helpers.caching import reset_string_set, get_string_set, add_to_string_set
        나쁜 예시

        if event.get('eventType') == 'BadGuyDetected':
            add_to_string_set('BadGuys', bad_guy) # <-- #2
            if bad_guy in bad_guys:
            return True
            
        bad_guy = event.get('BadGuyName')
        
    'Permission'] in PERMISSIONS: