자동화된 악성코드 제출 파이프라인 구축

자동화된 악성코드 제출 파이프라인 구축은(는) 무엇을 하나요?

사용 시기

이 스킬은 다음 상황에서 사용하세요:

• SOC 팀이 샌드박스 분석이 필요한 대량의 의심스러운 파일 알림을 처리할 때
• 수동 샌드박스 제출이 알림 분류 워크플로에서 병목 현상을 일으킬 때
• 엔드포인트 및 이메일 보안 도구가 자동화된 평결 결정이 필요한 파일을 격리할 때
• 사고 대응 시 신속한 악성코드 패밀리 식별 및 IOC 추출이 필요할 때

다음 경우에는 사용하지 마십시오: 프로덕션 환경에서 라이브 악성코드 샘플을 분석할 때 — 항상 격리된 샌드박스 인프라를 사용하세요.

사전 요구 사항

• 샌드박스 환경: Cuckoo Sandbox, Joe Sandbox, Any.Run 또는 VMRay
• VirusTotal API 키 (제출용 Enterprise, 조회용 무료)
• 알려진 악성코드 조회를 위한 MalwareBazaar API 액세스
• 파일 수집 메커니즘: EDR 격리 API, 이메일 게이트웨이 내보내기, 네트워크 캡처
• Python 3.8+ 및 requests, vt-py, pefile 라이브러리
• 프로덕션 연결이 없는 격리된 분석 네트워크

워크플로

1단계: 파일 수집 파이프라인 구축

여러 소스에서 의심스러운 파일 수집:

import requests
import hashlib
import os
from pathlib import Path
from datetime import datetime

class MalwareCollector:
    def __init__(self, quarantine_dir="/opt/malware_quarantine"):
        self.quarantine_dir = Path(quarantine_dir)
        self.quarantine_dir.mkdir(exist_ok=True)

    def collect_from_edr(self, edr_api_url, api_token):
        """CrowdStrike Falcon에서 격리된 파일 가져오기"""
        headers = {"Authorization": f"Bearer {api_token}"}

        # 최근 격리 이벤트 가져오기
        response = requests.get(
            f"{edr_api_url}/quarantine/queries/quarantined-files/v1",
            headers=headers,
            params={"filter": "state:'quarantined'", "limit": 50}
        )
        file_ids = response.json()["resources"]

        for file_id in file_ids:
            # 격리된 파일 다운로드
            dl_response = requests.get(
                f"{edr_api_url}/quarantine/entities/quarantined-files/v1",
                headers=headers,
                params={"ids": file_id}
            )
            file_data = dl_response.content
            sha256 = hashlib.sha256(file_data).hexdigest()

            filepath = self.quarantine_dir / f"{sha256}.sample"
            filepath.write_bytes(file_data)
            yield {"sha256": sha256, "path": str(filepath), "source": "edr"}

    def collect_from_email_gateway(self, smtp_quarantine_path):
        """이메일 게이트웨이 격리에서 첨부 파일 가져오기"""
        import email
        from email import policy

        for eml_file in Path(smtp_quarantine_path).glob("*.eml"):
            msg = email.message_from_binary_file(
                eml_file.open("rb"), policy=policy.default
            )
            for attachment in msg.iter_attachments():
                content = attachment.get_content()
                if isinstance(content, str):
                    content = content.encode()
                sha256 = hashlib.sha256(content).hexdigest()
                filename = attachment.get_filename() or "unknown"

                filepath = self.quarantine_dir / f"{sha256}.sample"
                filepath.write_bytes(content)
                yield {
                    "sha256": sha256,
                    "path": str(filepath),
                    "source": "email",
                    "original_filename": filename,
                    "sender": msg["From"],
                    "subject": msg["Subject"]
                }

    def compute_hashes(self, filepath):
        """파일의 MD5, SHA1, SHA256 계산"""
        with open(filepath, "rb") as f:
            content = f.read()
        return {
            "md5": hashlib.md5(content).hexdigest(),
            "sha1": hashlib.sha1(content).hexdigest(),
            "sha256": hashlib.sha256(content).hexdigest(),
            "size": len(content)
        }

2단계: 해시 조회로 사전 선별

샌드박스 제출 전에 파일이 이미 알려진 파일인지 확인:

import vt

