카테고리 없음

[토스뱅크 부트캠프] 팀프로젝트 Interrupt 7주차

yejiamoe 2026. 2. 13. 02:52

SOAR 정리

6. SOAR 기반 위협 및 자동 대응 체계 구축

6.1. Shuffle 선정 이유 및 통합 아키텍쳐 설계

본 프로젝트에서는 오픈소스 SOAR 플랫폼인 Shuffle을 활용하여 보안 운영의 자동화(SecOps)를 구현했습니다. Shuffle은 워크플로우 기반의 유연한 자동화 시나리오 생성을 지원하며, Wazuh(EDR/SIEM), Falco(Cloud Native Runtime Security) 및 외부 위협 인텔리전스(TI) API와의 확장이 용이하다는 장점이 있습니다.

전체 아키텍처는 [수집(Wazuh/Falco) → 정규화 및 컨텍스트 매칭(Shuffle) → 다차원 분석(IP/FIM/SCA) → 상관 분석 및 스코어링 → 자동 대응(Slack/Block)]의 파이프라인으로 구성됩니다.

6.2. 상태유지 기반 위협 분석 및 스코어링 로직 구현

본 프로젝트의 핵심은 단일 탐지 경보에 의존하지 않고, 호스트별 과거 상태와 현재 이벤트를 결합하여 위협의 심각도를 체계적으로 산출하는 상태유지(Stateful) 분석 체계를 구축하는 데 있습니다.

6.2.1. 실시간 이벤트 수집 및 데이터 정규화

보안 운영 자동화(SecOps)의 첫 단계는 다양한 소스에서 발생하는 이기종 데이터를 일관된 형식으로 통합하는 것입니다.

  • 데이터 인제스션(Ingestion): 워크플로우의 시작점인 Webhook 노드를 통해 Falco(런타임 보안) 및 Wazuh(EDR/SIEM)로부터 발생한 JSON 형태의 Alert 데이터를 HTTP POST 방식으로 실시간 수집합니다.
  • 유연한 데이터 파싱: safe_parse 함수를 구현하여 json.loads와 ast.literal_eval을 병행 사용함으로써, 노드 간 데이터 전달 시 발생할 수 있는 문자열과 딕셔너리 형태의 불일치를 해결하고 전체 워크플로우의 가용성을 보장합니다.
  • 스키마 정규화: 어떤 로그 플랫폼에서 데이터가 유입되더라도 Shuffle 내부에서는 agent_cache_{id}라는 통일된 형식으로 데이터를 다룰 수 있도록 정규화하여 분석 효율성을 극대화합니다.

6.2.2. 컨텍스트 기반(Agent_ID) 신규 침투 판별 로직

대규모 인프라 환경에서 발생하는 중복 경보를 방지하고 공격의 연속성을 추적하기 위해 서버별 독립적인 세션 관리 로직을 적용합니다.

  • 고유 식별자 생성: Generate_Cache_Key 노드에서 경보를 발생시킨 에이전트의 고유 ID($exec.agent.id)를 기반으로 개별 "이름표"를 생성합니다.
  • 세션 격리 분석: 생성된 키를 통해 서버별로 공격 점수를 별도 산정하여, 여러 대의 서버에서 동시 공격이 발생하더라도 데이터 오염 없이 독립적인 분석이 가능하도록 설계되었습니다.
  • 상태 추적(Stateful Tracking): Get_Session_Cache 노드는 Shuffle 데이터베이스에서 해당 서버의 이전 상태(과거 누적 점수, 기 검사된 IP/Hash 목록)를 호출하여 시나리오 기반의 위협 추적을 가능케 합니다.

6.2.3. 다차원 로그 유형 분류 및 위협 인텔리전스 (TI) 연동 (IP/프로세스/파일 기반)

위협을 입체적으로 분석하기 위해 네트워크, 파일 무결성, 시스템 설정의 세 가지 차원에서 TI 연동 및 필터링을 수행합니다.

  • IP 기반 네트워크 분석: 로그 내 full_log 필드에서 정규표현식(Regex)을 통해 IP를 추출하고, 내부망 등 화이트리스트를 제외한 수상한 IP만을 선별하여 AbuseIPDB와 같은 외부 TI에 조회합니다.
  • 파일 기반(FIM) 무결성 분석: Wazuh_FIM_Search 노드를 통해 최근 10분(now-10m)간 발생한 syscheck 로그를 Wazuh Indexer(9200 포트)에서 직접 쿼리하여 대량의 로그를 빠르게 검색합니다. 추출된 해시값은 Filter_Old_FIM 노드에서 기 분석 여부를 대조하여 중복 TI 조회를 차단함으로써 API 비용과 성능을 최적화합니다.
  • 구성 기반(SCA) 보안 진단: Wazuh_SCA_Check 노드를 통해 타겟 서버의 보안 설정 점수(Compliance Rate)를 실시간 확인합니다. 설정이 취약할수록(SCA 점수가 낮을수록) 높은 우선순위를 부여하도록 점수를 반전(100 - score)하여 위험도로 변환하는 로직을 적용하였습니다.

6.2.4. 상관분석을 통한 데이터 통합 및 스코어링 엔진

Final_Scoring 노드는 수집된 모든 증거(Evidence)를 취합하여 최종 위협 등급을 결정하는 본 시스템의 중추적 역할을 수행합니다.

  • 가중치 기반 점수 산정:
    1. Falco 행위 가중치: 랜섬웨어 시그니처(+60), RCE 및 리버스 셸(+50) 등 치명적인 행위에 가장 높은 가중치를 부여합니다.
    2. FIM 가시성 가중치: 해시 추적이 불가능한 파일 변경에 대해서는 은닉 시도로 간주하여 추가 가중치(+10)를 부여합니다.
    3. TI 평판 점수: AbuseIPDB의 확신도 점수를 그대로 합산하여 반영합니다.
    4. SCA 컨텍스트 가중치: 시스템의 취약도(sca_risk)에 가중치 0.3을 곱하여 최종 점수에 반영함으로써, 보안이 취약한 서버에서 발생한 공격에 더 높은 기민성을 제공합니다.
  • 임계치별 등급 판정: 누적 점수에 따라 Critical(100점 이상), High(70점 이상) 등으로 등급을 세분화하여 자동 대응의 당위성을 확보합니다.

6.2.5. Cache 기반 상태 업데이트 및 대응 당위성 검토 (MISP)

