#!/usr/bin/env python3 """ sigma_rule_writer.py - Generate Sigma detection rules from behavioral observations. Creates properly structured Sigma YAML rules with detection logic, ATT&CK tags, log source configuration, and false positive documentation. Usage: python3 sigma_rule_writer.py --title "Suspicious Process" \ --logsource-product windows --logsource-category process_creation \ --detection-field CommandLine --detection-modifier contains \ --detection-values "powershell,encodedcommand" \ --level high --output rule.yml python3 sigma_rule_writer.py --validate rules/my_rule.yml python3 sigma_rule_writer.py --from-iocs iocs.txt \ --logsource-category process_creation --output rule.yml """ from __future__ import annotations import argparse import json import logging import re import sys import uuid from datetime import date from pathlib import Path from typing import Optional logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- VALID_STATUSES = ["test", "experimental", "stable", "deprecated", "unsupported"] VALID_LEVELS = ["informational", "low", "medium", "high", "critical"] LOGSOURCE_CATEGORIES = [ "process_creation", "file_event", "file_change", "file_rename", "file_delete", "registry_event", "registry_add", "registry_delete", "registry_set", "network_connection", "dns_query", "image_load", "pipe_created", "pipe_connected", "driver_load", "wmi_event", "create_remote_thread", "create_stream_hash", "firewall", "proxy", "webserver", "dns", "antivirus", ] LOGSOURCE_PRODUCTS = [ "windows", "linux", "macos", "aws", "azure", "gcp", ] LOGSOURCE_SERVICES = [ "sysmon", "security", "system", "powershell", "powershell-classic", "taskscheduler", "wmi", "application", "dns-server", "firewall-as", "bits-client", "auditd", "sshd", "sudo", "clamav", ] DETECTION_MODIFIERS = [ "contains", "endswith", "startswith", "re", "cidr", "base64", "base64offset", "windash", "all", "contains|all", "endswith|all", "startswith|all", ] # --------------------------------------------------------------------------- # Rule generation # --------------------------------------------------------------------------- def generate_uuid() -> str: """Generate a UUIDv4 for the rule ID.""" return str(uuid.uuid4()) def build_sigma_rule( title: str, description: str = "", status: str = "experimental", level: str = "medium", author: str = "analyst", logsource_product: str = "", logsource_service: str = "", logsource_category: str = "", detection_field: str = "", detection_modifier: str = "", detection_values: Optional[list] = None, filter_field: str = "", filter_modifier: str = "", filter_values: Optional[list] = None, falsepositives: Optional[list] = None, attack_tags: Optional[list] = None, fields: Optional[list] = None, rule_id: Optional[str] = None, ) -> str: """Build a Sigma rule YAML string from parameters. Args: title: Rule title (max 100 chars recommended). description: Detailed description of what the rule detects. status: Rule status (experimental, test, stable, deprecated, unsupported). level: Detection severity level. author: Rule author name. logsource_product: Log source product (windows, linux, etc.). logsource_service: Log source service (sysmon, security, etc.). logsource_category: Log source category (process_creation, etc.). detection_field: Field name for the detection selection. detection_modifier: Sigma modifier (contains, endswith, re, etc.). detection_values: List of values to match. filter_field: Optional filter field for FP reduction. filter_modifier: Modifier for the filter. filter_values: Values for the filter. falsepositives: List of known false positive scenarios. attack_tags: List of ATT&CK tags (e.g., attack.execution, attack.t1059.001). fields: List of useful fields to include in output. rule_id: Optional UUIDv4; generated if not provided. Returns: A string containing the Sigma rule in YAML format. """ if not detection_values: detection_values = [] if not falsepositives: falsepositives = ["Unknown"] if not attack_tags: attack_tags = [] if not fields: fields = [] if not rule_id: rule_id = generate_uuid() today = date.today().strftime("%Y/%m/%d") lines = [] # Header lines.append(f"title: {title}") lines.append(f"id: {rule_id}") lines.append(f"status: {status}") lines.append(f"level: {level}") lines.append(f"description: {description or title}") lines.append(f"author: {author}") lines.append(f"date: {today}") lines.append(f"modified: {today}") # References lines.append("references:") lines.append(" - https://github.com/SigmaHQ/sigma") # Tags if attack_tags: lines.append("tags:") for tag in attack_tags: lines.append(f" - {tag}") # Logsource lines.append("logsource:") if logsource_product: lines.append(f" product: {logsource_product}") if logsource_service: lines.append(f" service: {logsource_service}") if logsource_category: lines.append(f" category: {logsource_category}") # Detection lines.append("detection:") if detection_field and detection_values: mod_suffix = f"|{detection_modifier}" if detection_modifier else "" lines.append(" selection:") lines.append(f" {detection_field}{mod_suffix}:") for val in detection_values: lines.append(f" - '{val}'") # Filter if filter_field and filter_values: fmod_suffix = f"|{filter_modifier}" if filter_modifier else "" lines.append(" filter:") lines.append(f" {filter_field}{fmod_suffix}:") for val in filter_values: lines.append(f" - '{val}'") lines.append(" condition: selection and not filter") else: lines.append(" condition: selection") else: # Placeholder detection lines.append(" selection:") lines.append(" FieldName|contains:") lines.append(" - 'placeholder_value'") lines.append(" condition: selection") # False positives lines.append("falsepositives:") for fp in falsepositives: lines.append(f" - {fp}") # Fields if fields: lines.append("fields:") for field in fields: lines.append(f" - {field}") return "\n".join(lines) + "\n" def generate_from_iocs( ioc_file: str, logsource_category: str = "process_creation", logsource_product: str = "windows", ) -> str: """Generate a Sigma rule from a file containing IOCs (one per line). Reads IOCs from a text file and creates a Sigma rule with detection logic matching those IOCs in the appropriate field based on the log source category. Args: ioc_file: Path to text file with one IOC per line. logsource_category: Log source category to target. logsource_product: Log source product. Returns: Sigma rule YAML string. """ ioc_path = Path(ioc_file) if not ioc_path.exists(): logger.error(f"IOC file not found: {ioc_file}") sys.exit(1) iocs = [line.strip() for line in ioc_path.read_text().splitlines() if line.strip() and not line.startswith("#")] if not iocs: logger.error("No IOCs found in file") sys.exit(1) # Determine detection field based on category field_map = { "process_creation": "CommandLine", "network_connection": "DestinationHostname", "dns_query": "QueryName", "file_event": "TargetFilename", "registry_event": "TargetObject", "image_load": "ImageLoaded", "proxy": "c-uri", "firewall": "dst_ip", } detection_field = field_map.get(logsource_category, "FieldName") return build_sigma_rule( title=f"IOC Detection - {ioc_path.stem}", description=f"Detects indicators from {ioc_path.name}", logsource_product=logsource_product, logsource_category=logsource_category, detection_field=detection_field, detection_modifier="contains", detection_values=iocs, level="high", ) def validate_rule(rule_path: str) -> dict: """Validate a Sigma rule file for required fields and quality. Checks for: - Valid YAML structure - Required fields (title, id, status, level, logsource, detection) - UUIDv4 format for id - Valid status and level values - ATT&CK tag presence - False positives section Args: rule_path: Path to the Sigma YAML rule file. Returns: Dictionary with validation results and quality score. """ import yaml # lazy import; yaml may not be available everywhere rule_file = Path(rule_path) if not rule_file.exists(): return {"valid": False, "error": f"File not found: {rule_path}"} try: with open(rule_file) as f: rule = yaml.safe_load(f) except yaml.YAMLError as e: return {"valid": False, "error": f"YAML parse error: {e}"} issues = [] warnings = [] # Required fields required_fields = ["title", "id", "status", "level", "logsource", "detection"] for field in required_fields: if field not in rule: issues.append(f"Missing required field: {field}") # UUID format if "id" in rule: try: uuid.UUID(str(rule["id"]), version=4) except ValueError: issues.append("Field 'id' is not a valid UUIDv4") # Status validation if "status" in rule and rule["status"] not in VALID_STATUSES: issues.append(f"Invalid status: {rule['status']}") # Level validation if "level" in rule and rule["level"] not in VALID_LEVELS: issues.append(f"Invalid level: {rule['level']}") # Quality warnings if "tags" not in rule: warnings.append("No ATT&CK tags present") if "falsepositives" not in rule: warnings.append("No false positives documented") if "description" not in rule: warnings.append("No description provided") if "author" not in rule: warnings.append("No author specified") if "title" in rule and len(rule["title"]) > 100: warnings.append("Title exceeds 100 characters") # Detection validation if "detection" in rule: if "condition" not in rule["detection"]: issues.append("Detection section missing 'condition'") valid = len(issues) == 0 quality = "high" if valid and len(warnings) == 0 else "medium" if valid else "low" return { "valid": valid, "issues": issues, "warnings": warnings, "quality_score": quality, "file": str(rule_file), } # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def parse_arguments() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Generate Sigma detection rules from behavioral observations.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: %(prog)s --title "Suspicious PowerShell" \\ --logsource-product windows --logsource-category process_creation \\ --detection-field CommandLine --detection-modifier "contains|all" \\ --detection-values "powershell,-encodedcommand" \\ --level high --attack-tags attack.execution,attack.t1059.001 \\ --output rule.yml %(prog)s --from-iocs iocs.txt --logsource-category dns_query --output dns_rule.yml %(prog)s --validate rules/my_rule.yml """, ) parser.add_argument("--input", "--title", help="Rule title") parser.add_argument("--description", help="Rule description") parser.add_argument("--status", default="experimental", choices=VALID_STATUSES, help="Rule status (default: experimental)") parser.add_argument("--level", default="medium", choices=VALID_LEVELS, help="Severity level (default: medium)") parser.add_argument("--author", default="analyst", help="Rule author") parser.add_argument("--logsource-product", help="Log source product (windows, linux, etc.)") parser.add_argument("--logsource-service", help="Log source service (sysmon, security, etc.)") parser.add_argument("--logsource-category", help="Log source category (process_creation, etc.)") parser.add_argument("--detection-field", help="Field name for detection selection") parser.add_argument("--detection-modifier", help="Sigma modifier (contains, endswith, re, etc.)") parser.add_argument("--detection-values", help="Comma-separated detection values") parser.add_argument("--filter-field", help="Filter field for FP reduction") parser.add_argument("--filter-modifier", help="Filter modifier") parser.add_argument("--filter-values", help="Comma-separated filter values") parser.add_argument("--falsepositives", help="Comma-separated false positive descriptions") parser.add_argument("--attack-tags", help="Comma-separated ATT&CK tags (e.g., attack.execution,attack.t1059.001)") parser.add_argument("--fields", help="Comma-separated output fields") parser.add_argument("--from-iocs", help="Generate rule from IOC file (one IOC per line)") parser.add_argument("--validate", help="Validate an existing Sigma rule file") parser.add_argument("--output", "-o", help="Output file path (default: stdout)") 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", ) # Validation mode if args.validate: result = validate_rule(args.validate) print(json.dumps(result, indent=2)) sys.exit(0 if result.get("valid") else 1) # IOC mode if args.from_iocs: rule_yaml = generate_from_iocs( ioc_file=args.from_iocs, logsource_category=args.logsource_category or "process_creation", logsource_product=args.logsource_product or "windows", ) elif args.title: # Parse list arguments detection_values = [v.strip() for v in args.detection_values.split(",")] if args.detection_values else [] filter_values = [v.strip() for v in args.filter_values.split(",")] if args.filter_values else [] falsepositives = [v.strip() for v in args.falsepositives.split(",")] if args.falsepositives else None attack_tags = [v.strip() for v in args.attack_tags.split(",")] if args.attack_tags else [] fields = [v.strip() for v in args.fields.split(",")] if args.fields else [] rule_yaml = build_sigma_rule( title=args.title, description=args.description or "", status=args.status, level=args.level, author=args.author, logsource_product=args.logsource_product or "", logsource_service=args.logsource_service or "", logsource_category=args.logsource_category or "", detection_field=args.detection_field or "", detection_modifier=args.detection_modifier or "", detection_values=detection_values, filter_field=args.filter_field or "", filter_modifier=args.filter_modifier or "", filter_values=filter_values, falsepositives=falsepositives, attack_tags=attack_tags, fields=fields, ) else: logger.error("Provide --title, --from-iocs, or --validate") 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"Sigma rule written to {out_path}") else: print(rule_yaml) if __name__ == "__main__": main()