class MalwarePreScreener:
    def __init__(self, vt_api_key, mb_api_url="https://mb-api.abuse.ch/api/v1/"):
        self.vt_client = vt.Client(vt_api_key)
        self.mb_api_url = mb_api_url

    def check_virustotal(self, sha256):
        """VirusTotal에서 해시 조회"""
        try:
            file_obj = self.vt_client.get_object(f"/files/{sha256}")
            stats = file_obj.last_analysis_stats
            return {
                "found": True,
                "malicious": stats.get("malicious", 0),
                "suspicious": stats.get("suspicious", 0),
                "undetected": stats.get("undetected", 0),
                "total": sum(stats.values()),
                "threat_label": getattr(file_obj, "popular_threat_classification", {}).get(
                    "suggested_threat_label", "Unknown"
                ),
                "type": getattr(file_obj, "type_description", "Unknown")
            }
        except vt.APIError:
            return {"found": False}

    def check_malwarebazaar(self, sha256):
        """MalwareBazaar에서 해시 조회"""
        response = requests.post(
            self.mb_api_url,
            data={"query": "get_info", "hash": sha256}
        )
        data = response.json()
        if data["query_status"] == "ok":
            entry = data["data"][0]
            return {
                "found": True,
                "signature": entry.get("signature", "Unknown"),
                "tags": entry.get("tags", []),
                "file_type": entry.get("file_type", "Unknown"),
                "first_seen": entry.get("first_seen", "Unknown")
            }
        return {"found": False}

    def pre_screen(self, sha256):
        """모든 사전 선별 검사 실행"""
        vt_result = self.check_virustotal(sha256)
        mb_result = self.check_malwarebazaar(sha256)

        verdict = "UNKNOWN"
        if vt_result["found"] and vt_result.get("malicious", 0) > 10:
            verdict = "KNOWN_MALICIOUS"
        elif vt_result["found"] and vt_result.get("malicious", 0) == 0:
            verdict = "LIKELY_CLEAN"

        return {
            "sha256": sha256,
            "virustotal": vt_result,
            "malwarebazaar": mb_result,
            "pre_screen_verdict": verdict,
            "needs_sandbox": verdict == "UNKNOWN"
        }

    def close(self):
        self.vt_client.close()

3단계: 동적 분석을 위해 샌드박스에 제출

Cuckoo Sandbox 제출:

class SandboxSubmitter:
    def __init__(self, cuckoo_url="http://cuckoo.internal:8090"):
        self.cuckoo_url = cuckoo_url

    def submit_to_cuckoo(self, filepath, timeout=300):
        """Cuckoo Sandbox에 파일 제출"""
        with open(filepath, "rb") as f:
            response = requests.post(
                f"{self.cuckoo_url}/tasks/create/file",
                files={"file": f},
                data={
                    "timeout": timeout,
                    "options": "procmemdump=yes,route=none",
                    "priority": 2,
                    "machine": "win10_x64"
                }
            )
        task_id = response.json()["task_id"]
        return task_id

    def wait_for_analysis(self, task_id, poll_interval=30, max_wait=600):
        """샌드박스 분석 완료 대기"""
        import time
        elapsed = 0
        while elapsed < max_wait:
            response = requests.get(f"{self.cuckoo_url}/tasks/view/{task_id}")
            status = response.json()["task"]["status"]
            if status == "reported":
                return self.get_report(task_id)
            elif status == "failed_analysis":
                return {"error": "분석 실패"}
            time.sleep(poll_interval)
            elapsed += poll_interval
        return {"error": "분석 시간 초과"}

    def get_report(self, task_id):
        """분석 보고서 검색"""
        response = requests.get(f"{self.cuckoo_url}/tasks/report/{task_id}")
        report = response.json()

        # 주요 지표 추출
        return {
            "task_id": task_id,
            "score": report.get("info", {}).get("score", 0),
            "signatures": [
                {"name": s["name"], "severity": s["severity"], "description": s["description"]}
                for s in report.get("signatures", [])
            ],
            "network": {
                "dns": [d["request"] for d in report.get("network", {}).get("dns", [])],
                "http": [
                    {"url": h["uri"], "method": h["method"]}
                    for h in report.get("network", {}).get("http", [])
                ],
                "hosts": report.get("network", {}).get("hosts", [])
            },
            "dropped_files": [
                {"name": f["name"], "sha256": f["sha256"], "size": f["size"]}
                for f in report.get("dropped", [])
            ],
            "processes": [
                {"name": p["process_name"], "pid": p["pid"], "command_line": p.get("command_line", "")}
                for p in report.get("behavior", {}).get("processes", [])
            ],
            "registry_keys": [
                k for k in report.get("behavior", {}).get("summary", {}).get("regkey_written", [])
            ]
        }

    def submit_to_joesandbox(self, filepath, joe_api_key, joe_url="https://jbxcloud.joesecurity.org/api"):
        """Joe Sandbox Cloud에 제출"""
        with open(filepath, "rb") as f:
            response = requests.post(
                f"{joe_url}/v2/submission/new",
                headers={"Authorization": f"Bearer {joe_api_key}"},
                files={"sample": f},
                data={
                    "systems": "w10_64",
                    "internet-access": False,
                    "report-cache": True
                }
            )
        return response.json()["data"]["webid"]

