# analyzing-cobaltstrike-malleable-c2-profiles Parse and analyze Cobalt Strike Malleable C2 profiles with dissect.cobaltstrike (profiles and beacon-payload configs) and pyMalleableC2 (AST parsing) to extract HTTP/DNS transforms, URIs, headers, sleep/jitter, and injection behavior, then generate network detection signatures. Use when reverse-engineering a captured malleable profile or building detections against Cobalt Strike Beacon traffic. - **Kind:** skill - **Source:** https://github.com/mukul975/Anthropic-Cybersecurity-Skills - **Page:** https://forefy.com/skills/7c8970c6-ec44-4937-adbe-6c65aeebe397 - **API (JSON + files):** https://forefy.com/api/asr/7c8970c6-ec44-4937-adbe-6c65aeebe397 --- ## LICENSE ``` ``` ## SKILL.md --- name: analyzing-cobaltstrike-malleable-c2-profiles description: Parse and analyze Cobalt Strike Malleable C2 profiles with dissect.cobaltstrike (profiles and beacon-payload configs) and pyMalleableC2 (AST parsing) to extract HTTP/DNS transforms, URIs, headers, sleep/jitter, and injection behavior, then generate network detection signatures. Use when reverse-engineering a captured malleable profile or building detections against Cobalt Strike Beacon traffic. domain: cybersecurity subdomain: malware-analysis tags: - cobalt-strike - malleable-c2 - c2-detection - beacon-analysis - network-signatures - threat-hunting - red-team-tools version: '1.0' author: mahipal license: Apache-2.0 nist_csf: - DE.AE-02 - RS.AN-03 - ID.RA-01 - DE.CM-01 mitre_attack: - T1071.001 - T1573.002 - T1001.003 - T1090.004 - T1102 --- # Analyzing CobaltStrike Malleable C2 Profiles ## Overview Cobalt Strike Malleable C2 profiles are domain-specific language scripts that customize how Beacon communicates with the team server, defining HTTP request/response transformations, sleep intervals, jitter values, user agents, URI paths, and process injection behavior. Threat actors use malleable profiles to disguise C2 traffic as legitimate services (Amazon, Google, Slack). Analyzing these profiles reveals network indicators for detection: URI patterns, HTTP headers, POST/GET transforms, DNS settings, and process injection techniques. The `dissect.cobaltstrike` library can parse both profile files and extract configurations from beacon payloads, while `pyMalleableC2` provides AST-based parsing using Lark grammar for programmatic profile manipulation and validation. ## When to Use - When investigating security incidents that require analyzing cobaltstrike malleable c2 profiles - When building detection rules or threat hunting queries for this domain - When SOC analysts need structured procedures for this analysis type - When validating security monitoring coverage for related attack techniques ## Prerequisites - Python 3.9+ with `dissect.cobaltstrike` and/or `pyMalleableC2` - Sample Malleable C2 profiles (available from public repositories) - Understanding of HTTP protocol and Cobalt Strike beacon communication model - Network monitoring tools (Suricata/Snort) for signature deployment - PCAP analysis tools for traffic validation ## Steps 1. Install libraries: `pip install dissect.cobaltstrike` or `pip install pyMalleableC2` 2. Parse profile with `C2Profile.from_path("profile.profile")` 3. Extract HTTP GET/POST block configurations (URIs, headers, parameters) 4. Identify user agent strings and spoof targets 5. Extract sleep time, jitter percentage, and DNS beacon settings 6. Analyze process injection settings (spawn-to, allocation technique) 7. Generate Suricata/Snort signatures from extracted network indicators 8. Compare profile against known threat actor profile collections 9. Extract staging URIs and payload delivery mechanisms 10. Produce detection report with IOCs and recommended network signatures ## Expected Output A JSON report containing extracted C2 URIs, HTTP headers, user agents, sleep/jitter settings, process injection config, spawned process paths, DNS settings, and generated Suricata-compatible detection rules. ## references ``` ``` ## references/api-reference.md # CobaltStrike Malleable C2 Profile Analysis API Reference ## Installation ```bash pip install dissect.cobaltstrike pip install 'dissect.cobaltstrike[full]' # With PCAP support pip install pyMalleableC2 # Alternative parser ``` ## dissect.cobaltstrike API ### Parse Beacon Configuration ```python from dissect.cobaltstrike.beacon import BeaconConfig bconfig = BeaconConfig.from_path("beacon.bin") print(hex(bconfig.watermark)) # 0x5109bf6d print(bconfig.protocol) # https print(bconfig.version) # BeaconVersion(...) print(bconfig.settings) # Full config dict ``` ### Parse Malleable C2 Profile ```python from dissect.cobaltstrike.c2profile import C2Profile profile = C2Profile.from_path("amazon.profile") config = profile.as_dict() print(config["useragent"]) print(config["http-get.uri"]) print(config["sleeptime"]) ``` ### PCAP Analysis ```bash # Extract beacons from PCAP beacon-pcap --extract-beacons traffic.pcap # Decrypt traffic with private key beacon-pcap -p team_server.pem traffic.pcap --beacon beacon.bin ``` ## pyMalleableC2 API ```python from malleableC2 import Profile profile = Profile.from_file("amazon.profile") print(profile.sleeptime) print(profile.useragent) print(profile.http_get.uri) print(profile.http_post.uri) ``` ## Key Profile Settings | Setting | Description | Detection Value | |---------|-------------|-----------------| | `sleeptime` | Callback interval (ms) | Low values = aggressive beaconing | | `jitter` | Sleep randomization % | Timing analysis evasion | | `useragent` | HTTP User-Agent string | Network signature | | `http-get.uri` | GET request URI path | URI-based detection | | `http-post.uri` | POST request URI path | URI-based detection | | `spawnto_x86` | 32-bit spawn process | Process creation detection | | `spawnto_x64` | 64-bit spawn process | Process creation detection | | `pipename` | Named pipe pattern | Named pipe monitoring | | `dns_idle` | DNS idle IP address | DNS beacon detection | | `watermark` | License watermark | Operator attribution | ## Suricata Rule Format ``` alert http $HOME_NET any -> $EXTERNAL_NET any ( msg:"MALWARE CobaltStrike C2 URI"; flow:established,to_server; http.uri; content:"/api/v1/status"; http.header; content:"User-Agent: Mozilla/5.0"; sid:9000001; rev:1; ) ``` ## CLI Usage ```bash python agent.py --input profile.profile --output report.json python agent.py --input parsed_config.json --output report.json ``` ## References - dissect.cobaltstrike: https://github.com/fox-it/dissect.cobaltstrike - pyMalleableC2: https://github.com/byt3bl33d3r/pyMalleableC2 - Unit42 Analysis: https://unit42.paloaltonetworks.com/cobalt-strike-malleable-c2-profile/ - Config Extractor: https://github.com/strozfriedberg/cobaltstrike-config-extractor ## scripts ``` ``` ## scripts/agent.py ```python #!/usr/bin/env python3 """CobaltStrike Malleable C2 Profile Analyzer - parses profiles to extract C2 indicators, detection signatures, and evasion techniques""" # For authorized security research and defensive analysis only import argparse import json import re from collections import Counter from datetime import datetime from pathlib import Path try: from dissect.cobaltstrike.c2profile import C2Profile HAS_DISSECT = True except ImportError: HAS_DISSECT = False RUN_KEY_SUSPICIOUS = ["powershell", "cmd.exe", "mshta", "rundll32", "regsvr32", "wscript", "cscript"] KNOWN_SPOOF_TARGETS = { "amazon": "Amazon CDN impersonation", "google": "Google services impersonation", "microsoft": "Microsoft services impersonation", "slack": "Slack API impersonation", "cloudfront": "CloudFront CDN impersonation", "jquery": "jQuery CDN impersonation", "outlook": "Outlook Web impersonation", "onedrive": "OneDrive impersonation", } def load_data(path): return json.loads(Path(path).read_text(encoding="utf-8")) def parse_profile_with_dissect(profile_path): """Parse a .profile file using dissect.cobaltstrike C2Profile.""" if not HAS_DISSECT: return None profile = C2Profile.from_path(profile_path) return profile.as_dict() def parse_profile_regex(content): """Regex-based parser for malleable C2 profile when dissect is unavailable.""" config = {} set_pattern = re.compile(r'set\s+(\w+)\s+"([^"]*)"', re.MULTILINE) for match in set_pattern.finditer(content): config[match.group(1)] = match.group(2) block_pattern = re.compile(r'(http-get|http-post|http-stager|https-certificate|dns-beacon|process-inject|post-ex)\s*\{', re.MULTILINE) for match in block_pattern.finditer(content): config.setdefault("blocks", []).append(match.group(1)) uri_pattern = re.compile(r'set\s+uri\s+"([^"]*)"', re.MULTILINE) for match in uri_pattern.finditer(content): config.setdefault("uris", []).append(match.group(1)) header_pattern = re.compile(r'header\s+"([^"]+)"\s+"([^"]*)"', re.MULTILINE) for match in header_pattern.finditer(content): config.setdefault("headers", []).append({"name": match.group(1), "value": match.group(2)}) spawn_pattern = re.compile(r'set\s+spawnto_x(?:86|64)\s+"([^"]*)"', re.MULTILINE) for match in spawn_pattern.finditer(content): config.setdefault("spawn_to", []).append(match.group(1)) return config def analyze_profile(config): """Analyze parsed profile configuration for detection opportunities.""" findings = [] ua = config.get("useragent", config.get("user_agent", "")) if ua: findings.append({ "type": "user_agent_identified", "severity": "info", "resource": "http-config", "detail": f"User-Agent: {ua[:100]}", "indicator": ua, }) for target, desc in KNOWN_SPOOF_TARGETS.items(): if target.lower() in ua.lower(): findings.append({ "type": "service_impersonation", "severity": "medium", "resource": "user-agent", "detail": f"{desc} detected in User-Agent string", }) sleeptime = config.get("sleeptime", config.get("sleep_time", "")) jitter = config.get("jitter", "") if sleeptime: try: sleep_ms = int(sleeptime) if sleep_ms < 1000: findings.append({ "type": "aggressive_beaconing", "severity": "high", "resource": "beacon-config", "detail": f"Very low sleep time: {sleep_ms}ms - aggressive C2 callback rate", }) except ValueError: pass uris = config.get("uris", []) for uri in uris: findings.append({ "type": "c2_uri", "severity": "high", "resource": "http-config", "detail": f"C2 URI path: {uri}", "indicator": uri, }) headers = config.get("headers", []) for h in headers: name = h.get("name", "") if isinstance(h, dict) else str(h) value = h.get("value", "") if isinstance(h, dict) else "" if name.lower() in ("host", "cookie", "authorization"): findings.append({ "type": "c2_header", "severity": "medium", "resource": "http-config", "detail": f"Custom header: {name}: {value[:60]}", }) spawn_to = config.get("spawn_to", config.get("spawnto_x86", [])) if isinstance(spawn_to, str): spawn_to = [spawn_to] for proc in spawn_to: findings.append({ "type": "spawn_to_process", "severity": "high", "resource": "process-inject", "detail": f"Beacon spawns to: {proc}", "indicator": proc, }) pipename = config.get("pipename", config.get("pipename_stager", "")) if pipename: findings.append({ "type": "named_pipe", "severity": "high", "resource": "process-inject", "detail": f"Named pipe: {pipename}", "indicator": pipename, }) dns_idle = config.get("dns_idle", "") if dns_idle: findings.append({ "type": "dns_beacon_config", "severity": "medium", "resource": "dns-beacon", "detail": f"DNS idle IP: {dns_idle}", }) watermark = config.get("watermark", "") if watermark: findings.append({ "type": "watermark", "severity": "info", "resource": "beacon-config", "detail": f"Beacon watermark: {watermark}", }) return findings def generate_suricata_rules(findings, sid_start=9000001): """Generate Suricata rules from extracted indicators.""" rules = [] sid = sid_start for f in findings: if f["type"] == "c2_uri" and f.get("indicator"): uri = f["indicator"].replace('"', '\\"') rules.append( f'alert http $HOME_NET any -> $EXTERNAL_NET any ' f'(msg:"MALWARE CobaltStrike Malleable C2 URI {uri}"; ' f'flow:established,to_server; ' f'http.uri; content:"{uri}"; ' f'sid:{sid}; rev:1;)' ) sid += 1 elif f["type"] == "named_pipe" and f.get("indicator"): pipe = f["indicator"] rules.append( f'# Named pipe detection requires endpoint monitoring: {pipe}' ) return rules def analyze(data): if isinstance(data, str): config = parse_profile_regex(data) elif isinstance(data, dict): config = data else: config = data[0] if isinstance(data, list) and data else {} return analyze_profile(config) def generate_report(input_path): path = Path(input_path) if path.suffix in (".profile", ".txt"): content = path.read_text(encoding="utf-8") config = parse_profile_regex(content) findings = analyze_profile(config) else: data = load_data(input_path) if isinstance(data, list): findings = [] for profile in data: findings.extend(analyze_profile(profile)) else: findings = analyze_profile(data) sev = Counter(f["severity"] for f in findings) iocs = [f.get("indicator", "") for f in findings if f.get("indicator")] rules = generate_suricata_rules(findings) return { "report": "cobaltstrike_malleable_c2_analysis", "generated_at": datetime.utcnow().isoformat() + "Z", "total_findings": len(findings), "severity_summary": dict(sev), "extracted_iocs": iocs, "suricata_rules": rules, "findings": findings, } def main(): ap = argparse.ArgumentParser(description="CobaltStrike Malleable C2 Profile Analyzer") ap.add_argument("--input", required=True, help="Input .profile file or JSON with parsed config") ap.add_argument("--output", help="Output JSON report path") args = ap.parse_args() report = generate_report(args.input) out = json.dumps(report, indent=2) if args.output: Path(args.output).write_text(out, encoding="utf-8") print(f"Report written to {args.output}") else: print(out) if __name__ == "__main__": main() ```