# PantherFlow 예제: 위협 헌팅 시나리오
## 알러트에서 로그 검색으로 전환
우리가 하나를 받았다고 가정해 봅시다 [Wiz](broken://pages/fcac338bacf953ccd5a49609e4a2c00cf3d4dd6b) 잠재적으로 잘못 구성된 EC2 인스턴스에 대한 알러트입니다. 알러트에서 관련 AWS 인스턴스 ID를 가져올 수 있습니다. 그런 다음 해당 인스턴스의 활동을 찾기 위해 모든 AWS 로그를 검색할 수 있습니다.
```kusto
let 알러트_data = panther_signals.public.signal_알러트s
| where p_event_time > time.ago(7d)
| where p_알러트_id == '00411934608291e0fccd928590194fd6'
| 요약 instances = arrays.flatten(agg.make_set(p_any_aws_instance_ids)),
mintime = agg.min(p_event_time),
maxtime = agg.max(p_event_time);
유니온 panther_logs.public.aws*
| where p_event_time between time.parse_timestamp(toscalar(알러트_data | project mintime)) - 30m
.. time.parse_timestamp(toscalar(알러트_data | project maxtime)) + 30m
| where arrays.overlap(p_any_aws_instance_ids, toscalar(알러트_data | project instances))
```
위의 문은 다음을 활용합니다:
* [`let` 문](/ko/pantherflow/statements.md#let-statements) 기능
* 연산자: [`union`](/ko/pantherflow/operators/union.md), [`프로젝트`](/ko/pantherflow/operators/project.md), [`요약`](/ko/pantherflow/operators/summarize.md), [`where`](/ko/pantherflow/operators/where.md), 그리고 [`사이`](/ko/pantherflow/expressions.md#between-comparisons)
* 함수: [`arrays.flatten()`](https://docs.panther.com/ko/pantherflow/example-queries/pages/c58ae0e898fd6f293efee986d6d1707148384f17#arrays.flatten), [`arrays.overlap()`](https://docs.panther.com/ko/pantherflow/example-queries/pages/c58ae0e898fd6f293efee986d6d1707148384f17#arrays.overlap), [`agg.make_set()`](https://docs.panther.com/ko/pantherflow/example-queries/pages/c558177b40b82223aa5247e48b550d9e8e0ee9b1#agg.make_set), [`agg.min()`](https://docs.panther.com/ko/pantherflow/example-queries/pages/c558177b40b82223aa5247e48b550d9e8e0ee9b1#agg.min), [`agg.max()`](https://docs.panther.com/ko/pantherflow/example-queries/pages/c558177b40b82223aa5247e48b550d9e8e0ee9b1#agg.max), [`time.ago()`](https://docs.panther.com/ko/pantherflow/example-queries/pages/fd7f081827d975ce84e4194b201dbd8b50df09a1#time.ago), [`time.parse_timestamp()`](https://docs.panther.com/ko/pantherflow/example-queries/pages/fd7f081827d975ce84e4194b201dbd8b50df09a1#time.parse_timestamp), 그리고 [`toscalar()`](/ko/pantherflow/functions/other.md#toscalar)
## 한 테이블에서 IP를 가져와 다른 테이블에서 검색하기
위협 헌팅을 할 때, 한 테이블에서 값을 가져와 그 값을 다른 테이블에서 검색하도록 전환해야 하는 경우가 흔합니다. 아래 쿼리는 VPC Flow 로그 테이블에서 IP 주소를 가져온 다음, Okta System 로그에서 해당 IP를 검색합니다.
```kusto
let IPs = panther_logs.public.aws_vpcflow
| where p_event_time > time.ago(5d)
| where flowDirection == 'ingress'
| summarize IP = agg.make_set( re.substr(srcAddr, '(\\d{1,3})\\.(\\d{1,3})\\.(\\d{1,3})\\.(\\d{1,3})') );
panther_logs.public.okta_systemlog
| where p_event_time > time.ago(1d)
| where arrays.overlap(toscalar(IPs), p_any_ip_addresses)
```
위의 문은 다음을 활용합니다:
* [`let` 문](/ko/pantherflow/statements.md#let-statements) 기능
* 연산자: [`요약`](/ko/pantherflow/operators/summarize.md) 및 [`where`](/ko/pantherflow/operators/where.md)
* 함수: [`arrays.overlap()`](https://docs.panther.com/ko/pantherflow/example-queries/pages/c58ae0e898fd6f293efee986d6d1707148384f17#arrays.overlap), [`agg.make_set()`](https://docs.panther.com/ko/pantherflow/example-queries/pages/c558177b40b82223aa5247e48b550d9e8e0ee9b1#agg.make_set), [`re.substr()`](https://docs.panther.com/ko/pantherflow/example-queries/pages/2839050edbff540a165c3db723f0dcb8fedc898f#re.substr), [`time.ago()`](https://docs.panther.com/ko/pantherflow/example-queries/pages/fd7f081827d975ce84e4194b201dbd8b50df09a1#time.ago), 그리고 [`toscalar()`](/ko/pantherflow/functions/other.md#toscalar)
## 정규식과 함께 CIDR 일치
이 쿼리는 AWS 로그에서 정규식 표현식과 일치하는 IP 주소를 검색합니다.
```kusto
union aws_alb , amazon_eks_audit, aws_cloudtrail, aws_s3serveraccess
| where p_event_time > time.ago(7d)
| extend ip = coalesce(clientIp, sourceIPs, sourceIPAddress, remoteip)
| summarize events=agg.count() by ip, p_log_type
| where re.matches(ip, "^34\\.222\\..+\\..+$")
| sort events desc
```
위의 문은 다음을 활용합니다:
* 연산자: [`union`](/ko/pantherflow/operators/union.md), [`where`](/ko/pantherflow/operators/where.md), [`extend`](/ko/pantherflow/operators/extend.md), [`요약`](/ko/pantherflow/operators/summarize.md), 그리고 [`정렬`](/ko/pantherflow/operators/sort.md)
* 함수: [`coalesce()`](/ko/pantherflow/functions/other.md#coalesce), [`agg.count()`](https://docs.panther.com/ko/pantherflow/example-queries/pages/c558177b40b82223aa5247e48b550d9e8e0ee9b1#agg.count), 그리고 [`re.matches()`](https://docs.panther.com/ko/pantherflow/example-queries/pages/2839050edbff540a165c3db723f0dcb8fedc898f#re.matches)
결과:
| events | ip | p\_log\_type |
| ------ | ---------------- | ---------------- |
| `5866` | `34.222.253.62` | `AWS.CloudTrail` |
| `184` | `34.222.140.16` | `AWS.CloudTrail` |
| `176` | `34.222.42.181` | `AWS.CloudTrail` |
| `171` | `34.222.87.204` | `AWS.CloudTrail` |
| `88` | `34.222.241.235` | `AWS.CloudTrail` |
| ... | | |
## API 키 생성에 대한 알러트를 조사하기
이 시나리오에서는 Panther가 관리하는 [AWS 사용자 API 키 생성됨](https://github.com/panther-labs/panther-analysis/blob/main/rules/aws_cloudtrail_rules/aws_iam_user_key_created.yml) 디택션, 이 디택션은 CloudTrail 데이터를 기반으로 실행되며 다른 사용자가 AWS 사용자의 AWS API 키를 생성할 때 알러트를 발생시킵니다.
{% hint style="info" %}
이 위협 시나리오는 또한 [Risky Business 팟캐스트의 영상에서 단계별로 설명됩니다](https://risky.biz/video/product-demo-the-pantherflow-piped-query-language-for-the-panther-siem/)뿐만 아니라 Panther 블로그에서도: [PantherFlow를 사용한 Amazon EKS 권한 상승 조사](https://panther.com/blog/investigating-amazon-eks-privilege-escalation-with-pantherflow).
{% endhint %}
이 이벤트는 `ariel.ropek` 라는 새 사용자에 대한 API 키를 생성했음을 알려줍니다. `snidely-whiplash`. 이것이 오탐지인지 실제 침해인지 확인하기 위해 조사해 봅시다.
1. 먼저, 다음을 살펴봅시다 `ariel.ropek`의 알러트 전후 1시간 동안의 활동:
```kusto
panther_logs.public.aws_cloudtrail
| where p_event_time between time.parse_timestamp('2024-11-13 19:00') .. time.parse_timestamp('2024-11-13 20:00')
| extend p_actor = strings.split(userIdentity.arn, '/')[arrays.len(strings.split(userIdentity.arn, '/'))-1]
| where p_actor == 'ariel.ropek'
| sort p_event_time desc
```
\
결과:
흥미롭네요! 다음 사용자가 `ariel.ropek` 라는 새 사용자를 생성했을 뿐만 아니라 `snidely-whiplash`에 `AdministratorAccess` 정책도 연결한 것을 볼 수 있습니다:
2. 다음을 추가해 봅시다 `snidely-whiplash` 를 쿼리에 추가하여 해당 사용자의 활동을 확인해 봅시다:
```kusto
panther_logs.public.aws_cloudtrail
| where p_event_time between time.parse_timestamp('2024-11-13 19:00') .. time.parse_timestamp('2024-11-13 20:00')
| extend p_actor = strings.split(userIdentity.arn, '/')[arrays.len(strings.split(userIdentity.arn, '/'))-1]
| where p_actor in ['ariel.ropek', 'snidely-whiplash']
| sort p_event_time desc
```
\
결과:
다음이 포함되면 결과가 훨씬 많아집니다 `snidely-whiplash` —해당 사용자가 EKS에서 명령을 실행한 것을 확인할 수 있습니다. 새 역할을 만들고, `AmazonEKSClusterAdminPolicy` 를 여기에 연결한 다음, 새로운 세션 이름인 `snidely-whiplash-session`.
3. 으로 그 역할을 수임했습니다. 이제 이 사용자가 EKS에서 작업을 수행한 것을 알았으므로, 다음을 사용해야 합니다 [`union`](/ko/pantherflow/operators/union.md) 를 사용해 EKS Audit 및 Authenticator 로그를 포함하도록 검색 범위를 확장해야 합니다.\
\
CloudTrail, EKS Audit, EKS Authenticator 로그는 각각 서로 다른 스키마를 가지고 있지만, 다음을 사용할 수 있습니다 [`coalesce()`](/ko/pantherflow/functions/other.md#coalesce) 를 사용하여 이러한 로그 소스를 공통 `actor` 및 `action` fields:
```kusto
union panther_logs.public.aws_cloudtrail, panther_logs.public.amazon_eks_audit, panther_logs.public.amazon_eks_authenticator
| where p_event_time between time.parse_timestamp('2024-11-13 19:00') .. time.parse_timestamp('2024-11-13 20:00')
| extend p_aws_arn = coalesce(userIdentity.arn, user.username, arn)
| extend p_actor = strings.split(p_aws_arn, '/')[arrays.len(strings.split(p_aws_arn, '/'))-1]
| extend p_action = coalesce(eventName, strings.cat(verb, ' ', objectRef.resource), path)
| where p_actor in ['ariel.ropek', 'snidely-whiplash', 'snidely-whiplash-session']
| sort p_event_time desc
```
\
결과:
다음을 보면 `p_action` 열에서, 다음을 사용하여 `snidely-whiplash-session`, 사용자가 Kubernetes pod를 생성했음을 알 수 있습니다 (`create pods`). 전체 이벤트를 클릭해 보면, 그들이 생성한 pod가 특권 모드임을 알 수 있습니다:
\
이 시점에서 이것이 악의적인 행위자일 가능성이 매우 높다고 결론지을 수 있습니다. 요약하면, 우리는 다음을 발견했습니다:
1. 한 AWS 사용자 (`ariel-ropek`)가 새로운 사용자 (`snidley-whiplash`)를 관리자 권한과 함께 생성했습니다.
2. `snidley-whiplash` EKS로 피벗한 후 그곳에서 권한을 상승시켜 클러스터 관리자가 되었습니다.
3. 그들은 EKS에서 특권 pod를 생성했습니다.
4. 전체 공격 체인을 차트로 볼 수 있으며 [`시각화`](/ko/pantherflow/operators/visualize.md):
```kusto
union panther_logs.public.aws_cloudtrail, panther_logs.public.amazon_eks_audit, panther_logs.public.amazon_eks_authenticator
| where p_event_time between time.parse_timestamp('2024-11-13 19:00') .. time.parse_timestamp('2024-11-13 20:00')
| extend p_aws_arn = coalesce(userIdentity.arn, user.username, arn)
| extend p_actor = strings.split(p_aws_arn, '/')[arrays.len(strings.split(p_aws_arn, '/'))-1]
| extend p_action = coalesce(eventName, strings.cat(verb, ' ', objectRef.resource), path)
| where p_actor in ['ariel.ropek', 'snidely-whiplash', 'snidely-whiplash-session']
| summarize events = agg.count() by p_actor, p_action
| visualize bar xcolumn = p_action
| sort p_actor
```
\
결과:
위의 시각화에서는 오른쪽에서 왼쪽으로 공격 체인을 추적할 수 있으며, 다음으로 시작합니다 `ariel.ropek`의 동작.
---
# 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/pantherflow/example-queries/threat-hunting.md?ask=
```
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.