#!/usr/bin/env python3 """Malware config extractor — extract embedded configurations from malware samples.""" from __future__ import annotations import argparse import base64 import hashlib import json import math import re import struct import sys from collections import Counter from pathlib import Path def compute_hashes(filepath: str) -> dict: """Compute file hashes.""" data = Path(filepath).read_bytes() return { "md5": hashlib.md5(data).hexdigest(), "sha256": hashlib.sha256(data).hexdigest(), "size_bytes": len(data), } def calculate_entropy(data: bytes) -> float: """Calculate Shannon entropy.""" if not data: return 0.0 freq = Counter(data) length = len(data) return -sum((c / length) * math.log2(c / length) for c in freq.values()) def extract_pe_resources(filepath: str) -> list[dict]: """Extract PE resource entries that may contain configs.""" try: import pefile except ImportError: return [{"error": "pefile not installed"}] try: pe = pefile.PE(filepath) except Exception: return [] resources = [] if hasattr(pe, "DIRECTORY_ENTRY_RESOURCE"): for entry in pe.DIRECTORY_ENTRY_RESOURCE.entries: type_name = str(entry.name) if entry.name else f"type_{entry.id}" if hasattr(entry, "directory"): for res in entry.directory.entries: res_name = str(res.name) if res.name else f"id_{res.id}" if hasattr(res, "directory"): for lang in res.directory.entries: offset = lang.data.struct.OffsetToData size = lang.data.struct.Size data = pe.get_data(offset, size) entropy = calculate_entropy(data) resources.append({ "type": type_name, "name": res_name, "size": size, "entropy": round(entropy, 2), "high_entropy": entropy > 6.5, "first_bytes": data[:16].hex(), }) return resources def extract_pe_overlay(filepath: str) -> dict: """Extract PE overlay data (data appended after PE sections).""" try: import pefile except ImportError: return {"error": "pefile not installed"} try: pe = pefile.PE(filepath) except Exception: return {"overlay_present": False} overlay_offset = pe.get_overlay_data_start_offset() if overlay_offset is None: return {"overlay_present": False} file_size = Path(filepath).stat().st_size overlay_size = file_size - overlay_offset overlay_data = Path(filepath).read_bytes()[overlay_offset:] entropy = calculate_entropy(overlay_data[:4096]) return { "overlay_present": True, "offset": overlay_offset, "size": overlay_size, "entropy": round(entropy, 2), "first_bytes": overlay_data[:32].hex(), "sha256": hashlib.sha256(overlay_data).hexdigest(), } def xor_scan(data: bytes, min_length: int = 8) -> list[dict]: """Scan for XOR-encrypted config data using known plaintext patterns.""" results = [] # Common config indicators that might be XOR-encrypted known_plaintexts = [ b"http://", b"https://", b".exe", b".dll", b"cmd.exe", b"powershell", b"User-Agent", b"Mozilla/", b"Content-Type", ] for key_byte in range(1, 256): decrypted = bytes(b ^ key_byte for b in data) for plaintext in known_plaintexts: idx = decrypted.find(plaintext) if idx != -1: # Extract surrounding context start = max(0, idx - 16) end = min(len(decrypted), idx + len(plaintext) + 64) context = decrypted[start:end] printable = "".join(chr(b) if 0x20 <= b < 0x7f else "." for b in context) results.append({ "xor_key": f"0x{key_byte:02x}", "offset": idx, "matched_plaintext": plaintext.decode(), "context": printable, }) return results def extract_urls(filepath: str) -> list[str]: """Extract URLs from binary data.""" data = Path(filepath).read_bytes() urls = [] url_pattern = rb'https?://[\x20-\x7e]{5,200}' for match in re.finditer(url_pattern, data): url = match.group().decode("ascii", errors="ignore") urls.append(url) return list(set(urls)) def extract_ip_addresses(filepath: str) -> list[str]: """Extract IP addresses from binary data.""" data = Path(filepath).read_bytes() ip_pattern = rb'(?:\d{1,3}\.){3}\d{1,3}' ips = set() for match in re.finditer(ip_pattern, data): ip = match.group().decode() octets = ip.split(".") if all(0 <= int(o) <= 255 for o in octets): # Filter out common false positives if not ip.startswith(("0.", "127.", "255.")): ips.add(ip) return sorted(ips) def extract_base64_configs(filepath: str) -> list[dict]: """Extract base64-encoded configurations.""" data = Path(filepath).read_bytes() configs = [] b64_pattern = rb'[A-Za-z0-9+/]{32,}={0,2}' for match in re.finditer(b64_pattern, data): try: decoded = base64.b64decode(match.group()) # Check if decoded data looks like a config text = decoded.decode("utf-8", errors="ignore") if any(kw in text.lower() for kw in ("host", "port", "password", "key", "http", "mutex", "install", "version")): configs.append({ "offset": match.start(), "encoded_length": len(match.group()), "decoded_text": text[:200], }) except Exception: pass return configs def find_config_structures(filepath: str) -> list[dict]: """Find structured config data (JSON, XML, INI-like patterns).""" data = Path(filepath).read_bytes() configs = [] # Look for embedded JSON json_pattern = rb'\{["\x27][a-zA-Z_]+["\x27]\s*:\s*["\x27\d\[\{]' for match in re.finditer(json_pattern, data): start = match.start() # Try to find matching closing brace depth = 0 end = start for i in range(start, min(start + 10000, len(data))): if data[i:i + 1] == b"{": depth += 1 elif data[i:i + 1] == b"}": depth -= 1 if depth == 0: end = i + 1 break if end > start: try: text = data[start:end].decode("utf-8", errors="ignore") parsed = json.loads(text) configs.append({ "type": "json", "offset": start, "size": end - start, "content": parsed, }) except json.JSONDecodeError: pass return configs def main() -> None: parser = argparse.ArgumentParser(description="Malware Config Extractor") parser.add_argument("--input", required=True, help="Input malware sample") parser.add_argument( "--mode", choices=["resources", "overlay", "xor-scan", "urls", "base64", "structures", "full"], default="full", help="Extraction mode", ) parser.add_argument("--output", default="config_extraction.json", help="Output file path") parser.add_argument("--format", choices=["json", "csv", "text"], default="json") args = parser.parse_args() input_path = Path(args.input) if not input_path.exists(): print(f"[!] File not found: {args.input}", file=sys.stderr) sys.exit(1) hashes = compute_hashes(args.input) print(f"[*] Config Extractor — mode: {args.mode}") print(f"[*] Input: {args.input} ({hashes['size_bytes']} bytes)") print(f"[*] SHA-256: {hashes['sha256']}") results = {"input_file": args.input, "hashes": hashes, "mode": args.mode} if args.mode in ("resources", "full"): print("[*] Extracting PE resources...") results["resources"] = extract_pe_resources(args.input) if args.mode in ("overlay", "full"): print("[*] Checking for PE overlay...") results["overlay"] = extract_pe_overlay(args.input) if args.mode in ("xor-scan", "full"): print("[*] Scanning for XOR-encrypted configs...") results["xor_scan"] = xor_scan(Path(args.input).read_bytes()) if args.mode in ("urls", "full"): print("[*] Extracting URLs...") results["urls"] = extract_urls(args.input) if args.mode in ("base64", "full"): print("[*] Extracting base64-encoded configs...") results["base64_configs"] = extract_base64_configs(args.input) if args.mode in ("structures", "full"): print("[*] Finding structured config data...") results["config_structures"] = find_config_structures(args.input) if args.mode == "full": results["ip_addresses"] = extract_ip_addresses(args.input) output_path = Path(args.output) output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_text(json.dumps(results, indent=2, default=str)) print(f"[*] Results written to {args.output}") if __name__ == "__main__": main()