恶意程序自动化分析流程编排

---
name: automation-orchestration
description: >
  Build and manage automated malware analysis pipelines that chain together
  multiple analysis stages. Covers sandbox API integration (CAPE, Cuckoo,
  Any.Run, Joe Sandbox), batch processing, API-driven enrichment (VirusTotal,
  MalwareBazaar, Shodan), result aggregation, alerting (Slack, email, webhooks),
  and queue management. Use when processing multiple samples or building
  repeatable analysis workflows.
---

# Automation & Orchestration

Build automated pipelines that chain analysis stages together — from sample
intake through triage, static/dynamic analysis, enrichment, and report
generation.

## Prerequisites

- **Python 3.10+**: `requests`, `celery`, `redis`, `pyyaml`
- **Optional**: CAPE/Cuckoo sandbox instance, VirusTotal API key
- **Infrastructure**: Redis or RabbitMQ for task queuing (optional)
- **API Keys**: `VT_API_KEY`, `MALWAREBAZAAR_API_KEY`, `OTX_API_KEY`

## Step-by-Step Instructions

### Step 1: Define the Analysis Pipeline

Create a pipeline configuration specifying which stages to run and in what order.

**Example pipeline config (`pipeline.yaml`):**
```yaml
pipeline:
  name: standard-analysis
  stages:
    - name: triage
      skill: initial-triage
      timeout: 120
      on_fail: skip
    - name: static
      skill: static-analysis
      timeout: 300
      on_fail: continue
    - name: dynamic
      skill: dynamic-analysis
      sandbox: cape
      timeout: 600
      on_fail: continue
    - name: enrichment
      sources: [virustotal, malwarebazaar, otx]
      timeout: 60
    - name: yara
      skill: yara-rule-development
      rule_paths: [/opt/yara-rules/]
      timeout: 120
    - name: report
      skill: malware-report-writing
      formats: [json, markdown]

  notifications:
    slack_webhook: "${SLACK_WEBHOOK_URL}"
    email: "soc@example.com"
    alert_on: [high, critical]
```

**Run a pipeline:**
```bash
python3 scripts/pipeline_runner.py --config pipeline.yaml --input-dir /samples/incoming/ --output-dir /results/ --mode batch
```

### Step 2: Integrate with Sandbox APIs

Submit samples to automated sandboxes and retrieve results.

**CAPE Sandbox:**
```bash
# Submit a sample
python3 scripts/pipeline_runner.py --mode submit --sandbox cape \
  --sandbox-url http://cape.local:8000 \
  --input sample.exe \
  --output submission.json

# Retrieve results
python3 scripts/pipeline_runner.py --mode retrieve --sandbox cape \
  --task-id 12345 \
  --output cape_results.json
```

**Any.Run (API):**
```bash
python3 scripts/pipeline_runner.py --mode submit --sandbox anyrun \
  --input sample.exe \
  --output anyrun_submission.json
```

**Joe Sandbox (Cloud API):**
```bash
python3 scripts/pipeline_runner.py --mode submit --sandbox joesandbox \
  --input sample.exe \
  --output joe_results.json
```

**Common sandbox API pattern:**
```python
import requests

def submit_to_cape(filepath: str, cape_url: str) -> dict:
    """Submit sample to CAPE sandbox."""
    with open(filepath, "rb") as f:
        response = requests.post(
            f"{cape_url}/apiv2/tasks/create/file/",
            files={"file": f},
            data={"timeout": 300, "enforce_timeout": True},
        )
    return response.json()

def get_cape_report(task_id: int, cape_url: str) -> dict:
    """Retrieve analysis report from CAPE."""
    response = requests.get(f"{cape_url}/apiv2/tasks/get/report/{task_id}/")
    return response.json()
```

### Step 3: Set Up Batch Processing

Process entire directories of samples automatically.