4단계: IOC 추출 및 평결 생성

class VerdictGenerator:
    def __init__(self):
        self.malicious_threshold = 7  # Cuckoo 점수 임계값

    def generate_verdict(self, pre_screen, sandbox_report):
        """사전 선별 및 샌드박스 결과를 결합하여 최종 평결 생성"""
        iocs = {
            "ips": [],
            "domains": [],
            "urls": [],
            "hashes": [],
            "registry_keys": [],
            "files_dropped": []
        }

        # 샌드박스 보고서에서 IOC 추출
        if sandbox_report:
            iocs["domains"] = sandbox_report.get("network", {}).get("dns", [])
            iocs["ips"] = sandbox_report.get("network", {}).get("hosts", [])
            iocs["urls"] = [
                h["url"] for h in sandbox_report.get("network", {}).get("http", [])
            ]
            iocs["hashes"] = [
                f["sha256"] for f in sandbox_report.get("dropped_files", [])
            ]
            iocs["registry_keys"] = sandbox_report.get("registry_keys", [])[:10]
            iocs["files_dropped"] = sandbox_report.get("dropped_files", [])

        # 평결 결정
        vt_malicious = pre_screen.get("virustotal", {}).get("malicious", 0)
        sandbox_score = sandbox_report.get("score", 0) if sandbox_report else 0
        sig_count = len(sandbox_report.get("signatures", [])) if sandbox_report else 0

        combined_score = (vt_malicious * 2) + (sandbox_score * 10) + (sig_count * 5)

        if combined_score >= 100:
            verdict = "MALICIOUS"
            confidence = "HIGH"
        elif combined_score >= 50:
            verdict = "SUSPICIOUS"
            confidence = "MEDIUM"
        elif combined_score >= 20:
            verdict = "POTENTIALLY_UNWANTED"
            confidence = "LOW"
        else:
            verdict = "CLEAN"
            confidence = "HIGH"

        return {
            "verdict": verdict,
            "confidence": confidence,
            "combined_score": combined_score,
            "iocs": iocs,
            "vt_detections": vt_malicious,
            "sandbox_score": sandbox_score,
            "signatures": sandbox_report.get("signatures", []) if sandbox_report else []
        }

5단계: 결과를 SIEM에 푸시

def push_to_splunk(verdict_result, splunk_url, splunk_token):
    """악성코드 분석 평결을 Splunk HEC로 전송"""
    import json

    event = {
        "sourcetype": "malware_analysis",
        "source": "malware_pipeline",
        "event": {
            "sha256": verdict_result["sha256"],
            "verdict": verdict_result["verdict"],
            "confidence": verdict_result["confidence"],
            "score": verdict_result["combined_score"],
            "vt_detections": verdict_result["vt_detections"],
            "sandbox_score": verdict_result["sandbox_score"],
            "malware_family": verdict_result.get("threat_label", "Unknown"),
            "iocs": verdict_result["iocs"],
            "signatures": [s["name"] for s in verdict_result["signatures"]]
        }
    }

    response = requests.post(
        f"{splunk_url}/services/collector/event",
        headers={
            "Authorization": f"Splunk {splunk_token}",
            "Content-Type": "application/json"
        },
        json=event,
        verify=not os.environ.get("SKIP_TLS_VERIFY", "").lower() == "true",  # 랩 환경에서 자체 서명된 인증서를 사용하려면 SKIP_TLS_VERIFY=true 설정
    )
    return response.status_code == 200

