# Python 룰 캐싱

## 개요

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

디택션은 임의의 키-값 쌍을 저장하고 검색할 수 있어, 디택션 실행 간 상태를 보존할 수 있습니다. 다음을 사용하는 대신 [예약 검색](/ko/search/scheduled-searches.md) 및 예약 룰, 디택션은 대신 실시간으로 이벤트 메타데이터를 수집하고 분석할 수 있습니다.

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

{% hint style="warning" %}
캐시를 사용하면 디택션 처리에 상당한 지연이 추가되며(그 결과 알러트 지연과 같은 하위 영향이 발생할 수 있음), 따라서 다음을 권장합니다:

* 데이터를 적게 수집하는 로그 유형의 디택션에서만 캐시를 사용하세요
* 캐시는 정말 필요한 경우에만 사용하도록 디택션을 작성하세요(참조 [함정: 필요하기 전에 캐시 사용하기](#pitfall-using-the-cache-before-it-is-necessary))
  {% endhint %}

## 일반적인 사용 사례

* **원시 이벤트, 보강, 외부 소스 등에서 데이터 집계**
  * 캐시를 활용하면 디택션이 데이터를 중복 제거한 다음 집계하여 이후 디택션 실행 및/또는 알러트 컨텍스트에 사용할 수 있습니다.
* **여러 이벤트 및/또는 로그 소스의 데이터 상호 연관**
  * 단일 이벤트만으로는 그 자체로 큰 통찰을 제공하지 못할 수 있습니다. 그러나 여러 이벤트가 모이면 훨씬 더 완전한 그림을 형성할 수 있으며, 이는 매우 유용할 수 있습니다.
  * DynamoDB 캐시는 Panther 전반의 어떤 디택션 실행에서도 참조할 수 있으므로, 디택션의 범위를 상당히 넓히는 데 사용할 수 있습니다.
* **위험 기반 알러트, 사용자 엔터티 및 행동 분석(UEBA)**
  * DynamoDB 캐시는 Panther로 들어온 이벤트를 기반으로 엔터티를 모니터링하고 점수를 매기는 데 사용할 수 있습니다. 이는 서로 다른 이벤트를 추상화하는 계층을 제공하여, 디택션이 위험한 행동을 추적, 점수화 및 분류할 수 있게 합니다.
  * 디택션은 명시적인 필드 기반 로직 없이도 겉보기에 무작위인 이벤트 조합에 점수를 반영할 수 있습니다.

{% hint style="warning" %}
캐싱은 *이* 이벤트를 집계하고 특정 이벤트 임계값에 도달한 후 알러트를 생성하는 데 사용할 수 있지만, 이를 위해서는 대신 내장 [중복 제거](/ko/detections/rules.md#deduplication-of-alerts) 기능을 사용하는 것이 좋습니다.
{% endhint %}

## DynamoDB의 키-값 쌍

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

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

* **키**: 행의 고유 식별자(테이블 내에서 중복될 수 없음)
* **값**: 주어진 키와 연결된 임의의 데이터

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

{% hint style="info" %}
DynamoDB에 저장된 값은 최대 400KB까지 가능합니다.
{% endhint %}

### 키 생성하기

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

* **디택션 런타임에 프로그래밍 방식으로 생성할 수 있어야 함**
  * 키를 생성하는 데 사용되는 코드는 종종 함수 안에 배치됩니다.
  * 키 생성기 함수를 다음에 저장하는 것을 권장합니다. [Global Helper](/ko/detections/rules/python/globals.md) 여러 디택션에서 동일한 키를 구현하기 위해.
* **이벤트 값 활용**
  * 예: IP 주소, 사용자 이름, 해시, ID, ARN.
* **의도된 범위 내에서 충분한 엔트로피와 고유성을 제공해야 함**
  * 캐시는 단일 디택션 내에서 구현될 수도 있고, 여러 디택션 및 로그 소스에 동시에 적용될 수도 있습니다.
  * 여러 디택션 및 로그 소스에서 동일한 캐시를 사용하려는 경우, 다음을 활용해야 할 수 있습니다. [Data Models](/ko/detections/rules/python/data-models.md) 공통 필드 값 분류 체계를 만들기 위해.
* **서로 충돌해서는 안 됨**
  * 키-값 쌍이 실수로 덮어써질 수 있으므로, 이를 방지하도록 키를 신중하게 구성해야 합니다.

{% hint style="info" %}
캐시된 값은 동일한 키를 사용하여 서로 다른 디택션에서 액세스할 수 있습니다.
{% endhint %}

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

Panther는 [`panther_디택션_helpers`](https://pypi.org/project/panther-detection-helpers/), 디택션에서 사용할 수 있는 pip 패키지를 유지합니다.

참조하려면 `panther_디택션_helpers` 를 디택션 파일에서 참조하려면 다음 import 문을 추가하세요:

```python
import panther_디택션_helpers
```

다음과 같은 문으로 특정 함수만 import할 수도 있습니다:

```python
from panther_디택션_helpers.caching import get_dictionary
```

### 딕셔너리

Panther에서 제공하는 이러한 헬퍼 함수는 디택션이 딕셔너리를 캐시할 수 있도록 합니다:

* `get_dictionary`: 딕셔너리의 현재 값을 가져옵니다
* `put_dictionary`: 딕셔너리를 덮어씁니다

딕셔너리는 Python `json` 라이브러리를 사용하여 직렬화 및 역직렬화됩니다. 따라서 캐시되는 딕셔너리에는 다음이 포함될 수 없습니다:

* 세트
* 복소수 또는 수식
* 사용자 지정 객체
* 문자열이 아닌 키

#### 예제

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

```python
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"):
        True를 반환합니다

    return False
```

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

```python
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")
        }
    )
```

{% hint style="info" %}
이 방법론은 매우 복잡한 데이터 세트를 DynamoDB에 저장하도록 확장할 수 있습니다.
{% endhint %}

### 문자열 세트

Panther에서 제공하는 이러한 헬퍼 함수는 탐지가 문자열 집합을 캐시할 수 있게 해줍니다:

* [`get_string_set`](/ko/detections/rules/python/globals.md#get_string_set): 문자열 집합의 현재 값을 가져옵니다
* [`put_string_set`](/ko/detections/rules/python/globals.md#put_string_set): 문자열 집합을 덮어씁니다
* `add_to_string_set`: 집합에 하나 이상의 문자열을 추가합니다
* `remove_from_string_set`: 집합에서 하나 이상의 문자열을 제거합니다
* `reset_string_set`: 집합을 비우기
* [`set_key_expiration`](#time-to-live-ttl): 문자열 집합의 수명을 설정하기

#### 예시

아래 룰은 문자열 집합 캐싱의 데모를 제공합니다.

```python
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:
        True를 반환합니다

    return False
```

### 카운터

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

* `get_counter`: 최신 카운터 값을 가져옵니다
* `increment_counter`: 카운터에 추가합니다(기본값은 1)
* `reset_counter`: 카운터를 0으로 재설정합니다
* `set_key_expiration`: 카운터의 수명을 설정합니다

#### 예시

아래 룰은 카운터 사용 예시를 제공합니다.

```python
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)
        True를 반환합니다
    return False
```

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

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

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

{% hint style="info" %}
타임스탬프는 키에 사용해서는 안 됩니다. 예측할 수 없는 일련의 이벤트 로그 전반에서 재현 가능한 경우가 매우 드물기 때문입니다.
{% endhint %}

### TTL(Time to Live)

TTL(Time to Live)을 사용하면 캐시의 항목에 만료 타임스탬프를 설정할 수 있습니다. 이러한 자동 삭제는 중복 제거 전략으로 유용할 수 있을 뿐만 아니라 효율적인 데이터 정리에도 도움이 됩니다. 모든 캐시 항목의 기본 TTL은 90일이지만, 가능합니다 [자체 TTL 값을 구성하기](#setting-the-ttl).

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

#### TTL 설정

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

* 해당 `epoch_seconds` 캐시에 쓰는 캐싱 헬퍼 함수에서 사용할 수 있는 매개변수(예: `put_string_set()` 와 `increment_counter()`
* 해당 `set_key_expiration()` 함수

둘 다 `epoch_seconds` 와 `set_key_expiration()` 항목이 삭제되어야 하는 타임스탬프를 정의합니다. 이러한 함수는 다음에서 사용할 수 있습니다 [`panther_디택션_helpers`](https://pypi.org/project/panther-detection-helpers/).

{% hint style="warning" %}
값을 전달하지 않는 경우 `epoch_seconds`, 반드시 호출하세요 `set_key_expiration()` 를 받는 모든 함수 이후에 `epoch_seconds.`를 받는 함수가 `epoch_seconds` 이후에 호출되고 `set_key_expiration()` 에 대해 값이 제공되지 않으면 `epoch_seconds`, TTL은 기본값인 90일로 재설정됩니다.
{% endhint %}

만료 타임스탬프를 생성하려면, 다음을 통해 이벤트 시간과 연결된 유닉스 타임스탬프를 가져오세요 `event.event_time_epoch()`, 그리고 지정된 초 수를 더합니다. 결과 타임스탬프가 지나면 해당 행은 48시간 이내에 자동으로 삭제됩니다.

{% hint style="info" %}
TTL의 기준으로는 처리 시간(`p_event_time)` 보다 이벤트 시간(`p_parse_time` 또는 `datetime.datetime.now()`)을 사용하는 것이 권장됩니다. 이렇게 하면 이벤트 처리 지연을 고려할 수 있고, 단위 테스트에서 발견되는 것과 같은 오래된 이벤트가 캐시를 어지럽히지 않도록 보장할 수 있습니다.
{% endhint %}

#### 예제

Panther의 다음 예시 `Geographically Improbable Okta Login` 다음을 사용하는 디택션 `epoch_seconds`:

```python
# 테이블이 과거 사용자로 가득 차지 않도록 1주일 후 항목을 만료시킵니다
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()`:

```python
# 테이블이 과거 사용자로 가득 차지 않도록 1주일 후 항목을 만료시킵니다
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 콘솔에서의 테스팅

* Unit Test 호출은 해당 함수가 모의(mock)로 재정의되지 않는 한 DynamoDB와 통신합니다.
* DynamoDB로부터/로 전송되고 수신된 데이터는 다음에 커밋될 수 있습니다 `알러트_context()` Unit Test 결과에서 디버깅을 위해.
* DynamoDB의 원시 내용을 탐색하는 것은 불가능합니다.

### CLI 워크플로를 사용한 테스팅

* Panther의 디택션은 DynamoDB와 통신하기 위해 AWS IAM 역할을 활용합니다.
  * 다음을 사용할 때 [panther\_analysis\_tool](/ko/panther/detections-repo/pat.md) 를 사용하여 로컬에서 또는 CI/CD 워크플로의 일부로 Unit Tests를 실행할 때, 이 IAM 역할에는 접근할 수 없습니다.
  * Panther Console의 맥락 밖에서는 DynamoDB 캐시와 상호작용할 수 없으므로, 테스트는 입력과 출력을 시뮬레이션해야 합니다.
* CI/CD 워크플로를 지원하기 위해, 예상되는 출력을 시뮬레이션하도록 DynamoDB와 상호작용하는 모든 함수를 목 처리(mocking)할 것을 권장합니다.
  * 다음에 대한 Panther의 문서를 참조하세요 [Mocks에 대한 자세한 정보](/ko/detections/testing.md#mocks).

## 캐시를 사용할 때의 일반적인 함정

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

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

{% code title="나쁜 예" %}

```python
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')
            True를 반환합니다

    return False
```

{% endcode %}

이 디택션을 크게 개선할 수 있는 지점은 두 곳입니다:

1. 그것은 `BadGuys` 문자열 집합을 가져온 다음, 이것이 `BadGuyDetected` 이벤트인지 확인합니다. 만약 *아니라면* 나쁜 놈 이벤트이므로 문자열 집합을 가져올 필요가 없습니다. 이 호출은 항상 필요하지 않더라도 모든 디택션 실행에 지연 시간을 추가합니다.
2. 그것은 새 `bad_guy` 를 문자열 집합에 추가한 다음, 이것이 반복되는 나쁜 놈인지 확인합니다. 만약 반복되는 나쁜 놈이라면, 우리는 알러트를 발생시키고 집합을 초기화할 것이므로 문자열 집합에 추가할 필요가 없습니다.

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

```python
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')
            True를 반환합니다
            
        add_to_string_set('BadGuys', bad_guy) # <-- #2
        
    return False
```


---

# Agent Instructions: Querying This Documentation

If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter:

```
GET https://docs.panther.com/ko/detections/rules/python/caching.md?ask=<question>
```

The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.