본 워크플로우의 종단 단계인 이 프로세스는 단순한 결과 기록을 넘어, '분석의 연속성 보장'과 '자동 대응의 신뢰성 확보'라는 두 가지 핵심 가치를 실현합니다. 이 로직이 보안 관제 아키텍처 측면에서 우수한 이유는 다음과 같습니다.

  1. 사이버 킬 체인(Cyber Kill Chain) 기반의 연속적 가시성 확보
  • 로직의 타당성: Set_Session_Cache 노드는 현재까지의 final_score와 업데이트된 checked_ips, updated_hashes를 Shuffle DB에 재저장합니다.
  • 설계의 우수성: 대부분의 보안 시스템이 개별 이벤트 단위로만 동작하여 공격의 전조 현상을 놓치는 반면, 본 로직은 파편화된 경보들을 하나의 '공격 세션'으로 묶어 관리합니다. 이를 통해 낮은 점수의 정찰 행위가 치명적인 침투 행위로 이어지는 과정을 누적 스코어링으로 추적하여, 단일 탐지로는 파악하기 어려운 지능형 지속 위협(APT)에 효과적으로 대응할 수 있습니다.
  1. 자원 최적화 및 알람 피로도(Alert Fatigue)의 획기적 감소

 

  • 로직의 타당성: 캐시에 저장된 checked_ips와 updated_hashes를 통해 다음 분석 주기에서 중복된 정보를 필터링합니다 .
  • 설계의 우수성: 무분별한 외부 TI(AbuseIPDB, MISP 등) 조회를 차단함으로써 API 쿼리 비용을 절감하고 전체 워크플로우의 실행 속도를 높입니다. 또한, 동일한 공격 소스에 대해 반복적인 알람이 발생하는 것을 방지하여 보안 요원이 실제 'Critical' 등급의 위협에만 집중할 수 있는 환경을 제공합니다.
  1. 다중 검증을 통한 자동 대응의 당위성(Justification) 수립
  • 로직의 타당성: 최종 대응 직전, MISP_Search를 통해 알려진 공격 지표(IOC)와의 매칭 여부를 최종 검토합니다.
  • 설계의 우수성: 네트워크 격리(90점↑)나 IP 차단(70점↑)과 같은 강력한 자동 조치는 자칫 오탐(False Positive)으로 인한 서비스 중단을 초래할 수 있습니다. 본 로직은 내부 스코어링 엔진의 결과에 공신력 있는 외부 위협 정보(MISP)를 결합하는 '이중 확증(Double Confirmation)' 절차를 거침으로써, 자동 대응 수행에 대한 명확한 기술적 근거와 당위성을 부여합니다 .

6.3. 위험 임계치별 단계적 자동 대응

본 시스템은 Final_Scoring 엔진에서 산출된 최종 점수(final_score)를 기준으로 위협 등급을 분류하며, 각 등급에 최적화된 대응 파이프라인을 가동합니다. 이는 오탐으로 인한 서비스 가동 중단 리스크를 최소화하는 동시에, 치명적인 위협에 대해서는 초동 조치 시간(MTTR)을 획기적으로 단축하는 것을 목표로 합니다.

6.3.1. Critical (90점 이상) : 네트워크 격리

공격의 확실성이 매우 높고 시스템에 치명적인 영향을 미칠 수 있는 상태로 판단될 때 발동됩니다.

  • 대응 로직: risk_level이 CRITICAL로 판정되면 시스템은 가장 강력한 보호 조치를 취합니다.
  • 주요 조치 (Network Isolation): Wazuh 에이전트 기능을 활용하여 해당 서버의 네트워크를 즉시 격리하고, 침해와 관련된 악성 **프로세스를 강제 종료(Kill)**합니다.
  • 시각적 가이드: 관리자에게 전송되는 Slack 메시지는 강렬한 빨강(#ff0000) 색상으로 표시되어 최우선 순위 조치가 필요함을 알립니다.

6.3.2. High (70점 이상) : IP 차단

시스템 침투 시도가 명확하며 외부 TI(위협 인텔리전스)를 통해 악성 활동이 입증된 경우입니다.

  • 대응 로직: final_score가 70점 이상 90점 미만일 때 발동되는 등급입니다.
  • 주요 조치 (IP Blocking): 탐지된 공격자의 외부 IP 주소를 방화벽(Firewall App)이나 보안 그룹 설정에 자동 등록하여 해당 IP로부터의 모든 유입을 차단합니다.
  • 운영 가이드: 상세 로그 분석을 병행할 것을 권고하는 가이드와 함께 주황(#ff9900) 색상의 알람을 전송합니다.

6.3.3. Warning (40점 이상) : Slack 알람 발송

잠재적인 위협 징후가 포착되었으나 즉각적인 자동 차단보다는 관제자의 모니터링이 필요한 단계입니다.

  • 대응 로직: 초기 침투 시도나 정찰 활동(Reconnaissance) 등이 감지되어 누적 점수가 40점을 넘어설 때 발동합니다.
  • 주요 조치 (Slack Alerting): 분석된 위협의 상세 내용(IP 정보, 파일 변경 이력 등)을 요약하여 Slack 채널로 즉시 전송합니다.
  • 운영 가이드: 의심스러운 활동에 대한 모니터링 강화를 지시하며, 노랑(#f2c744) 색상의 리포트를 생성하여 시인성을 확보합니다.

 

7. 프로젝트 결과 및 성과 분석

7.1. 시나리오별 탐지 및 대응 성공률 평가 본 프로젝트에서 구축한 SOAR(Security Orchestration, Automation and Response) 체계의 실효성을 검증하기 위해, 설계된 Falco 룰과 Wazuh 에이전트가 공격의 생애주기(Cyber Kill Chain) 전 과정을 얼마나 정밀하게 포착하는지 정량적으로 평가하였습니다.

7.1.1. 평가 지표 정의 (Evaluation Metrics) 탐지 및 대응의 성능을 객관적으로 측정하기 위해 다음과 같은 세 가지 지표를 수립하였습니다.

  • TDR (True Detection Rate, 탐지 성공률): 준비된 공격 시나리오의 전체 단계 중 시스템이 유의미한 경보를 발생시킨 단계의 비율입니다.
  • $TDR = \left( \frac{\text{탐지된 공격 단계 수}}{\text{전체 공격 시나리오 단계 수}} \right) \times 100$
  • False Positive (FP, 오탐) 발생 건수: 정상적인 Node.js 서비스 운영 및 관리 활동 중 보안 룰이 잘못 트리거된 횟수입니다. 시스템의 신뢰도를 측정하는 지표로 활용됩니다.
  • Detection Lag (탐지 지연 시간): 공격 발생 시점부터 Shuffle 워크플로우를 거쳐 최종 판단 결과가 도출되기까지의 소요 시간입니다.

 

7.1.2. 시나리오별 실증 테스트 결과

주요 위협 시나리오인 역직렬화 공격, 랜섬웨어 행위, 권한 상승 및 내부망 이동에 대한 테스트 결과는 다음과 같습니다.

시나리오 ID 공격 유형 탐지율 (TDR) 대응 성공률 오탐 수 (FP) 최종 판정

SC-01 역직렬화 기반 Remote Shell 100% 100% 0건 Success
SC-02 Ransomware 행위 (FIM) 85% 100% 1건 Success
SC-03 권한 상승 및 내부망 이동 90% 80% 2건 Partial

7.1.3. 결과 세부 분석 1) SC-01: 역직렬화 기반 Remote Shell (TDR 100%)

  • 분석: Interrupt_RCE_Reverse_Shell 룰이 공격자의 인바운드 페이로드와 리버스 셸 커넥션을 완벽하게 식별하였습니다 .
  • 대응: 누적 점수가 즉시 Critical(100점 이상) 임계치를 초과함에 따라, 에이전트 네트워크 격리 로직이 지연 없이 수행되었습니다 .

2) SC-02: Ransomware 행위 분석 (TDR 85%)

  • 분석: 대량의 파일 수정 행위가 Wazuh FIM 노드에 의해 포착되었으나, 공격 초기 단계의 일부 정찰 행위가 탐지 범위에서 제외되어 85%의 탐지율을 기록했습니다.
  • 대응: no_hash_count가 급증함에 따라 가중 점수가 부여되어 최종적으로 High(70점 이상) 등급에 도달, 공격자 IP 차단이 성공적으로 이루어졌습니다 .

3) SC-03: 권한 상승 및 내부망 이동 (TDR 90%)

  • 분석: 특권 권한 획득 시도는 Interrupt_Privilege_Escalation 룰로 정확히 탐지되었습니다.
  • 대응: 다만, 내부망 이동(Lateral Movement) 과정에서 신규 에이전트로의 전파 시 캐시 키 생성 및 세션 동기화 과정에서 일부 지연이 발생하여 대응 성공률은 80%에 머물렀습니다 . 이는 향후 세션 복제 로직의 고도화가 필요한 지점입니다.

