building-automated-malware-submission-pipeline

name	building-automated-malware-submission-pipeline
description	构建自动化恶意软件提交和分析流水线，从终端和邮件网关收集可疑文件， 将其提交至沙箱环境和多引擎扫描器，并生成带有失陷指标（IOC）的研判结论以集成到 SIEM。 适用于 SOC 团队需要将恶意软件分析扩展到高容量告警分诊的场景，超越手动沙箱提交的限制。
domain	cybersecurity
subdomain	soc-operations
tags	["soc","malware-analysis","sandbox","automation","virustotal","cuckoo","any-run","pipeline"]
version	1.0
author	mahipal
license	Apache-2.0

构建自动化恶意软件提交流水线

适用场景

以下情况使用本技能：

• SOC 团队面临大量可疑文件告警，需要沙箱分析
• 手动沙箱提交在告警分诊工作流中造成瓶颈
• 终端和邮件安全工具隔离了需要自动研判的文件
• 事件响应（Incident Response）需要快速识别恶意软件家族并提取失陷指标（IOC）

不适用于在生产环境中分析实时恶意软件样本——请始终使用隔离的沙箱基础设施。

前置条件

• 沙箱环境：Cuckoo Sandbox、Joe Sandbox、Any.Run 或 VMRay
• VirusTotal API 密钥（企业版用于提交，免费版用于查询）
• 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, : (filepath), : }

     ():
        
         email
         email  policy

         eml_file  Path(smtp_quarantine_path).glob():
            msg = email.message_from_binary_file(
                eml_file.(), policy=policy.default
            )
             attachment  msg.iter_attachments():
                content = attachment.get_content()
                 (content, ):
                    content = content.encode()
                sha256 = hashlib.sha256(content).hexdigest()
                filename = attachment.get_filename()  

                filepath = .quarantine_dir / 
                filepath.write_bytes(content)
                 {
                    : sha256,
                    : (filepath),
                    : ,
                    : filename,
                    : msg[],
                    : msg[]
                }

     ():
        
         (filepath, )  f:
            content = f.read()
         {
            : hashlib.md5(content).hexdigest(),
            : hashlib.sha1(content).hexdigest(),
            : hashlib.sha256(content).hexdigest(),
            : (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 ():
        
        response = requests.post(
            .mb_api_url,
            data={: , : sha256}
        )
        data = response.json()
         data[] == :
            entry = data[][]
             {
                : ,
                : entry.get(, ),
                : entry.get(, []),
                : entry.get(, ),
                : entry.get(, )
            }
         {: }

     ():
        
        vt_result = .check_virustotal(sha256)
        mb_result = .check_malwarebazaar(sha256)

        verdict = 
         vt_result[]  vt_result.get(, ) > :
            verdict = 
         vt_result[]  vt_result.get(, ) == :
            verdict = 

         {
            : sha256,
            : vt_result,
            : mb_result,
            : verdict,
            : verdict == 
        }

     ():
        .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":
                 .get_report(task_id)
             status == :
                 {: }
            time.sleep(poll_interval)
            elapsed += poll_interval
         {: }

     ():
        
        response = requests.get()
        report = response.json()

        
         {
            : task_id,
            : report.get(, {}).get(, ),
            : [
                {: s[], : s[], : s[]}
                 s  report.get(, [])
            ],
            : {
                : [d[]  d  report.get(, {}).get(, [])],
                : [
                    {: h[], : h[]}
                     h  report.get(, {}).get(, [])
                ],
                : report.get(, {}).get(, [])
            },
            : [
                {: f[], : f[], : f[]}
                 f  report.get(, [])
            ],
            : [
                {: p[], : p[], : p.get(, )}
                 p  report.get(, {}).get(, [])
            ],
            : [
                k  k  report.get(, {}).get(, {}).get(, [])
            ]
        }

     ():
        
         (filepath, )  f:
            response = requests.post(
                ,
                headers={: },
                files={: f},
                data={
                    : ,
                    : ,
                    : 
                }
            )
         response.json()[][]

步骤 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(, )
        sandbox_score = sandbox_report.get(, )  sandbox_report  
        sig_count = (sandbox_report.get(, []))  sandbox_report  

        combined_score = (vt_malicious * ) + (sandbox_score * ) + (sig_count * )

         combined_score >= :
            verdict = 
            confidence = 
         combined_score >= :
            verdict = 
            confidence = 
         combined_score >= :
            verdict = 
            confidence = 
        :
            verdict = 
            confidence = 

         {
            : verdict,
            : confidence,
            : combined_score,
            : iocs,
            : vt_malicious,
            : sandbox_score,
            : sandbox_report.get(, [])  sandbox_report  []
        }

步骤 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=False
    )
    return response.status_code == 200

def push_iocs_to_blocklist():
    
     ip  iocs.get(, []):
        requests.post(
            ,
            json={: , : ip, : , : }
        )
     domain  iocs.get(, []):
        requests.post(
            ,
            json={: , : domain, : , : }
        )

步骤 6：编排完整流水线

def run_malware_pipeline(sample_path, config):
    """执行完整的恶意软件分析流水线"""
    collector = MalwareCollector()
    screener = MalwarePreScreener(config["vt_key"])
    submitter = SandboxSubmitter(config["cuckoo_url"])
    generator = VerdictGenerator()

    # 第一步：哈希计算和预筛选
    hashes = collector.compute_hashes(sample_path)
    pre_screen = screener.pre_screen(hashes["sha256"])

    # 第二步：如未知则提交沙箱
    sandbox_report = None
    if pre_screen["needs_sandbox"]:
        task_id = submitter.submit_to_cuckoo(sample_path)
        sandbox_report = submitter.wait_for_analysis(task_id)

    # 第三步：生成研判结论
    verdict = generator.generate_verdict(pre_screen, sandbox_report)
    verdict["sha256"] = hashes["sha256"]
    verdict["threat_label"] = pre_screen.get("virustotal", {}).get("threat_label", "Unknown")

    # 第四步：推送至 SIEM
    push_to_splunk(verdict, config["splunk_url"], config["splunk_token"])

    # 第五步：如为恶意则封锁
    if verdict["verdict"] == "MALICIOUS":
        push_iocs_to_blocklist(verdict["iocs"], config["firewall_api"])

    screener.close()
    return verdict

核心概念

术语	定义
动态分析（Dynamic Analysis）	在沙箱中执行恶意软件以观察运行时行为（进程创建、网络、文件系统变更）
静态分析（Static Analysis）	不执行恶意软件的检查（哈希查询、字符串分析、PE 头检查）
沙箱规避（Sandbox Evasion）	恶意软件用于检测沙箱环境并改变行为以规避分析的技术
IOC 提取（IOC Extraction）	从沙箱报告自动识别网络指标、文件取证痕迹和注册表变更的过程
多 AV 扫描（Multi-AV Scanning）	将样本提交至多个杀毒引擎（VirusTotal）以进行基于共识的检测
研判结论（Verdict）	样本的最终分类：Malicious（恶意）、Suspicious（可疑）、Potentially Unwanted（可能不需要）或 Clean（干净）

工具与系统

• Cuckoo Sandbox：开源自动化恶意软件分析平台，具备行为分析和网络捕获功能
• Joe Sandbox：商业沙箱，具有深度行为分析、YARA 匹配和 MITRE ATT&CK 映射
• Any.Run：交互式沙箱服务，允许在分析过程中实时操作，用于调试规避型恶意软件
• VirusTotal：多引擎扫描服务，提供 70+ 杀毒引擎结果和行为分析报告
• CAPE Sandbox：社区维护的 Cuckoo 分支，增强了载荷提取和配置转储功能

常见场景

• 邮件附件分诊：自动提交隔离的邮件附件，在 5 分钟内生成研判结论
• EDR 隔离文件处理：批量处理终端安全隔离的文件以进行详细分析
• 事件调查：提交 IR 过程中发现的可疑二进制文件以识别恶意软件家族并提取 IOC
• 威胁情报富化：分析来自威胁情报订阅的样本以提取 C2 基础设施并更新封锁
• 零日检测：沙箱通过行为分析捕获基于签名的 AV 遗漏的新型恶意软件

输出格式

MALWARE ANALYSIS REPORT — Pipeline Submission
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Sample:       invoice_march.docx
SHA256:       a1b2c3d4e5f6a7b8...
File Type:    Microsoft Word Document (macro-enabled)

Pre-Screening:
  VirusTotal:    34/72 malicious (Emotet.Downloader)
  MalwareBazaar: Tags: emotet, macro, downloader

Sandbox Analysis (Cuckoo):
  Score:         9.2/10 (MALICIOUS)
  Signatures:
    - Macro executes PowerShell download cradle (severity: 8)
    - Process injection into explorer.exe (severity: 9)
    - Connects to known Emotet C2 server (severity: 9)

Extracted IOCs:
  C2 IPs:       185.234.218[.]50:8080, 45.77.123[.]45:443
  Domains:       update-service[.]evil[.]com
  Dropped Files: payload.dll (SHA256: b2c3d4e5...)
  Registry:      HKCU\Software\Microsoft\Windows\CurrentVersion\Run\Update

VERDICT: MALICIOUS (Emotet Downloader) — Confidence: HIGH
ACTIONS:
  [DONE] IOCs pushed to Splunk threat intel
  [DONE] C2 IPs blocked on firewall
  [DONE] Domain sinkholed on DNS
  [DONE] Hash blocked on endpoint