**Batch analysis:**
```bash
python3 scripts/pipeline_runner.py --config pipeline.yaml \
  --input-dir /samples/batch_2024/ \
  --output-dir /results/batch_2024/ \
  --mode batch \
  --parallel 4 \
  --resume
```

**Batch processing features:**
| Feature | Flag | Description |
|---------|------|-------------|
| Parallelism | `--parallel N` | Process N samples concurrently |
| Resume | `--resume` | Skip already-analyzed samples |
| Filter | `--filter "*.exe,*.dll"` | Only process matching files |
| Priority | `--priority high` | Process high-priority samples first |
| Dedup | `--dedup` | Skip duplicate hashes |

### Step 4: API-Driven Enrichment

Enrich analysis results with external threat intelligence.

**VirusTotal enrichment:**
```bash
python3 scripts/pipeline_runner.py --mode enrich --source virustotal \
  --hash sha256:abc123... \
  --output vt_enrichment.json
```

**Multi-source enrichment:**
```python
import os
import requests

def enrich_hash(sha256: str) -> dict:
    """Enrich a hash across multiple threat intel sources."""
    results = {}

    # VirusTotal
    vt_key = os.environ.get("VT_API_KEY")
    if vt_key:
        resp = requests.get(
            f"https://www.virustotal.com/api/v3/files/{sha256}",
            headers={"x-apikey": vt_key},
        )
        if resp.status_code == 200:
            data = resp.json()["data"]["attributes"]
            results["virustotal"] = {
                "detections": f"{data['last_analysis_stats']['malicious']}/{sum(data['last_analysis_stats'].values())}",
                "family": data.get("popular_threat_classification", {}).get("suggested_threat_label"),
            }

    # MalwareBazaar
    resp = requests.post(
        "https://mb-api.abuse.ch/api/v1/",
        data={"query": "get_info", "hash": sha256},
    )
    if resp.status_code == 200 and resp.json().get("query_status") == "ok":
        results["malwarebazaar"] = resp.json()["data"][0]

    return results
```

### Step 5: Aggregate and Correlate Results

Combine results from all stages into a unified analysis.

**Aggregate results:**
```bash
python3 scripts/pipeline_runner.py --mode aggregate \
  --results-dir /results/sample_abc123/ \
  --output unified_report.json
```

**Aggregation combines:**
- Triage results (file type, hashes, initial assessment)
- Static analysis findings (imports, strings, entropy)
- Dynamic analysis behaviors (API calls, file/registry changes)
- Sandbox reports (behavioral score, screenshots)
- Enrichment data (VT detections, known family, tags)
- YARA matches
- IOC extraction results

**Correlation logic:**
```python
def correlate_findings(results: dict) -> dict:
    """Cross-reference findings across analysis stages."""
    correlations = []

    # If static analysis found crypto imports AND dynamic shows file encryption
    static_imports = results.get("static", {}).get("imports", [])
    dynamic_behaviors = results.get("dynamic", {}).get("behaviors", [])

    crypto_imports = [i for i in static_imports if "crypt" in i.lower()]
    file_writes = [b for b in dynamic_behaviors if b.get("type") == "file_write"]

    if crypto_imports and len(file_writes) > 10:
        correlations.append({
            "finding": "Likely ransomware behavior",
            "evidence": ["Crypto API imports", f"{len(file_writes)} file modifications"],
            "confidence": "high",
        })

    return {"correlations": correlations}
```

### Step 6: Set Up Notifications and Alerting

Alert analysts when high-priority samples are detected.

**Slack notification:**
```python
import requests

def notify_slack(webhook_url: str, sample: dict, verdict: str):
    """Send analysis alert to Slack channel."""
    color = {"critical": "#FF0000", "high": "#FF8C00", "medium": "#FFD700"}.get(verdict, "#36A64F")
    requests.post(webhook_url, json={
        "attachments": [{
            "color": color,
            "title": f"Malware Analysis Alert: {verdict.upper()}",
            "fields": [
                {"title": "Sample", "value": sample["filename"], "short": True},
                {"title": "SHA-256", "value": sample["sha256"][:16] + "...", "short": True},
                {"title": "Family", "value": sample.get("family", "Unknown"), "short": True},
                {"title": "Score", "value": str(sample.get("score", "N/A")), "short": True},
            ],
        }],
    })
```

**Email notification:**
```bash
python3 scripts/pipeline_runner.py --mode notify \
  --results unified_report.json \
  --notify-email soc@example.com \
  --notify-threshold high
```

### Step 7: Queue Management with Celery

Scale processing with distributed task queues.

**Celery worker setup:**
```python
from celery import Celery

app = Celery("malware_pipeline", broker="redis://localhost:6379/0")

@app.task(bind=True, max_retries=3)
def analyze_sample(self, filepath: str, config: dict) -> dict:
    """Process a single sample through the analysis pipeline."""
    try:
        results = {}
        for stage in config["stages"]:
            results[stage["name"]] = run_stage(stage, filepath)
        return results
    except Exception as exc:
        self.retry(exc=exc, countdown=60)
```

**Start workers:**
```bash
# Start 4 analysis workers
celery -A pipeline_runner worker --concurrency=4 -Q analysis

# Start enrichment worker (rate-limited)
celery -A pipeline_runner worker --concurrency=1 -Q enrichment

# Monitor queue status
celery -A pipeline_runner inspect active
```

### Step 8: Schedule and Monitor Pipelines

Set up recurring analysis jobs and monitor pipeline health.

**Cron-based scheduling:**
```bash
# Process incoming samples every 15 minutes
*/15 * * * * python3 /opt/malware-pipeline/scripts/pipeline_runner.py --config /opt/malware-pipeline/pipeline.yaml --input-dir /samples/incoming/ --output-dir /results/ --mode batch --resume 2>&1 >> /var/log/malware-pipeline.log
```

**Pipeline health monitoring:**
```bash
python3 scripts/pipeline_runner.py --mode status --output pipeline_status.json
```

**Status dashboard metrics:**
| Metric | Description |
|--------|-------------|
| Samples processed (24h) | Total samples through pipeline |
| Average processing time | Time per sample across all stages |
| Failure rate | Percentage of failed analyses |
| Queue depth | Pending samples awaiting analysis |
| API quota remaining | Remaining calls for VT, etc. |

## Output Format

```json
{
  "pipeline": "standard-analysis",
  "sample": {
    "filename": "suspicious.exe",
    "sha256": "abc123...",
    "size_bytes": 245760
  },
  "stages": {
    "triage": {"status": "completed", "duration_seconds": 15, "verdict": "suspicious"},
    "static": {"status": "completed", "duration_seconds": 45},
    "dynamic": {"status": "completed", "duration_seconds": 300, "sandbox": "cape"},
    "enrichment": {"status": "completed", "sources": ["virustotal", "malwarebazaar"]},
    "yara": {"status": "completed", "matches": 3},
    "report": {"status": "completed", "formats": ["json", "markdown"]}
  },
  "verdict": "malicious",
  "confidence": "high",
  "family": "AgentTesla",
  "score": 92,
  "processing_time_seconds": 425,
  "notifications_sent": ["slack"]
}
```

## Tips

- Start simple — a bash script chaining tools is a valid first pipeline
- Rate-limit API calls to avoid quota exhaustion (VT: 4 req/min on free tier)
- Always deduplicate by hash before processing to avoid wasted resources
- Use `--resume` for batch jobs so you can restart without reprocessing
- Keep sandbox VMs clean — revert snapshots between analyses
- Log everything — pipeline failures are debugging nightmares without logs
- Set reasonable timeouts for each stage to prevent hung analyses
- Monitor disk space — dynamic analysis generates large artifacts
- Use separate queues for CPU-intensive (analysis) vs I/O-intensive (enrichment) tasks