#!/usr/bin/env python3 """ capa_rule_writer.py - Generate custom capa rule YAML skeletons from behavior descriptions. Creates properly structured capa rule files ready for customization, with ATT&CK/MBC mappings, feature placeholders, and documentation. Usage: python3 capa_rule_writer.py --name "detect custom packer" --output rules/custom.yml python3 capa_rule_writer.py --name "HTTP C2 beacon" --attack T1071.001 --mbc C0002.005 python3 capa_rule_writer.py --interactive python3 capa_rule_writer.py --from-api CreateRemoteThread,VirtualAllocEx,WriteProcessMemory """ from __future__ import annotations import argparse import datetime import logging import os import re import sys from pathlib import Path from typing import Optional logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # ATT&CK and MBC reference data # --------------------------------------------------------------------------- ATTACK_TACTICS = { "TA0001": "Initial Access", "TA0002": "Execution", "TA0003": "Persistence", "TA0004": "Privilege Escalation", "TA0005": "Defense Evasion", "TA0006": "Credential Access", "TA0007": "Discovery", "TA0008": "Lateral Movement", "TA0009": "Collection", "TA0010": "Exfiltration", "TA0011": "Command and Control", "TA0040": "Impact", "TA0042": "Resource Development", "TA0043": "Reconnaissance", } # Common technique to tactic mapping for auto-suggestion TECHNIQUE_TACTIC = { "T1055": ("TA0005", "Process Injection"), "T1055.001": ("TA0005", "Process Injection: Dynamic-link Library Injection"), "T1055.003": ("TA0005", "Process Injection: Thread Execution Hijacking"), "T1055.012": ("TA0005", "Process Injection: Process Hollowing"), "T1027": ("TA0005", "Obfuscated Files or Information"), "T1027.002": ("TA0005", "Software Packing"), "T1059": ("TA0002", "Command and Scripting Interpreter"), "T1059.001": ("TA0002", "PowerShell"), "T1059.003": ("TA0002", "Windows Command Shell"), "T1071": ("TA0011", "Application Layer Protocol"), "T1071.001": ("TA0011", "Web Protocols"), "T1071.004": ("TA0011", "DNS"), "T1082": ("TA0007", "System Information Discovery"), "T1083": ("TA0007", "File and Directory Discovery"), "T1105": ("TA0011", "Ingress Tool Transfer"), "T1106": ("TA0002", "Native API"), "T1112": ("TA0005", "Modify Registry"), "T1140": ("TA0005", "Deobfuscate/Decode Files or Information"), "T1486": ("TA0040", "Data Encrypted for Impact"), "T1547.001": ("TA0003", "Registry Run Keys / Startup Folder"), "T1560": ("TA0009", "Archive Collected Data"), "T1569.002": ("TA0002", "Service Execution"), } # Common API to feature mapping for auto-generation API_FEATURES = { "CreateRemoteThread": { "namespace": "host-interaction/process/inject", "attack": "T1055", "description": "inject code into another process", }, "VirtualAllocEx": { "namespace": "host-interaction/process/inject", "attack": "T1055", "description": "allocate memory in another process", }, "WriteProcessMemory": { "namespace": "host-interaction/process/inject", "attack": "T1055", "description": "write to another process memory", }, "NtUnmapViewOfSection": { "namespace": "host-interaction/process/inject", "attack": "T1055.012", "description": "process hollowing", }, "RegSetValueEx": { "namespace": "host-interaction/registry", "attack": "T1112", "description": "modify registry value", }, "CryptEncrypt": { "namespace": "data-manipulation/encryption", "attack": "T1486", "description": "encrypt data", }, "InternetOpen": { "namespace": "communication/http", "attack": "T1071.001", "description": "HTTP communication", }, "HttpSendRequest": { "namespace": "communication/http", "attack": "T1071.001", "description": "send HTTP request", }, "WSAStartup": { "namespace": "communication/socket", "attack": "T1071", "description": "initialize network socket", }, "CreateService": { "namespace": "host-interaction/service", "attack": "T1569.002", "description": "create Windows service", }, "IsDebuggerPresent": { "namespace": "anti-analysis/anti-debugging", "attack": "", "description": "detect debugger", }, "GetTickCount": { "namespace": "anti-analysis/anti-debugging", "attack": "", "description": "timing-based anti-debug check", }, } # Capa namespace suggestions NAMESPACES = [ "anti-analysis/anti-debugging", "anti-analysis/anti-vm", "anti-analysis/anti-disassembly", "anti-analysis/packer", "collection", "collection/file-managers", "collection/keylog", "collection/screenshot", "communication/dns", "communication/http", "communication/socket", "communication/named-pipe", "data-manipulation/encryption", "data-manipulation/encoding", "data-manipulation/hashing", "executable/installer", "executable/packer", "host-interaction/file-system", "host-interaction/gui", "host-interaction/network", "host-interaction/os", "host-interaction/process", "host-interaction/process/inject", "host-interaction/registry", "host-interaction/service", "impact/ransomware", "persistence", "persistence/registry", "persistence/scheduled-task", ] # --------------------------------------------------------------------------- # Rule generation # --------------------------------------------------------------------------- def sanitize_rule_name(name: str) -> str: """Create a valid rule name.""" return re.sub(r"[^a-z0-9-]", "-", name.lower().strip()).strip("-") def generate_rule( name: str, description: str = "", namespace: str = "", authors: Optional[list] = None, scope: str = "function", attack_ids: Optional[list] = None, mbc_ids: Optional[list] = None, api_calls: Optional[list] = None, strings: Optional[list] = None, references: Optional[list] = None, ) -> str: """Generate a capa rule YAML string.""" if not authors: authors = [os.getenv("USER", os.getenv("USERNAME", "analyst"))] if not attack_ids: attack_ids = [] if not mbc_ids: mbc_ids = [] if not api_calls: api_calls = [] if not strings: strings = [] if not references: references = [] # Auto-detect namespace from APIs if not specified if not namespace and api_calls: for api in api_calls: if api in API_FEATURES: namespace = API_FEATURES[api]["namespace"] break # Build YAML lines = [] lines.append(f"rule:") lines.append(f" meta:") lines.append(f" name: {name}") if namespace: lines.append(f" namespace: {namespace}") lines.append(f" authors:") for author in authors: lines.append(f" - {author}") lines.append(f" scope: {scope}") if description: lines.append(f" description: {description}") if attack_ids: lines.append(f" att&ck:") for tid in attack_ids: if tid in TECHNIQUE_TACTIC: tactic_id, tech_name = TECHNIQUE_TACTIC[tid] tactic_name = ATTACK_TACTICS.get(tactic_id, "Unknown") lines.append(f" - {tactic_name}::{tech_name} [{tid}]") else: lines.append(f" - {tid}") if mbc_ids: lines.append(f" mbc:") for mid in mbc_ids: lines.append(f" - {mid}") if references: lines.append(f" references:") for ref in references: lines.append(f" - {ref}") lines.append(f" examples:") lines.append(f" - # Add SHA256 hashes of matching samples") lines.append(f" features:") if api_calls and strings: lines.append(f" - and:") for api in api_calls: lines.append(f" - api: {api}") for s in strings: lines.append(f' - string: "{s}"') elif api_calls: if len(api_calls) == 1: lines.append(f" - api: {api_calls[0]}") else: lines.append(f" - and:") for api in api_calls: lines.append(f" - api: {api}") elif strings: if len(strings) == 1: lines.append(f' - string: "{strings[0]}"') else: lines.append(f" - and:") for s in strings: lines.append(f' - string: "{s}"') else: # Placeholder features lines.append(f" - and:") lines.append(f" - api: PLACEHOLDER_API # Replace with target API") lines.append(f" # Additional feature options:") lines.append(f" # - api: FunctionName") lines.append(f' # - string: "search string"') lines.append(f' # - string: /regex pattern/i') lines.append(f" # - number: 0x1234") lines.append(f" # - bytes: 00 11 22 33 = description") lines.append(f" # - offset: 0x1000") lines.append(f" # - mnemonic: rdtsc") lines.append(f" # - arch: i386 / amd64") lines.append(f" # - os: windows / linux") lines.append(f" # - match: other rule name") lines.append(f" # - property/read: System.Environment.UserName") lines.append(f" # - optional:") lines.append(f" # - api: OptionalAPI") return "\n".join(lines) + "\n" def generate_from_apis(api_list: list) -> str: """Generate a rule from a list of API calls.""" # Determine namespace and attack from APIs attack_ids = set() namespace = "" descriptions = [] for api in api_list: if api in API_FEATURES: info = API_FEATURES[api] if info["attack"]: attack_ids.add(info["attack"]) if not namespace: namespace = info["namespace"] descriptions.append(info["description"]) name = " and ".join(descriptions[:3]) if descriptions else "custom API detection" description = f"Detect use of {', '.join(api_list)}" return generate_rule( name=name, description=description, namespace=namespace, attack_ids=list(attack_ids), api_calls=api_list, ) # --------------------------------------------------------------------------- # Interactive mode # --------------------------------------------------------------------------- def interactive_mode() -> str: """Interactively build a capa rule.""" print("=== Capa Rule Builder (Interactive) ===") print() name = input("Rule name: ").strip() if not name: print("Name is required.") sys.exit(1) description = input("Description: ").strip() namespace = input(f"Namespace (e.g., {', '.join(NAMESPACES[:5])}): ").strip() scope = input("Scope [function]: ").strip() or "function" authors = [a.strip() for a in (input("Authors (comma-sep): ").strip() or "analyst").split(",")] attack_input = input("ATT&CK technique IDs (comma-sep, e.g., T1055,T1112): ").strip() attack_ids = [t.strip() for t in attack_input.split(",") if t.strip()] if attack_input else [] mbc_input = input("MBC IDs (comma-sep, e.g., C0002.005): ").strip() mbc_ids = [m.strip() for m in mbc_input.split(",") if m.strip()] if mbc_input else [] api_input = input("API calls (comma-sep, e.g., CreateRemoteThread,VirtualAllocEx): ").strip() api_calls = [a.strip() for a in api_input.split(",") if a.strip()] if api_input else [] string_input = input("String matches (comma-sep): ").strip() strings = [s.strip() for s in string_input.split(",") if s.strip()] if string_input else [] return generate_rule( name=name, description=description, namespace=namespace, authors=authors, scope=scope, attack_ids=attack_ids, mbc_ids=mbc_ids, api_calls=api_calls, strings=strings, ) # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def parse_arguments() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Generate custom capa rule YAML skeletons.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: %(prog)s --name "detect process injection" --attack T1055 --output rules/injection.yml %(prog)s --name "HTTP C2 beacon" --api InternetOpen,HttpSendRequest --attack T1071.001 %(prog)s --from-api CreateRemoteThread,VirtualAllocEx,WriteProcessMemory %(prog)s --interactive """, ) parser.add_argument("--input", "--name", "-n", help="Rule name") parser.add_argument("--description", "-d", help="Rule description") parser.add_argument("--namespace", help="Capa namespace (e.g., host-interaction/process/inject)") parser.add_argument("--scope", default="function", choices=["function", "file", "basic block"], help="Rule scope (default: function)") parser.add_argument("--authors", help="Comma-separated author names") parser.add_argument("--attack", help="Comma-separated ATT&CK technique IDs (e.g., T1055,T1112)") parser.add_argument("--mbc", help="Comma-separated MBC IDs") parser.add_argument("--api", help="Comma-separated API calls for features") parser.add_argument("--strings", help="Comma-separated strings for features") parser.add_argument("--from-api", help="Auto-generate rule from API call list (comma-separated)") parser.add_argument("--interactive", action="store_true", help="Interactive rule builder") parser.add_argument("--output", "-o", help="Output file path") parser.add_argument("--verbose", "-v", action="store_true", help="Verbose output") parser.add_argument("--format", default="json", choices=["json", "text", "csv"], help="Output format (default: json)") return parser.parse_args() def main() -> None: args = parse_arguments() logging.basicConfig( level=logging.DEBUG if args.verbose else logging.INFO, format="%(levelname)s: %(message)s", ) if args.interactive: rule_yaml = interactive_mode() elif args.from_api: api_list = [a.strip() for a in args.from_api.split(",")] rule_yaml = generate_from_apis(api_list) elif args.name: attack_ids = [t.strip() for t in args.attack.split(",")] if args.attack else [] mbc_ids = [m.strip() for m in args.mbc.split(",")] if args.mbc else [] api_calls = [a.strip() for a in args.api.split(",")] if args.api else [] strings = [s.strip() for s in args.strings.split(",")] if args.strings else [] authors = [a.strip() for a in args.authors.split(",")] if args.authors else None references = [] rule_yaml = generate_rule( name=args.name, description=args.description or "", namespace=args.namespace or "", authors=authors, scope=args.scope, attack_ids=attack_ids, mbc_ids=mbc_ids, api_calls=api_calls, strings=strings, references=references, ) else: logger.error("Provide --name, --from-api, or --interactive") sys.exit(1) if args.output: out_path = Path(args.output) out_path.parent.mkdir(parents=True, exist_ok=True) out_path.write_text(rule_yaml) logger.info(f"Rule written to {out_path}") else: print(rule_yaml) if __name__ == "__main__": main()