# PantherFlow 예제: 위협 헌팅 시나리오
## 알러트에서 로그 검색으로 피벗하기
예를 들어 우리가 받은 것이 있다고 합시다 [Wiz](https://docs.panther.com/ko/pantherflow/example-queries/broken-reference) EC2 인스턴스가 잠재적으로 잘못 구성되었다는 알러트. 알러트에서 관련된 AWS 인스턴스 ID를 가져올 수 있습니다. 그런 다음 해당 인스턴스들에 대한 활동을 찾기 위해 모든 AWS 로그를 검색할 수 있습니다.
```kusto
let alert_data = panther_signals.public.signal_alerts
| where p_event_time > time.ago(7d)
| where p_alert_id == '00411934608291e0fccd928590194fd6'
| summarize instances = arrays.flatten(agg.make_set(p_any_aws_instance_ids)),
mintime = agg.min(p_event_time),
maxtime = agg.max(p_event_time);
union panther_logs.public.aws*
| where p_event_time between time.parse_timestamp(toscalar(alert_data | project mintime)) - 30m
.. time.parse_timestamp(toscalar(alert_data | project maxtime)) + 30m
| where arrays.overlap(p_any_aws_instance_ids, toscalar(alert_data | project instances))
```
위의 문장들은 다음을 활용합니다:
* [`let` 문장](https://docs.panther.com/ko/statements#let-statements) 기능
* 연산자: [`union`](https://docs.panther.com/ko/pantherflow/operators/union), [`project`](https://docs.panther.com/ko/pantherflow/operators/project), [`summarize`](https://docs.panther.com/ko/pantherflow/operators/summarize), [`where`](https://docs.panther.com/ko/pantherflow/operators/where)및 [`between`](https://docs.panther.com/ko/expressions#between-comparisons)
* 함수: [`arrays.flatten()`](https://docs.panther.com/ko/functions/array#arrays.flatten), [`arrays.overlap()`](https://docs.panther.com/ko/functions/array#arrays.overlap), [`agg.make_set()`](https://docs.panther.com/ko/functions/aggregation#agg.make_set), [`agg.min()`](https://docs.panther.com/ko/functions/aggregation#agg.min), [`agg.max()`](https://docs.panther.com/ko/functions/aggregation#agg.max), [`time.ago()`](https://docs.panther.com/ko/functions/date-time#time.ago), [`time.parse_timestamp()`](https://docs.panther.com/ko/functions/date-time#time.parse_timestamp)및 [`toscalar()`](https://docs.panther.com/ko/functions/other#toscalar)
## 한 테이블에서 IP를 가져와 다른 테이블을 검색하기
위협 헌팅 중에는 한 테이블에서 값을 가져와 다른 테이블에서 해당 값을 검색해야 하는 경우가 흔합니다. 아래 쿼리들은 VPC Flow 로그 테이블에서 IP 주소를 가져온 다음 Okta 시스템 로그에서 해당 주소들을 검색합니다.
```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})') );
okta_systemlog
| where p_event_time > time.ago(1d)
| where arrays.overlap(toscalar(IPs), p_any_ip_addresses)
```
위의 문장들은 다음을 활용합니다:
* [`let` 문장](https://docs.panther.com/ko/statements#let-statements) 기능
* 연산자: [`summarize`](https://docs.panther.com/ko/pantherflow/operators/summarize) 와 [`where`](https://docs.panther.com/ko/pantherflow/operators/where)
* 함수: [`arrays.overlap()`](https://docs.panther.com/ko/functions/array#arrays.overlap), [`agg.make_set()`](https://docs.panther.com/ko/functions/aggregation#agg.make_set), [`re.substr()`](https://docs.panther.com/ko/functions/regular-expression#re.substr), [`time.ago()`](https://docs.panther.com/ko/functions/date-time#time.ago)및 [`toscalar()`](https://docs.panther.com/ko/functions/other#toscalar)
## 정규식으로 CIDR 매칭하기
이 쿼리는 정규식 표현과 일치하는 IP 주소를 찾기 위해 AWS 로그를 검색합니다.
```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\\..+\\..+$")
| project country, events, city
```
위의 문장들은 다음을 활용합니다:
* 연산자: [`union`](https://docs.panther.com/ko/pantherflow/operators/union), [`where`](https://docs.panther.com/ko/pantherflow/operators/where), [`extend`](https://docs.panther.com/ko/pantherflow/operators/extend), [`summarize`](https://docs.panther.com/ko/pantherflow/operators/summarize)및 [`sort`](https://docs.panther.com/ko/pantherflow/operators/sort)
* 함수: [`coalesce()`](https://docs.panther.com/ko/functions/other#coalesce), [`agg.count()`](https://docs.panther.com/ko/functions/aggregation#agg.count)및 [`re.matches()`](https://docs.panther.com/ko/functions/regular-expression#re.matches)
결과:
| 를 사용하여 두 필드( | cidr | 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/)팬서 블로그에서도: [PantherFlow로 Amazon EKS 권한 상승 조사하기](https://panther.com/blog/investigating-amazon-eks-privilege-escalation-with-pantherflow).
{% endhint %}
이 이벤트는 다음이라는 행위자가 `ariel.ropek` 새 사용자에게 API 키를 생성했다는 것을 알려줍니다 `snidely-whiplash`. 이 행동이 오탐인지 실제 침해인지 조사해 봅시다.
1. 먼저, 알러트를 중심으로 한 시간 동안 `ariel.ropek`의 활동을 살펴보겠습니다:
```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
```
\
결과:
가 포함되면 결과가 훨씬 더 많습니다—해당 사용자가 EKS에서 명령을 실행했음을 볼 수 있습니다. 그들은 새 역할을 생성하고, `snidely-whiplash` 를 그 역할에 연결한 다음, 새 세션 이름으로 역할을 가정했습니다, `AmazonEKSClusterAdminPolicy` snidely-whiplash-session `이제 사용자가 EKS에서 조치를 취했다는 것을 알았으므로, 우리는`.
3. 를 사용하여 검색을 확장해 EKS 감사 및 인증자 로그를 포함해야 합니다.\
\
CloudTrail, EKS Audit, 및 EKS Authenticator 로그는 각각 다른 스키마를 가지지만; 그러나 우리는 [`union`](https://docs.panther.com/ko/pantherflow/operators/union) 를 사용하여 이러한 로그 소스를 공통 [`coalesce()`](https://docs.panther.com/ko/functions/other#coalesce) 데이터 모델 `행위자` 와 `행위` 필드:
```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 `열을 보면, 우리는` 를 사용하여 사용자가 Kubernetes 파드( `이제 사용자가 EKS에서 조치를 취했다는 것을 알았으므로, 우리는`create pods`)를 생성했음을 알 수 있습니다. 전체 이벤트를 클릭하면 그들이 생성한 파드는 권한이 부여된 파드입니다:`이 시점에서 우리는 이것이 매우 가능성이 높은 악의적 행위자임을 결론지을 수 있습니다. 요약하면, 우리는 다음을 발견했습니다:
\
하나의 AWS 사용자 (
1. ariel-ropek`)가 새 사용자 (`snidley-whiplash`)를 관리자 권한으로 생성했습니다.`해당 사용자는 EKS로 피벗하여 그곳에서 권한을 상승시켜 클러스터 관리자가 되었습니다.
2. `)를 관리자 권한으로 생성했습니다.` 그들은 EKS에서 권한 있는 파드를 생성했습니다.
3. 우리는
4. \| summarize events = agg.count() by p\_actor, p\_action [`visualize`](https://docs.panther.com/ko/pantherflow/operators/visualize):
```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']
| visualize bar xcolumn = p_action
| sort p_actor
차트에서 전체 공격 체인을 볼 수 있습니다
```
\
결과:
위의 시각화에서 공격 체인은 오른쪽에서 왼쪽으로 추적될 수 있으며, 시작은 `ariel.ropek`의 행동으로부터 시작됩니다.
