PantherFlow 예제: SOC 운영

마지막으로 데이터를 수신한 로그 소스 보기

union panther_logs.public.*
| where p_event_time > time.ago(1d)
| summarize last_received_data=agg.max(p_parse_time) by p_source_label, p_source_id
| sort last_received_data desc

이 쿼리는 다음을 활용합니다 summarize, sort및 agg.max().

예시 출력:

심각도별 경보 수

panther_signals.public.signal_alerts
| where p_event_time > time.ago(14d)
| summarize alert_count = agg.count() by severity
| extend sort_key = case(severity == "CRITICAL", 5, severity == "HIGH", 4, severity == "MEDIUM", 3, severity == "LOW", 2, severity == "INFO", 1)
| sort sort_key asc
| visualize bar orientation=horizontal

이 쿼리는 다음을 활용합니다:

• 연산자 where, summarize, extend, sort, visualize,

• 함수: time.ago(), agg.count()및 case()

예시 출력:

지난 2주 동안 일별 심각도별 경보

panther_signals.public.signal_alerts
| where p_event_time > time.ago(14d)
| extend bucket=time.trunc('day', p_event_time)
| summarize eventcount=agg.count() by bucket, severity
| summarize eventcount=agg.sum(eventcount) by bucket, severity
| extend severity_sort_key = case(severity == "CRITICAL", 5, severity == "HIGH", 4, severity == "MEDIUM", 3, severity == "LOW", 2, severity == "INFO", 1)
| sort bucket asc, severity_sort_key asc
| visualize line xcolumn=bucket, ycolumn=eventcount, series=severity, legend=bottom, title="Count of Panther Alerts Per Day Severity Over Last 14 days"

이 쿼리는 다음을 활용합니다:

• 연산자: where, extend, summarize, sort및 visualize

• 함수: time.ago(), time.trunc(), agg.count(), agg.sum()및 case()

예시 출력:\

심각도별 평균 해결 시간

let update_actions = panther_logs.public.panther_audit
| where actionName == 'UPDATE_ALERT_STATUS' and actionParams.dynamic.input.status == "RESOLVED" and p_event_time > time.ago(14d) 
| extend alertId = actionParams.dynamic.input.ids
| extend resolved_timestamp = timestamp
| project resolved_timestamp, actionParams, actionName, alertId;

let alerts = panther_signals.public.signal_alerts
| where p_event_time > time.ago(365d)
| project creationTime, updateTime, status, severity, alertId;

경고 
| join kind=inner record=(update_actions) on $left.alertId in $right.alertId
| extend resolved_timestamp = record.resolved_timestamp
| extend ttr = time.diff("m", creationTime, resolved_timestamp)
| summarize minutes=agg.avg(ttr) by severity
| extend severity_sort_key = case(severity == "CRITICAL", 5, severity == "HIGH", 4, severity == "MEDIUM", 3, severity == "LOW", 2, severity == "INFO", 1)
| sort severity_sort_key asc
| visualize bar orientation=horizontal, legend=right, title="Mean Time (minutes) to Resolution by Severity"

이 쿼리는 다음을 활용합니다:

• let 문장 기능

• 연산자: where, extend, project, join, summarize, sort및 visualize

• 함수: time.ago(), time.diff(), agg.avg()및 case()

예시 출력:

심각도별 시간대별 생성된 경보

다른 연산자와 결합하여 시계열 분석을 수행할 수 있습니다. 아래 예시는 Panther 감사 로그를 시간별 버킷으로 매핑합니다: 
let all_times = range N from 0 to 23 step 1 
| project bucket=time.add(time.now(), -1*N, "h")

let all_alerts = panther_signals.public.signal_alerts
| where p_event_time > time.ago(3d)
| summarize by severity;

| summarize by actor=actor.name;
| join kind=cross alerts=(all_alerts)
| project bucket, severity=alerts.severity, eventcount=0;

panther_signals.public.signal_alerts
| where p_event_time > time.ago(3d)
panther_logs.public.panther_audit
| summarize eventcount=agg.count() by bucket, severity
| summarize eventcount=agg.count() by bucket, actor=actor.name
| summarize eventcount=agg.sum(eventcount) by bucket, severity
| extend severity_sort_key = case(severity == "CRITICAL", 5, severity == "HIGH", 4, severity == "MEDIUM", 3, severity == "LOW", 2, severity == "INFO", 1)
| sort bucket asc, severity_sort_key asc
| visualize line xcolumn=bucket, ycolumn=eventcount, series=severity, legend=bottom, title="Count of Panther Alerts Per Day Severity Over Last day"

이 쿼리는 다음을 활용합니다:

• let 문장 기능

• 연산자: range, project, where, summarize, join, extend, union, sort및 visualize

• 함수: time.trunc(), time.ago(), agg.count(), agg.sum(), case()

예시 출력: