#!/usr/bin/env python3 """Download malware samples from MalwareBazaar and other repositories. Supports search by hash, tag, family, and signature. Downloads to quarantine directory with automatic hash verification. """ from __future__ import annotations import argparse import hashlib import json import os import sys import zipfile from datetime import datetime from io import BytesIO from pathlib import Path try: import requests HAS_REQUESTS = True except ImportError: HAS_REQUESTS = False MALWAREBAZAAR_API = "https://mb-api.abuse.ch/api/v1/" def compute_sha256(filepath) -> dict: """Compute SHA256 hash of a file.""" h = hashlib.sha256() with open(filepath, "rb") as f: for chunk in iter(lambda: f.read(8192), b""): h.update(chunk) return h.hexdigest() def search_bazaar(query_type, query_value, limit=10) -> dict: """Search MalwareBazaar for samples.""" if not HAS_REQUESTS: print("[!] requests library required. Install: pip install requests", file=sys.stderr) return None data = {"limit": limit} if query_type == "hash": data["query"] = "get_info" data["hash"] = query_value elif query_type == "tag": data["query"] = "get_taginfo" data["tag"] = query_value elif query_type == "signature": data["query"] = "get_siginfo" data["signature"] = query_value else: print(f"[!] Unknown query type: {query_type}", file=sys.stderr) return None try: response = requests.post(MALWAREBAZAAR_API, data=data, timeout=30) response.raise_for_status() return response.json() except requests.RequestException as e: print(f"[!] API request failed: {e}", file=sys.stderr) return None def download_sample(sha256_hash, output_dir) -> dict: """Download a sample from MalwareBazaar by SHA256 hash.""" if not HAS_REQUESTS: print("[!] requests library required. Install: pip install requests", file=sys.stderr) return None output_path = Path(output_dir) output_path.mkdir(parents=True, exist_ok=True) data = { "query": "get_file", "sha256_hash": sha256_hash, } try: print(f"[*] Downloading {sha256_hash[:16]}...") response = requests.post(MALWAREBAZAAR_API, data=data, timeout=120) if response.status_code != 200: print(f"[!] Download failed: HTTP {response.status_code}", file=sys.stderr) return None # MalwareBazaar returns password-protected ZIP (password: "infected") zip_path = output_path / f"{sha256_hash[:8]}.zip" zip_path.write_bytes(response.content) # Extract with password try: with zipfile.ZipFile(BytesIO(response.content)) as zf: zf.extractall(path=str(output_path), pwd=b"infected") extracted_files = zf.namelist() print(f"[+] Extracted: {', '.join(extracted_files)}") except (zipfile.BadZipFile, RuntimeError) as e: print(f"[!] Extraction failed: {e}. ZIP saved to {zip_path}") return str(zip_path) # Verify hash for fname in extracted_files: fpath = output_path / fname if fpath.exists(): actual_hash = compute_sha256(fpath) if actual_hash.lower() == sha256_hash.lower(): print(f"[+] Hash verified: {actual_hash}") # Rename to safe extension safe_name = f"{sha256_hash[:8]}.sample" safe_path = output_path / safe_name fpath.rename(safe_path) print(f"[+] Saved as: {safe_path}") # Log acquisition log_acquisition(safe_path, sha256_hash, "malwarebazaar") return str(safe_path) else: print(f"[!] Hash mismatch! Expected {sha256_hash[:16]}, got {actual_hash[:16]}") return str(zip_path) except requests.RequestException as e: print(f"[!] Download failed: {e}", file=sys.stderr) return None def log_acquisition(filepath, sha256, source) -> None: """Log sample acquisition for chain of custody.""" log_dir = Path(filepath).parent log_file = log_dir / "acquisition_log.jsonl" entry = { "timestamp": datetime.now().isoformat(), "file": str(filepath), "sha256": sha256, "source": source, "handler": os.environ.get("USER", "unknown"), "hostname": os.uname().nodename if hasattr(os, "uname") else "unknown", } with open(log_file, "a") as f: f.write(json.dumps(entry) + "\n") print(f"[+] Acquisition logged to {log_file}") def main() -> None: parser = argparse.ArgumentParser( description="Download malware samples from MalwareBazaar" ) group = parser.add_mutually_exclusive_group(required=True) group.add_argument("--input", "--hash", help="SHA256 hash to download") group.add_argument("--tag", help="Search by tag (e.g., 'emotet')") group.add_argument("--signature", help="Search by signature (e.g., 'win.emotet')") parser.add_argument("--output", "-o", default="./quarantine", help="Output directory") parser.add_argument("--limit", type=int, default=10, help="Max results for search") parser.add_argument("--download-all", action="store_true", help="Download all search results") parser.add_argument("--format", choices=["json", "text"], default="text") args = parser.parse_args() if not HAS_REQUESTS: print("[!] Install requests: pip install requests", file=sys.stderr) sys.exit(1) if args.hash: # Direct download result = download_sample(args.hash, args.output) if result: print(f"\n[+] Sample saved to: {result}") else: print("\n[!] Download failed") sys.exit(1) else: # Search query_type = "tag" if args.tag else "signature" query_value = args.tag or args.signature print(f"[*] Searching MalwareBazaar for {query_type}: {query_value}") results = search_bazaar(query_type, query_value, args.limit) if not results or results.get("query_status") != "ok": print(f"[!] Search returned no results: {results.get('query_status', 'error')}") sys.exit(1) samples = results.get("data", []) print(f"[+] Found {len(samples)} samples\n") for i, sample in enumerate(samples): sha256 = sample.get("sha256_hash", "unknown") fname = sample.get("file_name", "unknown") ftype = sample.get("file_type", "unknown") tags = ", ".join(sample.get("tags", [])) print(f" [{i+1}] {sha256[:16]}... | {fname} | {ftype} | Tags: {tags}") if args.download_all: print(f"\n[*] Downloading {len(samples)} samples...") for sample in samples: sha256 = sample.get("sha256_hash") if sha256: download_sample(sha256, args.output) if __name__ == "__main__": main()