def push_iocs_to_blocklist(iocs, firewall_api):
    """추출된 IOC를 차단 인프라에 푸시"""
    for ip in iocs.get("ips", []):
        requests.post(
            f"{firewall_api}/block",
            json={"type": "ip", "value": ip, "action": "block", "source": "malware_pipeline"}
        )
    for domain in iocs.get("domains", []):
        requests.post(
            f"{firewall_api}/block",
            json={"type": "domain", "value": domain, "action": "sinkhole", "source": "malware_pipeline"}
        )

6단계: 전체 파이프라인 오케스트레이션

def run_pipeline():
    """전체 악성코드 분석 파이프라인 실행"""
    # 설정
    VT_API_KEY = os.environ.get("VT_API_KEY")
    JOE_API_KEY = os.environ.get("JOE_API_KEY")
    SPLUNK_URL = os.environ.get("SPLUNK_URL")
    SPLUNK_TOKEN = os.environ.get("SPLUNK_TOKEN")
    FIREWALL_API = os.environ.get("FIREWALL_API")
    
    # 수집 및 분석 실행
    collector = MalwareCollector()
    screener = MalwarePreScreener(VT_API_KEY)
    submitter = SandboxSubmitter()
    verdict_gen = VerdictGenerator()
    
    for file_info in collector.collect_from_edr(
        edr_api_url=os.environ.get("EDR_API_URL"),
        api_token=os.environ.get("EDR_API_TOKEN")
    ):
        sha256 = file_info["sha256"]
        filepath = file_info["path"]
        
        # 사전 선별
        pre_screen_result = screener.pre_screen(sha256)
        
        if pre_screen_result["needs_sandbox"]:
            # 샌드박스 제출 및 분석
            task_id = submitter.submit_to_cuckoo(filepath)
            sandbox_report = submitter.wait_for_analysis(task_id)
            
            # 평결 생성
            final_result = verdict_gen.generate_verdict(
                pre_screen_result, sandbox_report
            )
            final_result["sha256"] = sha256
            
            # 결과 푸시
            push_to_splunk(final_result, SPLUNK_URL, SPLUNK_TOKEN)
            
            if final_result["verdict"] == "MALICIOUS":
                push_iocs_to_blocklist(final_result["iocs"], FIREWALL_API)
        else:
            # 알려진 상태 기록
            print(f"파일 {sha256}의 사전 선별 평결: {pre_screen_result['pre_screen_verdict']}")

if __name__ == "__main__":
    run_pipeline()

환경 설정

# 필수 변수
export VT_API_KEY="your_virustotal_api_key"
export EDR_API_URL="https://api.crowdstrike.com"
export EDR_API_TOKEN="your_edr_token"

# 선택 사항 - Joe Sandbox 사용 시
export JOE_API_KEY="your_joesandbox_key"

# 선택 사항 - Splunk 출력 사용 시
export SPLUNK_URL="https://splunkhec.internal:8088"
export SPLUNK_TOKEN="your_splunk_hec_token"

# 선택 사항 - 자동 차단 사용 시
export FIREWALL_API="https://firewall.internal/api"

# 선택 사항 - 랩 환경에서 자체 서명된 인증서 허용
export SKIP_TLS_VERIFY="true"

보안 고려 사항

• 악성코드를 프로덕션 환경에 노출하지 마십시오. 항상 격리된 분석 네트워크를 사용하십시오.
• API 키와 토큰을 안전하게 관리하십시오 (환경 변수, 볼트 사용).
• 수집된 샘플을 로컬에 유지하는 경우 저장 데이터를 암호화하십시오.
• 처리 후 정기적으로 격리 디렉토리를 정리하십시오.
• 분석 네트워크용 방화벽 규칙을 구현하여 아웃바운드 악성코드 통신을 방지하십시오.

문제 해결

문제	해결 방법
Cuckoo API가 연결을 거부함	Cuckoo 웹 서비스가 실행 중이고 방화벽이 포트 8090을 허용하는지 확인
VirusTotal API 시간 초과	요금 제한 확인; 지수 백오프 재시도 구현
샌드박스 분석 실패	Cuckoo 가상 머신 스냅샷이 유효한지 확인; vmcloak으로 재설정
Splunk HEC 401 오류	Splunk HEC 토큰이 활성화되었고 인덱서 확인이 올바른지 확인
MalwareBazaar 응답 없음	MalwareBazaar가 차단 목록에 없는지 확인; API가 변경되었는지 확인

관련 리소스

• Cuckoo Sandbox 문서
• VirusTotal API 문서
• MalwareBazaar API
• Splunk HTTP 이벤트 수집기
• Joe Sandbox Cloud API