# PantherFlow 예제: 위협 헌팅 시나리오
## 알러트에서 로그 검색으로 전환
우리가 하나를 받았다고 가정해 봅시다 [Wiz](https://docs.panther.com/ko/pantherflow/example-queries/broken-reference) 잠재적으로 잘못 구성된 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` 문](https://docs.panther.com/ko/statements#let-statements) 기능
* 연산자: [`union`](https://docs.panther.com/ko/pantherflow/operators/union), [`프로젝트`](https://docs.panther.com/ko/pantherflow/operators/project), [`요약`](https://docs.panther.com/ko/pantherflow/operators/summarize), [`where`](https://docs.panther.com/ko/pantherflow/operators/where), 그리고 [`사이`](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 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` 문](https://docs.panther.com/ko/statements#let-statements) 기능
* 연산자: [`요약`](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 일치
이 쿼리는 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`](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), [`요약`](https://docs.panther.com/ko/pantherflow/operators/summarize), 그리고 [`정렬`](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)
결과:
| 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`](https://docs.panther.com/ko/pantherflow/operators/union) 를 사용해 EKS Audit 및 Authenticator 로그를 포함하도록 검색 범위를 확장해야 합니다.\
\
CloudTrail, EKS Audit, EKS Authenticator 로그는 각각 서로 다른 스키마를 가지고 있지만, 다음을 사용할 수 있습니다 [`coalesce()`](https://docs.panther.com/ko/functions/other#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. 전체 공격 체인을 차트로 볼 수 있으며 [`시각화`](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']
| summarize events = agg.count() by p_actor, p_action
| visualize bar xcolumn = p_action
| sort p_actor
```
\
결과:
위의 시각화에서는 오른쪽에서 왼쪽으로 공격 체인을 추적할 수 있으며, 다음으로 시작합니다 `ariel.ropek`의 동작.