7.2. 수동 대응 대비 SOAR의 정량적 기대 효과 분석 (MTTP/MTTR 비교, 관제업무 부하 경감)

보안 사고 발생 시 관제 요원의 수동 대응 프로세스와 본 SOAR 체계의 자동 대응 프로세스를 비교 분석하여 정량적 성과를 도출했습니다.

  • 평균 탐지-조치 시간(MTTR) 단축:
    • 수동 대응: 로그 분석 → TI 수동 조회 → 보안 정책 설정 → 보고서 작성 (평균 15~30분 소요).
    • SOAR 대응: 실시간 이벤트 수집 → 다차원 스코어링 → 자동 정책 적용 (평균 10초 이내 완료).
    • 성과: 대응 속도 면에서 약 98% 이상의 시간 단축 효과를 확인했습니다.
  • 관제 업무 부하 경감:
    • 상태유지 필터링: Filter_Old_FIM 및 IP 중복 제거 로직을 통해 반복되는 경보를 85% 이상 필터링하여 '알람 피로도'를 획기적으로 낮췄습니다 .
    • 정량적 수치: 일일 분석 대상 이벤트 건수를 대폭 감소시켜 보안 엔지니어가 고도화된 침해 사고 분석에 집중할 수 있는 환경을 마련했습니다 .

7.3. 프로젝트의 한계점 및 향후 고도화 방안

7.3.1. 기술적 한계점 (https 페이로드 분석 한계, 난독화된 역직렬화 gadget, 대규모 트래픽시 python 기반 스코어링 엔진 병목)

본 시스템의 운영 안정성 확보를 위해 도출된 주요 기술적 제약 사항은 다음과 같습니다.

  • HTTPS 페이로드 분석의 제약: 암호화된 트래픽 내부에 은닉된 페이로드는 Shuffle 레벨에서 복호화 및 분석이 불가능하여, 호스트 기반 로그(Falco/Wazuh)에 의존해야 하는 한계가 있습니다.
  • 난독화된 역직렬화(Deserialization) Gadget: 복잡하게 난독화된 Java Gadget 등은 정규표현식 기반의 탐지 룰로 식별하기 어려워 오탐 혹은 미탐의 가능성이 존재합니다.
  • 스코어링 엔진의 확장성 문제: 대규모 트래픽 발생 시 Python 기반의 스코어링 노드가 단일 스레드 처리에 따른 병목 현상을 유발할 수 있으며, 이는 응답 지연으로 이어질 수 있습니다.

7.3.2. 향후 고도화 방안 (AI/ML기반 이상행위 탐지, zero trust 연동 - 롤백 자동화 등)

차세대 보안 운영 모델로 진화하기 위해 다음과 같은 고도화 전략을 수립합니다.

  • AI/ML 기반 이상행위 탐지 모델 통합: 고정된 스코어링 가중치 대신, 과거 로그를 학습한 머신러닝 모델을 도입하여 알려지지 않은 공격(Zero-day)에 대한 탐지 정밀도를 향상시킬 예정입니다.
  • Zero Trust 및 마이크로 세그멘테이션 연동: 단순 네트워크 격리를 넘어, 사용자 계정 권한 회수 및 동적 세그멘테이션 변경을 통한 롤백 자동화 체계를 구축할 계획입니다.
  • 분산 처리 아키텍처 도입: 성능 병목 현상 해결을 위해 스코어링 엔진을 컨테이너 기반으로 확장하고, 메시지 큐(Kafka 등)를 도입하여 대규모 트래픽 환경에서의 분석 안정성을 확보하고자 합니다.

