#!/usr/bin/env python3 """ stix_generator.py - Generate STIX 2.1 threat intelligence bundles. Creates valid STIX 2.1 JSON bundles with Malware, Indicator, AttackPattern, ThreatActor, Campaign, and Relationship SDOs. Applies TLP marking definitions. Usage: python3 stix_generator.py --malware-name TrojanX --malware-type trojan \ --hashes sha256:abc123 --network-iocs evil.com --attack-patterns T1055 \ --tlp green --output report.json python3 stix_generator.py --malware-name "APT Backdoor" --malware-type backdoor \ --threat-actor "APT-X" --campaign "Operation Shadow" \ --hashes sha256:aabbcc --attack-patterns T1055.012,T1071.001 \ --tlp amber --output apt_report.json """ from __future__ import annotations import argparse import json import logging import re import sys import uuid from datetime import datetime, timezone from pathlib import Path from typing import Optional logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # STIX 2.1 Constants # --------------------------------------------------------------------------- STIX_SPEC_VERSION = "2.1" TLP_MARKINGS = { "clear": { "type": "marking-definition", "spec_version": "2.1", "id": "marking-definition--613f2e26-407d-48c7-9eca-b8e91df99dc9", "created": "2017-01-20T00:00:00.000Z", "definition_type": "tlp", "name": "TLP:CLEAR", "definition": {"tlp": "clear"}, }, "white": { "type": "marking-definition", "spec_version": "2.1", "id": "marking-definition--613f2e26-407d-48c7-9eca-b8e91df99dc9", "created": "2017-01-20T00:00:00.000Z", "definition_type": "tlp", "name": "TLP:CLEAR", "definition": {"tlp": "clear"}, }, "green": { "type": "marking-definition", "spec_version": "2.1", "id": "marking-definition--34098fce-860f-48ae-8e50-ebd3cc5e41da", "created": "2017-01-20T00:00:00.000Z", "definition_type": "tlp", "name": "TLP:GREEN", "definition": {"tlp": "green"}, }, "amber": { "type": "marking-definition", "spec_version": "2.1", "id": "marking-definition--f88d31f6-486f-44da-b317-01333bde0b82", "created": "2017-01-20T00:00:00.000Z", "definition_type": "tlp", "name": "TLP:AMBER", "definition": {"tlp": "amber"}, }, "amber+strict": { "type": "marking-definition", "spec_version": "2.1", "id": "marking-definition--826578e1-40a3-4b46-a8d0-ad56a0667d91", "created": "2017-01-20T00:00:00.000Z", "definition_type": "tlp", "name": "TLP:AMBER+STRICT", "definition": {"tlp": "amber+strict"}, }, "red": { "type": "marking-definition", "spec_version": "2.1", "id": "marking-definition--5e57c739-391a-4eb3-b6be-7d15ca92d5ed", "created": "2017-01-20T00:00:00.000Z", "definition_type": "tlp", "name": "TLP:RED", "definition": {"tlp": "red"}, }, } MALWARE_TYPES = [ "adware", "backdoor", "bot", "bootkit", "ddos", "downloader", "dropper", "exploit-kit", "keylogger", "ransomware", "remote-access-trojan", "resource-exploitation", "rogue-security-software", "rootkit", "screen-capture", "spyware", "trojan", "unknown", "virus", "webshell", "wiper", "worm", ] # Common ATT&CK techniques with names ATTACK_TECHNIQUES = { "T1055": {"name": "Process Injection", "kill_chain": "defense-evasion"}, "T1055.001": {"name": "Dynamic-link Library Injection", "kill_chain": "defense-evasion"}, "T1055.003": {"name": "Thread Execution Hijacking", "kill_chain": "defense-evasion"}, "T1055.012": {"name": "Process Hollowing", "kill_chain": "defense-evasion"}, "T1027": {"name": "Obfuscated Files or Information", "kill_chain": "defense-evasion"}, "T1027.002": {"name": "Software Packing", "kill_chain": "defense-evasion"}, "T1059": {"name": "Command and Scripting Interpreter", "kill_chain": "execution"}, "T1059.001": {"name": "PowerShell", "kill_chain": "execution"}, "T1059.003": {"name": "Windows Command Shell", "kill_chain": "execution"}, "T1071": {"name": "Application Layer Protocol", "kill_chain": "command-and-control"}, "T1071.001": {"name": "Web Protocols", "kill_chain": "command-and-control"}, "T1071.004": {"name": "DNS", "kill_chain": "command-and-control"}, "T1082": {"name": "System Information Discovery", "kill_chain": "discovery"}, "T1083": {"name": "File and Directory Discovery", "kill_chain": "discovery"}, "T1105": {"name": "Ingress Tool Transfer", "kill_chain": "command-and-control"}, "T1106": {"name": "Native API", "kill_chain": "execution"}, "T1112": {"name": "Modify Registry", "kill_chain": "defense-evasion"}, "T1140": {"name": "Deobfuscate/Decode Files or Information", "kill_chain": "defense-evasion"}, "T1486": {"name": "Data Encrypted for Impact", "kill_chain": "impact"}, "T1547.001": {"name": "Registry Run Keys / Startup Folder", "kill_chain": "persistence"}, "T1560": {"name": "Archive Collected Data", "kill_chain": "collection"}, "T1569.002": {"name": "Service Execution", "kill_chain": "execution"}, } # --------------------------------------------------------------------------- # STIX object helpers # --------------------------------------------------------------------------- def stix_id(obj_type: str) -> str: """Generate a STIX 2.1 identifier.""" return f"{obj_type}--{uuid.uuid4()}" def stix_timestamp() -> str: """Generate an ISO 8601 timestamp.""" return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z" def base_object(obj_type: str, tlp_id: Optional[str] = None) -> dict: """Create a base STIX object with common properties.""" now = stix_timestamp() obj = { "type": obj_type, "spec_version": STIX_SPEC_VERSION, "id": stix_id(obj_type), "created": now, "modified": now, } if tlp_id: obj["object_marking_refs"] = [tlp_id] return obj # --------------------------------------------------------------------------- # SDO creators # --------------------------------------------------------------------------- def create_malware( name: str, malware_type: str = "unknown", description: str = "", is_family: bool = True, aliases: Optional[list] = None, tlp_id: Optional[str] = None, ) -> dict: """Create a STIX Malware SDO.""" obj = base_object("malware", tlp_id) obj["name"] = name obj["malware_types"] = [malware_type] if malware_type else ["unknown"] obj["is_family"] = is_family if description: obj["description"] = description if aliases: obj["aliases"] = aliases return obj def create_indicator( pattern: str, pattern_type: str = "stix", name: str = "", description: str = "", indicator_types: Optional[list] = None, valid_from: Optional[str] = None, tlp_id: Optional[str] = None, ) -> dict: """Create a STIX Indicator SDO.""" obj = base_object("indicator", tlp_id) obj["pattern"] = pattern obj["pattern_type"] = pattern_type obj["valid_from"] = valid_from or obj["created"] if name: obj["name"] = name if description: obj["description"] = description obj["indicator_types"] = indicator_types or ["malicious-activity"] return obj def create_attack_pattern( technique_id: str, name: str = "", description: str = "", kill_chain_phase: str = "", tlp_id: Optional[str] = None, ) -> dict: """Create a STIX AttackPattern SDO mapped to ATT&CK.""" obj = base_object("attack-pattern", tlp_id) tech_info = ATTACK_TECHNIQUES.get(technique_id, {}) obj["name"] = name or tech_info.get("name", technique_id) if description: obj["description"] = description obj["external_references"] = [{ "source_name": "mitre-attack", "external_id": technique_id, "url": f"https://attack.mitre.org/techniques/{technique_id.replace('.', '/')}", }] kc = kill_chain_phase or tech_info.get("kill_chain", "") if kc: obj["kill_chain_phases"] = [{ "kill_chain_name": "mitre-attack", "phase_name": kc, }] return obj def create_threat_actor( name: str, description: str = "", threat_actor_types: Optional[list] = None, aliases: Optional[list] = None, tlp_id: Optional[str] = None, ) -> dict: """Create a STIX ThreatActor SDO.""" obj = base_object("threat-actor", tlp_id) obj["name"] = name obj["threat_actor_types"] = threat_actor_types or ["unknown"] if description: obj["description"] = description if aliases: obj["aliases"] = aliases return obj def create_campaign( name: str, description: str = "", first_seen: Optional[str] = None, tlp_id: Optional[str] = None, ) -> dict: """Create a STIX Campaign SDO.""" obj = base_object("campaign", tlp_id) obj["name"] = name if description: obj["description"] = description if first_seen: obj["first_seen"] = first_seen return obj def create_relationship( source_ref: str, relationship_type: str, target_ref: str, description: str = "", tlp_id: Optional[str] = None, ) -> dict: """Create a STIX Relationship SRO.""" obj = base_object("relationship", tlp_id) obj["relationship_type"] = relationship_type obj["source_ref"] = source_ref obj["target_ref"] = target_ref if description: obj["description"] = description return obj def create_report( name: str, object_refs: list, description: str = "", report_types: Optional[list] = None, published: Optional[str] = None, tlp_id: Optional[str] = None, ) -> dict: """Create a STIX Report SDO.""" obj = base_object("report", tlp_id) obj["name"] = name obj["object_refs"] = object_refs obj["published"] = published or obj["created"] obj["report_types"] = report_types or ["malware"] if description: obj["description"] = description return obj # --------------------------------------------------------------------------- # Pattern builders # --------------------------------------------------------------------------- def build_hash_pattern(hash_type: str, hash_value: str) -> str: """Build a STIX pattern for a file hash.""" hash_map = { "md5": "MD5", "sha1": "SHA-1", "sha256": "SHA-256", "sha512": "SHA-512", "ssdeep": "SSDEEP", } stix_hash = hash_map.get(hash_type.lower(), hash_type.upper()) return f"[file:hashes.'{stix_hash}' = '{hash_value}']" def build_domain_pattern(domain: str) -> str: """Build a STIX pattern for a domain name.""" return f"[domain-name:value = '{domain}']" def build_ip_pattern(ip: str) -> str: """Build a STIX pattern for an IP address.""" ip_type = "ipv6-addr" if ":" in ip else "ipv4-addr" return f"[{ip_type}:value = '{ip}']" def build_url_pattern(url: str) -> str: """Build a STIX pattern for a URL.""" return f"[url:value = '{url}']" def build_filename_pattern(filename: str) -> str: """Build a STIX pattern for a filename.""" return f"[file:name = '{filename}']" def is_ip_address(value: str) -> bool: """Check if string looks like an IP address.""" # Simple check - supports IPv4 and basic IPv6 ipv4_pattern = re.compile(r"^\d{1,3}(\.\d{1,3}){3}(:\d+)?$") return bool(ipv4_pattern.match(value)) or "::" in value # --------------------------------------------------------------------------- # Bundle builder # --------------------------------------------------------------------------- def build_bundle( malware_name: str, malware_type: str = "unknown", description: str = "", hashes: Optional[list] = None, network_iocs: Optional[list] = None, file_indicators: Optional[list] = None, attack_patterns: Optional[list] = None, threat_actor: Optional[str] = None, campaign: Optional[str] = None, custom_pattern: Optional[str] = None, tlp: str = "clear", ) -> dict: """Build a complete STIX 2.1 bundle.""" objects = [] tlp_def = TLP_MARKINGS.get(tlp.lower(), TLP_MARKINGS["clear"]) tlp_id = tlp_def["id"] objects.append(tlp_def) # Create Malware SDO malware_sdo = create_malware( name=malware_name, malware_type=malware_type, description=description, tlp_id=tlp_id, ) objects.append(malware_sdo) all_indicator_ids = [] all_attack_ids = [] # Create hash indicators if hashes: for h in hashes: if ":" in h: hash_type, hash_value = h.split(":", 1) else: # Auto-detect hash type if len(h) == 32: hash_type, hash_value = "md5", h elif len(h) == 40: hash_type, hash_value = "sha1", h elif len(h) == 64: hash_type, hash_value = "sha256", h else: hash_type, hash_value = "sha256", h pattern = build_hash_pattern(hash_type, hash_value) indicator = create_indicator( pattern=pattern, name=f"{malware_name} - {hash_type.upper()} hash", description=f"File hash indicator for {malware_name}", indicator_types=["malicious-activity"], tlp_id=tlp_id, ) objects.append(indicator) all_indicator_ids.append(indicator["id"]) # Relationship: indicator -> indicates -> malware rel = create_relationship( source_ref=indicator["id"], relationship_type="indicates", target_ref=malware_sdo["id"], tlp_id=tlp_id, ) objects.append(rel) # Create network IOC indicators if network_iocs: for ioc in network_iocs: ioc = ioc.strip() if is_ip_address(ioc.split(":")[0]): ip_part = ioc.split(":")[0] pattern = build_ip_pattern(ip_part) ind_name = f"{malware_name} - C2 IP {ip_part}" elif ioc.startswith("http"): pattern = build_url_pattern(ioc) ind_name = f"{malware_name} - C2 URL" else: pattern = build_domain_pattern(ioc) ind_name = f"{malware_name} - C2 domain {ioc}" indicator = create_indicator( pattern=pattern, name=ind_name, description=f"Network indicator for {malware_name}", indicator_types=["malicious-activity"], tlp_id=tlp_id, ) objects.append(indicator) all_indicator_ids.append(indicator["id"]) rel = create_relationship( source_ref=indicator["id"], relationship_type="indicates", target_ref=malware_sdo["id"], tlp_id=tlp_id, ) objects.append(rel) # Create file name indicators if file_indicators: for fname in file_indicators: pattern = build_filename_pattern(fname.strip()) indicator = create_indicator( pattern=pattern, name=f"{malware_name} - file {fname.strip()}", description=f"File indicator for {malware_name}", indicator_types=["malicious-activity"], tlp_id=tlp_id, ) objects.append(indicator) all_indicator_ids.append(indicator["id"]) rel = create_relationship( source_ref=indicator["id"], relationship_type="indicates", target_ref=malware_sdo["id"], tlp_id=tlp_id, ) objects.append(rel) # Create custom pattern indicator if custom_pattern: indicator = create_indicator( pattern=custom_pattern, name=f"{malware_name} - custom indicator", description=f"Custom detection pattern for {malware_name}", indicator_types=["malicious-activity"], tlp_id=tlp_id, ) objects.append(indicator) all_indicator_ids.append(indicator["id"]) rel = create_relationship( source_ref=indicator["id"], relationship_type="indicates", target_ref=malware_sdo["id"], tlp_id=tlp_id, ) objects.append(rel) # Create ATT&CK pattern SDOs if attack_patterns: for tid in attack_patterns: tid = tid.strip() ap = create_attack_pattern(technique_id=tid, tlp_id=tlp_id) objects.append(ap) all_attack_ids.append(ap["id"]) # Relationship: malware -> uses -> attack-pattern rel = create_relationship( source_ref=malware_sdo["id"], relationship_type="uses", target_ref=ap["id"], tlp_id=tlp_id, ) objects.append(rel) # Create Threat Actor ta_sdo = None if threat_actor: ta_sdo = create_threat_actor( name=threat_actor, description=f"Threat actor associated with {malware_name}", tlp_id=tlp_id, ) objects.append(ta_sdo) # Relationship: threat-actor -> uses -> malware rel = create_relationship( source_ref=ta_sdo["id"], relationship_type="uses", target_ref=malware_sdo["id"], tlp_id=tlp_id, ) objects.append(rel) # Create Campaign campaign_sdo = None if campaign: campaign_sdo = create_campaign( name=campaign, description=f"Campaign involving {malware_name}", tlp_id=tlp_id, ) objects.append(campaign_sdo) # Relationship: campaign -> uses -> malware rel = create_relationship( source_ref=campaign_sdo["id"], relationship_type="uses", target_ref=malware_sdo["id"], tlp_id=tlp_id, ) objects.append(rel) if ta_sdo: # Relationship: threat-actor -> attributed-to -> campaign rel = create_relationship( source_ref=campaign_sdo["id"], relationship_type="attributed-to", target_ref=ta_sdo["id"], tlp_id=tlp_id, ) objects.append(rel) # Create Report SDO that references everything all_refs = [o["id"] for o in objects if o["type"] != "marking-definition"] report = create_report( name=f"Malware Analysis Report: {malware_name}", object_refs=all_refs, description=description or f"Analysis report for {malware_name}", tlp_id=tlp_id, ) objects.append(report) # Build bundle bundle = { "type": "bundle", "id": f"bundle--{uuid.uuid4()}", "objects": objects, } return bundle # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def parse_arguments() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Generate STIX 2.1 threat intelligence bundles from analysis findings.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: %(prog)s --malware-name TrojanX --malware-type trojan \\ --hashes sha256:aabbccdd --network-iocs evil.com \\ --attack-patterns T1055,T1071.001 --tlp green --output report.json %(prog)s --malware-name "APT Backdoor" --malware-type backdoor \\ --threat-actor "APT-X" --campaign "Operation Shadow" \\ --hashes md5:abc123,sha256:def456 --tlp amber --output apt.json %(prog)s --malware-name RansomY --malware-type ransomware \\ --custom-pattern "[file:hashes.'SHA-256' = 'abc']" \\ --output ransom.json """, ) parser.add_argument("--input", "--malware-name", required=True, help="Malware name or family") parser.add_argument("--malware-type", default="unknown", choices=MALWARE_TYPES, help="Malware type (default: unknown)") parser.add_argument("--description", help="Malware description") parser.add_argument("--hashes", help="File hashes (comma-sep, format: type:value or auto-detect)") parser.add_argument("--network-iocs", help="Network IOCs: domains, IPs, URLs (comma-sep)") parser.add_argument("--file-indicators", help="File name indicators (comma-sep)") parser.add_argument("--attack-patterns", help="ATT&CK technique IDs (comma-sep, e.g., T1055,T1071.001)") parser.add_argument("--threat-actor", help="Threat actor name") parser.add_argument("--campaign", help="Campaign name") parser.add_argument("--custom-pattern", help="Custom STIX pattern expression") parser.add_argument("--tlp", default="clear", choices=list(TLP_MARKINGS.keys()), help="TLP marking (default: clear)") parser.add_argument("--output", "-o", help="Output JSON file (default: stdout)") parser.add_argument("--pretty", action="store_true", default=True, help="Pretty-print JSON (default)") parser.add_argument("--compact", action="store_true", help="Compact JSON output") 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", ) # Parse list arguments hashes = [h.strip() for h in args.hashes.split(",")] if args.hashes else None network_iocs = [n.strip() for n in args.network_iocs.split(",")] if args.network_iocs else None file_indicators = [f.strip() for f in args.file_indicators.split(",")] if args.file_indicators else None attack_patterns = [t.strip() for t in args.attack_patterns.split(",")] if args.attack_patterns else None bundle = build_bundle( malware_name=args.malware_name, malware_type=args.malware_type, description=args.description or "", hashes=hashes, network_iocs=network_iocs, file_indicators=file_indicators, attack_patterns=attack_patterns, threat_actor=args.threat_actor, campaign=args.campaign, custom_pattern=args.custom_pattern, tlp=args.tlp, ) indent = None if args.compact else 2 output_json = json.dumps(bundle, indent=indent, ensure_ascii=False) if args.output: Path(args.output).write_text(output_json) logger.info(f"STIX bundle written to {args.output}") obj_counts = {} for obj in bundle["objects"]: t = obj["type"] obj_counts[t] = obj_counts.get(t, 0) + 1 logger.info(f"Objects: {obj_counts}") else: print(output_json) if __name__ == "__main__": main()