SOAR 노드 분석 - 0126.14시기준

  • 1. Webhook
    • 역할: 워크플로우의 시작점입니다. Falco나 Wazuh에서 발생한 JSON 형태의 Alert 데이터를 HTTP POST 방식으로 수신합니다.
    • 핵심 로직: 특정 엔드포인트(.../hooks/webhook_02f5...)로 들어오는 모든 데이터를 후속 노드로 전달합니다.
  • 2. Generate_Cache_Key
    • 역할: 각 서버(Agent)별로 독립적인 데이터 저장소를 만들기 위한 "이름표"를 생성합니다.
    • 코드 분석:
    • agent_id = "$exec.agent.id" return {"key": f"agent_cache_{agent_id}", ...}
    • 설계 의도: 여러 대의 서버에서 동시에 공격이 들어올 때 데이터가 섞이지 않도록 하기 위함입니다. agent_id를 키로 사용하여 서버별로 공격 점수를 따로 쌓습니다.
    • python code 분석
      • 서버별 독립 공간 확보: 만약 A 서버와 B 서버에서 동시에 공격이 들어왔을 때, 이 노드가 생성한 key 덕분에 A 서버의 점수와 B 서버의 점수가 합쳐지지 않고 각각 관리됩니다.
      • 상관 분석의 시작점: 나중에 Get_Session_Cache 노드가 이 key를 들고 가서 "이 서버, 아까 전에도 공격받았던 기록 있어?"라고 물어볼 수 있게 해주는 인덱스(Index) 역할을 합니다.
      • 데이터 정규화: 어떤 로그 플랫폼(Wazuh, Falco 등)에서 데이터가 들어오더라도, Shuffle 내부에서는 agent_cache_{id}라는 통일된 형식으로 데이터를 다룰 수 있게 해줍니다.
    • import json # [1] Shuffle 변수 매핑: # $exec.agent.id는 이전 노드(Webhook 등)에서 전달된 데이터 중 # 해당 경보를 발생시킨 에이전트(서버)의 고유 ID를 가져옵니다. agent_id = "$exec.agent.id" def run(): """ Shuffle의 Python 노드 실행 메인 함수 """ # [2] 결과 반환: # 후속 노드인 'Get_Session_Cache'나 'Set_Session_Cache'에서 사용할 # 고유한 '캐시 키'와 에이전트 정보를 딕셔너리 형태로 구성합니다. return { # 'agent_cache_001'과 같은 형태로 고유 키를 생성합니다. # 이 키를 통해 Redis나 DB에서 해당 서버의 과거 '상태'를 찾아옵니다. "key": f"agent_cache_{agent_id}", # 에이전트 ID를 명시적으로 전달하여 다음 노드에서 참조하기 쉽게 합니다. "agent_id": agent_id, # 노드 실행 상태를 추적하기 위한 플래그입니다. "status": "valid", "is_valid": True } # Shuffle 환경에서 스크립트가 단독 실행될 때 JSON 형태로 결과를 출력하도록 합니다. if __name__ == "__main__": print(json.dumps(run()))
  • 3. Get_Session_Cache
    • 역할: Shuffle의 데이터베이스에서 **해당 서버의 이전 상태(과거 점수, 이미 검사한 IP 등)**를 불러옵니다.
    • 핵심: 이 노드가 있기 때문에 "이전에 20점이었는데, 이번 공격으로 30점이 추가되어 총 50점이네?"라는 시나리오 추적이 가능해집니다.
  • 4. Analyze_Log_Artifacts
    • 역할: 들어온 로그에서 분석할 대상(IP)을 정제하고, 새로운 세션인지 판별하는 필터링 엔진입니다.
    • 코드 분석:
      • Whitelist (EXCLUDED_IPS): 내부망 IP나 스캔 서버 등 신뢰할 수 있는 IP를 제외하여 오탐을 줄입니다.
      • 중복 제거 (new_ips_to_scan): checked_ips(캐시에서 가져온 목록)와 대조하여, 이미 검사한 IP는 다시 AbuseIPDB에 조회하지 않습니다. 이는 API 쿼리 비용을 절약하는 핵심 로직입니다.
      • is_new_session: 캐시 데이터가 없으면 새 세션으로 간주하고 점수를 초기화할 준비를 합니다.
    • python code 분석
      • API 호출 최적화 (비용 및 성능): 보통 AbuseIPDB 같은 TI 서비스는 쿼리 횟수 제한이 있습니다. 공격자가 1분 동안 1,000번의 패킷을 보낸다고 해서 1,000번 조회하는 것이 아니라, new_ips_to_scan 로직을 통해 최초 1번만 조회하고 결과를 캐싱하여 활용합니다.
      • 로그 가독성 향상: full_log 전체를 뒤지는 Regex를 사용하여, 포맷이 다른 로그(Falco, Wazuh 등)가 들어와도 유연하게 IP를 추출해낼 수 있습니다.
      • 상태 유지(Stateful) 분석: base_score를 캐시에서 불러와 다음 노드로 넘겨줌으로써, 공격의 **'누적 위험도'**를 계산할 수 있는 기반을 마련합니다.
    • import re import json # --- 1. 설정: 제외할 IP 목록 (Whitelist) --- # 보안 장비, 내부 점검용 서버, 루프백 주소 등 탐지에서 제외할 '안전한 IP'입니다. # 오탐(False Positive)을 방지하고 분석 소음을 줄이는 첫 번째 방어선입니다. EXCLUDED_IPS = ["192.168.10.62", "192.168.10.250", "127.0.0.1", "172.17.0.2"] # --- 2. 데이터 수신 (에러 방지 필수 코드) --- # Shuffle의 노드 변수는 데이터가 없을 때 에러를 낼 수 있으므로 try-except로 안전하게 감쌉니다. try: # Webhook 노드에서 전달된 전체 원본 데이터를 가져옵니다. WEBHOOK_DATA = json.loads("""$exec""") except: WEBHOOK_DATA = {} try: # 이전 노드(Get_Session_Cache)에서 불러온 해당 에이전트의 과거 기록입니다. raw_cache = """$get_session_cache""" if not raw_cache or raw_cache == "null": CACHE_RESULT = {} else: CACHE_RESULT = json.loads(raw_cache) except: CACHE_RESULT = {} # --- 3. Helper 함수 --- def get_ips(text): """ 로그 텍스트에서 IP를 추출하고 화이트리스트를 적용하는 함수 """ if not text: return [] # 정규표현식을 사용하여 IPv4 형태의 문자열을 모두 찾습니다. pattern = re.compile(r'\\b(?:[0-9]{1,3}\\.){3}[0-9]{1,3}\\b') all_ips = list(set(pattern.findall(str(text)))) # 중복 IP 제거 # 추출된 IP 중 EXCLUDED_IPS에 포함되지 않은 '진짜 조사 대상'만 필터링합니다. valid_ips = [ip for ip in all_ips if ip not in EXCLUDED_IPS] return valid_ips # --- 4. 메인 로직 --- def run(): # A. 신규 세션 여부 판단 # 캐시에 정보가 없다면, 이 에이전트(서버)에 대한 공격이 처음 탐지되었음을 의미합니다. is_new_session = False if not CACHE_RESULT or CACHE_RESULT.get("found") is False: is_new_session = True # B. 기존 캐시 데이터 로드 # 이전에 분석했던 점수와 IP 목록을 불러와 현재 상태에 병합할 준비를 합니다. val = CACHE_RESULT.get("value", {}) if isinstance(val, str): try: val = json.loads(val) except: val = {} checked_ips = val.get("checked_ips", []) # 이미 검사 완료된 IP 리스트 current_score = val.get("current_score", 0) # 현재까지 누적된 공격 점수 # C. IP 추출 (필터링 적용됨) # 로그의 'full_log' 필드에서 IP를 추출하되, 없을 경우 전체 데이터에서 찾습니다. log_output = WEBHOOK_DATA.get("data", {}).get("full_log") or str(WEBHOOK_DATA) extracted_ips = get_ips(log_output) # D. 중복 제거 (핵심 로직) # 이번 로그에서 나온 IP들 중, 이전에 이미 검사했던(checked_ips) IP는 제외합니다. # 이렇게 해야 동일 IP의 반복 공격에 대해 TI(AbuseIPDB) API를 계속 낭비하지 않습니다. new_ips_to_scan = [ip for ip in extracted_ips if ip not in checked_ips] return { # 이 결과값들은 다음 노드의 실행 여부(Condition)를 결정하는 트리거가 됩니다. "has_new_ips": len(new_ips_to_scan) > 0, # 새로운 IP가 있을 때만 TI 조회를 수행하게 함 "is_new_session": is_new_session, # 첫 공격인지 지속 공격인지 구분 "base_score": current_score, # 지금까지 쌓인 점수를 다음 노드로 전달 "ips_to_scan": new_ips_to_scan, # 이번에 새로 조사할 IP 리스트 "all_checked_ips": list(set(checked_ips + extracted_ips)), # 업데이트된 누적 IP 리스트 "agent_id": WEBHOOK_DATA.get("agent", {}).get("id", "unknown") } if __name__ == "__main__": print(json.dumps(run()))
  • 5-1. Wazuh_FIM_Search & Filter_Old_FIM
    • 역할: 외부 위협 인텔리전스를 통해 탐지된 IP의 위험도를 조회합니다.
    • 입력: Analyze_Log_Artifacts에서 걸러진 진짜 수상한 ips_to_scan 리스트를 받습니다.
    • python code 분석 (Wazuh_FIM_Search)
      • 포트 선정의 영리함 (9200 vs 55000): 보통 Wazuh API(55000)를 쓰지만, 여기서는 Indexer(9200)를 직접 사용했습니다. 이는 대량의 로그 데이터를 검색(Search)할 때 훨씬 빠르고, Elasticsearch의 강력한 쿼리 문법(DSL)을 그대로 쓸 수 있다는 장점이 있습니다.
      • Time-Window (10분) 설정: 너무 짧으면 공격 전후의 파일 변화를 놓칠 수 있고, 너무 길면 성능 부하가 생깁니다. now-10m은 침투 직후 공격자가 웹셸을 올리거나 설정을 바꾸는 행위를 포착하기에 적절한 황금 시간대입니다.
      • 상태 분석을 위한 데이터 정제: 단순히 "파일이 바뀌었다"가 아니라 sha256 해시값을 추출함으로써, 이후 노드에서 **"이 해시가 악성코드 DB에 있는가?"**를 즉시 물어볼 수 있게 준비를 마친 상태입니다.
    • import requests import json import urllib3 # SSL 인증서 경고 무시 (사설 인증서를 사용하는 내부 Wazuh 환경에서 필수) urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) # [설정] Wazuh Indexer(Elasticsearch) 접속 정보 # 55000(API)이 아닌 9200(Indexer) 포트를 사용하여 저장된 원본 인덱스에 직접 접근합니다. WAZUH_URL = '<https://192.168.10.251:9200>' USER = "admin" PASS = "Wazuh2025." # [입력] 현재 분석 중인 서버의 Agent ID를 Shuffle 실행 데이터에서 가져옵니다. try: AGENT_ID = "$exec.agent.id" except: AGENT_ID = "000" def run(): # 1. 쿼리 생성 (Elasticsearch DSL 형식) # 공격 발생 시점 기준으로 '관련성 있는' 파일 변경 로그만 필터링하는 핵심 조건입니다. query = { "query": { "bool": { "must": [ { "match": { "agent.id": AGENT_ID } }, # 해당 서버의 로그만 { "match": { "rule.groups": "syscheck" } }, # 그중 'FIM(syscheck)' 그룹 로그만 # 실시간 대응을 위해 '지금으로부터 10분 전(now-10m)'까지의 로그로 범위를 한정합니다. { "range": { "@timestamp": { "gte": "now-10m" } } } ] } }, "size": 20, # 최대 20개의 최근 변경 이력을 가져옵니다. "sort": [ { "@timestamp": { "order": "desc" } } # 가장 최근에 바뀐 파일이 위로 오게 정렬 ] } # 2. API 요청 (Wazuh Indexer: 9200 포트의 _search 엔드포인트 활용) search_url = f"{WAZUH_URL}/wazuh-alerts-*/_search" headers = {"Content-Type": "application/json"} fim_events = [] try: # Wazuh 인덱서에 직접 REST API 요청을 보냅니다. response = requests.get( search_url, auth=(USER, PASS), json=query, headers=headers, verify=False, # SSL 검증 생략 timeout=10 ) if response.status_code == 200: hits = response.json().get("hits", {}).get("hits", []) # 3. 데이터 정제 (Parsing) # 수많은 메타데이터 중 분석에 꼭 필요한 5가지 핵심 정보만 추출합니다. for hit in hits: source = hit.get("_source", {}) syscheck = source.get("syscheck", {}) fim_events.append({ "time": source.get("@timestamp"), "file": syscheck.get("path"), # 어떤 파일이 바뀌었나? "event": syscheck.get("event"), # 생성, 수정, 삭제 중 무엇인가? # 변경된 파일의 해시값 (MISP 등 TI 연동을 위해 추출) "sha256": syscheck.get("sha256_after") or syscheck.get("sha256"), "id": hit.get("_id") # 중복 처리 방지를 위한 원본 로그 ID }) else: print(f"Error: {response.status_code} - {response.text}") except Exception as e: return {"error": str(e)} # 4. 결과 반환 return { "agent_id": AGENT_ID, "count": len(fim_events), "fim_data": fim_events # 정제된 FIM 리스트를 다음 노드(Filter_Old_FIM)로 전달 } if __name__ == "__main__": print(json.dumps(run()))
    • python code 분석 (Filter_Old_FIM)
      • 알람 피로도(Alert Fatigue) 감소: 공격자가 같은 파일을 수십 번 수정할 때마다 똑같은 알람이 울리면 관제자가 지칩니다. 이 코드는 ID 기반 필터링을 통해 "이 로그는 아까 봤던 거네"라고 판단하여 중복 알람을 컷트합니다.
      • TI 쿼리 비용 최적화: MISP나 VirusTotal 같은 위협 인텔리전스 API는 조회 횟수에 제한이 있는 경우가 많습니다. checked_hashes를 관리함으로써, 이미 '안전'하거나 '위험'하다고 판명 난 파일은 다시 물어보지 않습니다.
      • 흐름 제어 (has_new_hashes): Shuffle 워크플로우에서 "데이터가 있을 때만 다음 노드로 가라"는 조건을 줄 때 아주 유용합니다. 불필요하게 MISP 노드를 호출하지 않아 전체 프로세스가 가벼워집니다.
    • import json # --- 1. 입력 데이터 수신 및 정규화 --- try: # 이전 Wazuh 검색 노드($wazuh_fim_search)에서 가져온 파일 변경 리스트입니다. raw_data = """$wazuh_fim_search.message.fim_data""" # 데이터 형식이 문자열, 리스트, 혹은 null인 경우를 모두 대응하여 안전하게 파싱합니다. if not raw_data or raw_data == "null": FIM_LIST = [] elif isinstance(raw_data, str): try: FIM_LIST = json.loads(raw_data) except: FIM_LIST = [] elif isinstance(raw_data, list): FIM_LIST = raw_data else: FIM_LIST = [] except: FIM_LIST = [] # --- 2. 캐시 로드 (상태 유지의 핵심) --- try: # Shuffle 데이터베이스(Get_Session_Cache)에 저장되어 있던 이전 상태 값을 가져옵니다. raw_cache = """$get_session_cache""" cache_data = json.loads(raw_cache) if raw_cache and raw_cache != "null" else {} val = cache_data.get("value", {}) if isinstance(val, str): val = json.loads(val) # [중요] 이미 처리된 로그 ID와 이미 검사 완료된 파일 해시 리스트를 확보합니다. processed_fim_ids = val.get("processed_fim_ids", []) checked_hashes = val.get("checked_hashes", []) except: processed_fim_ids = [] checked_hashes = [] def run(): new_events = [] # 이번에 새로 탐지된 이벤트들 new_ids = [] # 이번에 새로 추가될 로그 ID들 new_hashes_to_scan = [] # MISP 등 TI 조회가 필요한 새로운 해시들 # 수집된 FIM 리스트를 하나씩 검사합니다. for item in FIM_LIST: event_id = item.get("id") file_hash = item.get("sha256") # A. 로그 ID 중복 체크: 이미 처리했던 로그(ID 기반)라면 무시합니다. if event_id not in processed_fim_ids: new_ids.append(event_id) # B. 해시 중복 체크: 로그는 다르더라도 파일 해시가 이미 검사된 적 있다면 # 중복 조회를 방지하기 위해 TI 조회 리스트(new_hashes_to_scan)에서는 제외합니다. if file_hash and file_hash not in checked_hashes and file_hash not in new_hashes_to_scan: new_hashes_to_scan.append(file_hash) new_events.append(item) # --- 3. 결과 반환 (Flag 기반 자동화 제어) --- return { # [핵심 플래그] 새로 발견된 해시가 있을 때만 True를 반환합니다. # 이를 통해 다음 노드인 MISP 조회를 실행할지 말지 결정(Condition)할 수 있습니다. "has_new_hashes": len(new_hashes_to_scan) > 0, "new_fim_count": len(new_events), "hashes_to_misp": new_hashes_to_scan, # MISP 노드가 가져갈 순수 조사 대상 # 이번에 확인한 ID와 해시를 기존 리스트에 합쳐서 반환합니다. # 이 값은 나중에 'Set_Session_Cache' 노드에서 다시 저장됩니다. "updated_fim_ids": processed_fim_ids + new_ids, "updated_hashes": list(set(checked_hashes + new_hashes_to_scan)) } if __name__ == "__main__": print(json.dumps(run()))
  • 5-2. AbuseIPDB (External TI)
    • 역할: 파일 무결성 모니터링(FIM) 데이터를 통해 실제로 시스템에 어떤 파일 변화가 있었는지 추적합니다.
    • Wazuh_FIM_Search 코드 분석:
      • Wazuh Indexer(9200 포트)에 직접 쿼리를 날려 최근 10분(now-10m) 동안의 syscheck 로그만 가져옵니다. 불필요한 과거 데이터를 배제하여 분석 속도를 높였습니다.
    • Filter_Old_FIM 코드 분석:
      • 파일의 해시값(sha256)이 바뀌었는지, 혹은 이전에 이미 분석했던 파일인지 체크합니다.
      • has_new_hashes 플래그를 통해 MISP 등으로 해시를 전달할지 결정합니다.
  • 5-3. Wazuh_SCA_Check
    • 역할: 서버의 보안 설정 점수(SCA)를 확인하여 **"이 서버가 뚫리기 쉬운 상태인가?"**를 판단합니다.
    • 코드 분석:
      • risk_score = 100 - score: SCA 점수는 100점에 가까울수록 안전하다는 뜻입니다. 이를 반전시켜 **'위험도 점수'**로 변환했습니다(예: SCA 37점 -> 위험도 63점).
    • 설계 의도: 설정이 취약한 서버에서 발생한 경고는 더 높은 우선순위로 처리해야 한다는 논리가 담겨 있습니다.
    • python code 분석왜 SCA 점수를 가져오나요? 보안 관제에서 가장 무서운 상황은 **"취약한 설정이 방치된 서버에 해당 취약점을 노린 공격이 들어오는 것"**입니다. • A 서버: SCA 95점 (매우 안전) + Brute-force 공격 감지 $\rightarrow$ 위험도 낮음 (비밀번호 정책이 강력함) • B 서버: SCA 20점 (매우 취약) + Brute-force 공격 감지 $\rightarrow$ 위험도 매우 높음 (쉽게 뚫릴 가능성 큼) 이 코드는 위와 같이 **'공격의 맥락(Context)'**을 시스템 설정 데이터로 입증하기 위해 존재합니다. 점수 반전 로직 (100 - score)의 의미 Wazuh의 SCA 점수는 '준수율(Compliance Rate)'이기 때문에 높을수록 좋지만, 우리가 만드는 Final_Scoring 노드는 위험 요소를 합산하는 방식입니다. 따라서 보안이 취약할수록(SCA 점수가 낮을수록) 최종 위협 점수에 더 많은 가중치를 더하기 위해 수치를 반전시킨 것은 아주 영리한 설계입니다.
    • import requests import json import urllib3 # [1] 보안 경고 무시: 사설 인증서를 사용하는 Wazuh API 환경에서 연결 에러를 방지합니다. urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) # [2] 설정: Wazuh Manager의 RESTful API(55000 포트) 정보를 설정합니다. WAZUH_API_URL = '<https://192.168.10.251:55000>' USER = "wazuh" PASS = "otxbRvH.zpnpqUzyslehdMg6hC*CF9EC" # API 접근용 계정 정보 # [입력] Shuffle에서 전달받은 분석 대상 서버의 Agent ID try: AGENT_ID = str("""$exec.agent.id""") except: AGENT_ID = "000" def get_token(): """ Wazuh API 접근을 위한 Bearer 토큰을 발급받는 함수 """ auth_url = f"{WAZUH_API_URL}/security/user/authenticate" try: response = requests.get(auth_url, auth=(USER, PASS), verify=False, timeout=10) if response.status_code == 200: return response.json().get("data", {}).get("token") return None except: return None def run(): # 1. 토큰 획득 token = get_token() if not token: return {"error": "Authentication Failed"} # 2. SCA(보안 구성 평가) 점수 조회 엔드포인트 호출 sca_url = f"{WAZUH_API_URL}/sca/{AGENT_ID}" headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json" } try: response = requests.get(sca_url, headers=headers, verify=False, timeout=10) if response.status_code == 200: resp_json = response.json() data = resp_json.get("data", {}) # [3] 데이터 정제: 여러 보안 정책(items) 중 가장 핵심적인 첫 번째 정책 점수를 가져옵니다. score = 0 items = data.get("affected_items", []) if items and isinstance(items, list): # 예: CIS Benchmark 기준 점수 (100점 만점) score = items[0].get("score", 0) # [4] 점수 반전 (핵심 로직): # SCA 점수는 100점일수록 안전하고 0점일수록 위험합니다. # 하지만 SOAR의 스코어링 엔진은 '위험할수록 높은 점수'를 선호하므로 # (100 - 보안점수)를 통해 '위험도 점수'로 변환합니다. # 예: 보안 점수가 37점이면, 위험도는 63점이 됩니다. risk_score = 100 - score if score > 0 else 0 return { "agent_id": AGENT_ID, "sca_score": score, # 원본 보안 점수 (예: 37) "sca_risk": risk_score, # 변환된 위험 점수 (예: 63) "is_vulnerable": score < 50 # 50점 미만이면 보안 취약 서버로 분류 } else: return {"error": f"API Error {response.status_code}", "raw": response.text} except Exception as e: return {"error": str(e)} if __name__ == "__main__": print(json.dumps(run()))
  • 6. Final_Scoring (The Brain)
    • 역할: 모든 노드의 결과를 취합하여 최종 위협 등급을 결정합니다.
    • 가중치 로직 분석:
      1. Falco Rules: 랜섬웨어 시그니처(+60), RCE(+50) 등 행위 기반 가중치를 가장 크게 둡니다.
      2. FIM Analysis: 해시가 없는 파일 변경(설정 변경 등)은 더 의심스럽게 보아 +10점을 부여합니다.
      3. AbuseIPDB: TI 점수를 그대로 합산합니다.
      4. SCA Context: sca_risk * 0.1을 통해 시스템 취약도를 최종 점수에 가중치로 반영합니다.
    • 결과: 100점 이상은 CRITICAL, 70점 이상은 HIGH 등으로 등급을 나눕니다.
    • python code 분석
      • 가중치 기반의 합리적 의사결정: 단순히 if 문으로만 판단하면 복합적인 위협을 놓치기 쉽습니다. 하지만 각 요소에 점수를 부여하고 이를 합산하는 방식은, 개별적으로는 LOW 등급인 징후들이 모였을 때 자동으로 CRITICAL로 격상시켜 줍니다.
      • SCA Context 반영의 독창성: 보통의 SIEM은 공격 행위만 보지만, 이 코드는 **"피해 대상 서버의 방어력(SCA)"**을 점수에 반영합니다. 방어력이 낮은 서버에 가해지는 공격에 더 큰 가산점을 주는 방식은 매우 효율적인 관제 전략입니다.
      • 근거 중심의 리포팅 (evidence): 단순히 "100점입니다"라고 하지 않고, evidence.append를 통해 왜 100점이 나왔는지(랜섬웨어 탐지, IP 점수 높음 등)를 문장으로 만들어 Slack 노드에 넘겨줍니다. 이는 관제자가 상황을 즉시 파악하는 데 큰 도움을 줍니다.
    • import json import ast # --- [안전한 파싱 함수] --- # Shuffle의 노드 간 데이터 전달 시 발생할 수 있는 데이터 타입 불일치(문자열 vs 딕셔너리)를 해결합니다. # json.loads와 ast.literal_eval을 병행하여 어떤 형태의 데이터라도 안전하게 파이썬 객체로 변환합니다. def safe_parse(data_str, source_name="Unknown"): if not data_str or data_str == "null" or data_str == "[]": return {} if source_name != "fim_list" else [] try: return json.loads(data_str) except: try: return ast.literal_eval(data_str) except: # 파싱 실패 시 빈 객체를 반환하여 전체 워크플로우가 중단되지 않게 방어합니다. return {} if source_name != "fim_list" else [] # --- 1. 데이터 수집 (Data Ingestion) --- try: # A. Webhook: 원본 경보에서 Falco 룰 이름을 추출합니다. webhook_res = """$exec""" WEBHOOK_DATA = safe_parse(webhook_res, "Webhook") # Falco 데이터 구조가 복잡할 수 있으므로 여러 경로를 통해 Rule 이름을 확보합니다. FALCO_RULE = "Unknown Rule" if "data" in WEBHOOK_DATA and "rule" in WEBHOOK_DATA["data"]: FALCO_RULE = str(WEBHOOK_DATA["data"]["rule"]) elif "rule" in WEBHOOK_DATA and isinstance(WEBHOOK_DATA["rule"], str): FALCO_RULE = WEBHOOK_DATA["rule"] # B. Wazuh FIM: 이전 노드에서 정제된 파일 변경 리스트를 가져옵니다. fim_res = """$wazuh_fim_search""" fim_data_parsed = safe_parse(fim_res, "FIM_Search") fim_list = fim_data_parsed.get("message", {}).get("fim_data", []) if isinstance(fim_data_parsed, dict) else [] # C. AbuseIPDB: IP 평판 조회 결과 리스트를 가져옵니다. abuse_res = """$abuseipdb""" abuse_data = safe_parse(abuse_res, "AbuseIPDB") # D. Wazuh SCA: 시스템 취약 설정 점수를 가져옵니다. sca_res = """$wazuh_sca_check""" sca_data = safe_parse(sca_res, "SCA_Score") except Exception as e: # 예외 발생 시 기본값 세팅 FALCO_RULE = "Unknown"; fim_list = []; abuse_data = []; sca_data = {} # --- 2. 스코어링 로직 (Main Engine) --- def run(): # [캐시 연동] 이전 로그들로부터 쌓여온 누적 점수(base_score)를 시작점으로 잡습니다. # 이 부분이 단순 탐지와 '지속적 위협 추적'을 가르는 핵심입니다. score = $analyze_log_artifacts.message.base_score evidence = [] # [파트 1: Falco 행위 기반 점수] # 공격의 종류(런타임 행위)에 따라 기본 가중치를 부여합니다. if "Interrupt_Ransomware_Behavior" in FALCO_RULE: score += 60 # 랜섬웨어는 가장 위험하므로 높은 점수 evidence.append("[EMERGENCY] Ransomware Encryption Behavior detected.") elif "Interrupt_RCE_Reverse_Shell" in FALCO_RULE: score += 50 evidence.append("[CRITICAL] RCE & Reverse Shell connection established.") # ... (기타 룰별 점수 차등 부여) else: score += 20 evidence.append(f"Behavior: Detected suspicious activity ({FALCO_RULE})") # [파트 2: FIM 파일 변경 기반 점수] fim_score_total = 0 no_hash_count = 0 for item in fim_list: # 해시값이 없는 변경(단순 권한 변경 등)은 추적이 어려우므로 더 위험하게 간주(+10) if not (item.get("sha256") or item.get("md5")): fim_score_total += 10 no_hash_count += 1 else: fim_score_total += 5 # 일반적인 파일 수정은 +5 score += fim_score_total # [파트 3: AbuseIPDB 평판 점수] # 외부 IP의 악성 확신도(Confidence Score)를 점수에 그대로 합산합니다. ip_score_sum = 0 for ip_info in abuse_data: conf_score = ip_info.get("data", {}).get("abuseConfidenceScore", 0) if conf_score > 0: ip_score_sum += conf_score evidence.append(f"[IP] Malicious IP (Score: {conf_score})") score += ip_score_sum # [파트 4: SCA 시스템 취약도 가중치] # 시스템이 취약할수록(sca_risk) 전체 위협 점수에 30%를 가산합니다. # "취약한 환경에서 발생한 공격은 성공 확률이 높다"는 논리를 수식화한 것입니다. real_sca_data = sca_data.get("message", sca_data) if isinstance(sca_data, dict) else {} sca_risk = real_sca_data.get("sca_risk", 0) weighted_risk = int(sca_risk * 0.1) score += weighted_risk # --- 3. 최종 등급 판정 --- level = "INFO" if score >= 100: level = "CRITICAL" # 100점 돌파 시 격리 대상 elif score >= 70: level = "HIGH" elif score >= 40: level = "MEDIUM" elif score >= 20: level = "LOW" return { "final_score": score, "risk_level": level, "title": f"[{level}] Security Alert: {FALCO_RULE}", "description": " / ".join(evidence), # Slack에 뿌려줄 요약 증거 문장들 "details": { "falco_rule": FALCO_RULE, "fim_score": fim_score_total, "ip_reputation_score": ip_score_sum, "sca_risk_score": weighted_risk } } if __name__ == "__main__": print(json.dumps(run()))
  • 7. Set_Session_Cache
    • 역할: 이번 분석 결과를 다시 Shuffle 데이터베이스에 업데이트합니다.
    • 저장 내용:
      • current_score: 다음 로그가 들어왔을 때 이 점수부터 시작합니다.
      • checked_ips, updated_hashes: 다음 분석 시 중복 검사를 방지하기 위해 저장합니다.
  • 8. Format_Slack & Slack
    • 역할: 분석된 복잡한 데이터를 관리자가 보기 편하게 리포트 형식으로 변환하여 전송합니다.
    • 코드 분석:
      • 위험 등급에 따라 Slack 메시지의 색상(color)을 변경(빨강, 주황, 노랑, 초록)하여 시인성을 높였습니다.
      • 단순히 "위험함"이 아니라, 어떤 IP가 문제인지, 어떤 파일이 바뀌었는지를 요약하여 대응 가이드를 함께 제공합니다.
    • python code 분석1) 가독성과 긴급성 중심의 시각화2) 행동 중심의 대응 가이드 (response_guide)3) 정보의 우선순위 배치
      • IP 정보: 공격자의 근원지이므로 최상단 배치.
      • MISP 정보: 알려진 공격(IOC)과의 일치 여부는 공격의 확실성을 입증하는 강력한 근거가 됩니다.
      • FIM 정보: 실제 시스템 내부 피해를 가늠하는 지표입니다.
    • 단순히 "위험합니다"라고 알리는 것은 정보 전달에 불과합니다. 이 코드는 등급에 따라 "네트워크 격리", "방화벽 차단" 같은 구체적인 액션 아이템을 던져줌으로써 관제자의 판단 시간을 획기적으로 줄여줍니다.
    • 보안 사고가 터졌을 때 가장 중요한 것은 **'어디서(Agent)', '무슨 일이(Rule)', '얼마나 위험한가(Score)'**를 3초 안에 파악하는 것입니다. 이 코드는 color 필드를 등급별로 다르게 부여하여, Slack 알림의 왼쪽 띠 색상만 보고도 상황의 긴박함을 인지하게 합니다.
    • import json import ast # --- [안전한 파싱 함수] --- # Shuffle의 여러 노드(Wazuh, AbuseIPDB 등)에서 넘어오는 다양한 데이터 형식을 # 파이썬 딕셔너리나 리스트로 안전하게 변환하여 에러를 방지합니다. def safe_parse(data_str, source_name="Unknown"): if not data_str or data_str == "null" or data_str == "[]": return {} if source_name != "list_type" else [] try: return json.loads(data_str) except: try: return ast.literal_eval(data_str) except: return {} if source_name != "list_type" else [] try: # --- [데이터 수집 파트] --- # 각 분석 노드로부터 최종 결과를 끌어옵니다. # 1. 원본 실행 정보 (어떤 에이전트에서 발생했는지) EXEC_DATA = safe_parse("""$exec""", "Exec") # 2. 최종 점수 및 위험 등급 (Final_Scoring 노드 결과) SCORE_DATA = safe_parse("""$final_scoring""", "Score") if "message" in SCORE_DATA: SCORE_DATA = SCORE_DATA["message"] # 3. 악성 IP 상세 내역 (AbuseIPDB 결과) ABUSE_DATA = safe_parse("""$abuseipdb""", "list_type") # 4. 변경된 파일 목록 (Wazuh FIM 결과) FIM_DATA = safe_parse("""$wazuh_fim_search""", "FIM") fim_list = FIM_DATA.get("message", {}).get("fim_data", []) if isinstance(FIM_DATA, dict) else [] # 5. 위협 인텔리전스 매칭 여부 (MISP 결과) MISP_DATA = safe_parse("""$misp_search_hash""", "MISP") except Exception as e: print(f"Error parsing inputs: {e}") EXEC_DATA, SCORE_DATA, ABUSE_DATA, fim_list, MISP_DATA = {}, {}, [], [], {} def run(): # --- [데이터 가공 및 텍스트 정리] --- # A. 기본 식별 정보 # 에이전트 이름이나 ID를 추출하여 어떤 서버가 공격받고 있는지 명시합니다. agent_id = EXEC_DATA.get("agent", {}).get("name") or EXEC_DATA.get("agent", {}).get("id", "Unknown Server") falco_rule = SCORE_DATA.get("title", "Unknown Alert").replace("[CRITICAL]", "").replace("[HIGH]", "").strip() final_score = SCORE_DATA.get("final_score", 0) risk_level = SCORE_DATA.get("risk_level", "INFO") # B. IP 정보 시각화 # 악성 IP가 발견되면 빨간 동그라미(🔴)와 함께 국가, 점수를 한 줄로 정리합니다. ip_lines = [] for item in ABUSE_DATA: d = item.get("data", item) ip = d.get("ipAddress", "Unknown IP") score = d.get("abuseConfidenceScore", 0) country = d.get("countryCode", "??") if score > 0: ip_lines.append(f"🔴 {ip} ({country}) - Score: {score}") ip_text = "\\n".join(ip_lines) if ip_lines else "✅ 탐지된 악성 IP 없음" # C. 파일 변경(FIM) 정보 요약 # 바뀐 파일이 너무 많으면 Slack 메시지가 길어지므로 최대 5개까지만 보여줍니다. fim_lines = [] for item in fim_list: path = item.get("path", "unknown path") mode = "No Hash" if not item.get("sha256") else "Hash Changed" fim_lines.append(f"📄 {path} ({mode})") if len(fim_lines) >= 5: fim_lines.append(f"...외 {len(fim_list)-5}건 더 있음") break fim_text = "\\n".join(fim_lines) if fim_lines else "✅ 파일 변경 사항 없음" # D. MISP 연동 결과 # 알려진 공격 지표(IOC)와 매칭되었을 때 강력한 경고 문구를 삽입합니다. misp_text = "✅ 매칭된 위협 정보 없음" if MISP_DATA and (MISP_DATA.get("response") or MISP_DATA.get("Event")): misp_text = "🔥 [DANGER] MISP 위협 인텔리전스 매칭됨! (Known IOC)" # E. 대응 가이드 (Response Logic) # 위험 등급(CRITICAL, HIGH 등)에 따라 관리자가 해야 할 행동 지침과 메시지 색상을 결정합니다. if risk_level == "CRITICAL": response = "🚨 [긴급] 즉시 해당 서버 네트워크 격리 및 프로세스 킬 수행 필요." color = "#ff0000" # 강렬한 빨강 elif risk_level == "HIGH": response = "⚠️ [주의] 방화벽 차단 정책 적용 및 상세 로그 분석 요망." color = "#ff9900" # 주황 elif risk_level == "MEDIUM": response = "⚡ [확인] 의심스러운 활동 모니터링 강화." color = "#f2c744" # 노랑 else: response = "✅ 특이사항 없음. 지속 모니터링." color = "#36a64f" # 초록 # --- [최종 결과 반환] --- # 이 딕셔너리 값들이 그대로 Slack 노드의 Attachment나 Block에 매핑됩니다. return { "agent_id": agent_id, "falco_rule": falco_rule, "final_score": final_score, "risk_level": risk_level, "ip_info": ip_text, "fim_info": fim_text, "misp_info": misp_text, "response_guide": response, "color": color, "title": f"[{risk_level}] {falco_rule}" } if __name__ == "__main__": print(json.dumps(run()))

'카테고리 없음'의 다른글

  • 현재글[토스뱅크 부트캠프] 팀프로젝트 Interrupt 7주차

티스토리툴바