Sigma Detection Rules

# Handles both explicit columns and nested Windows EventData fields
"EventData.CommandLine" → tries "EventData.CommandLine" → falls back to "CommandLine"
"Image" → matches "Image", "image", or "IMAGE"

title: Anomalous Logon Patterns – Account Usage and Type Anomalies
id: d4e5f6a7-3434-4d8e-e9f0-a1b2c3d4e5f6
status: stable
description: |
  Detects anomalous Windows logon patterns beyond brute force.
  Covers logon type mismatches (service accounts logging interactively),
  impossible travel, off-hours auth, and cleartext logons.
tags:
  - attack.initial_access
  - attack.t1078
  - attack.lateral_movement
logsource:
  product: windows
  service: security
detection:
  service_account_interactive:
    EventID: '4624'
    LogonType:
      - '2'   # Interactive console
      - '10'  # RDP
    TargetUserName|contains|any:
      - 'svc_'
      - 'service'
      - 'sql'
  cleartext_logon:
    EventID: '4624'
    LogonType: '8'  # NetworkCleartext (unencrypted)
  condition: service_account_interactive or cleartext_logon
level: high

title: Suspicious PowerShell Encoded Command Execution
id: b2c3d4e5-f6a7-8901-bcde-f12345678901
description: |
  Detects PowerShell with Base64-encoded commands—common
  attacker obfuscation technique.
tags:
  - attack.execution
  - attack.t1059.001
  - attack.defense_evasion
detection:
  powershell_process:
    Image|endswith:
      - '\powershell.exe'
      - '\pwsh.exe'
  encoded_param:
    CommandLine|contains|any:
      - ' -EncodedCommand '
      - ' -Enc '
      - ' -ec '
  condition: powershell_process and encoded_param
level: high

title: Suspicious Prefetch Entries
description: |
  Detects execution of suspicious binaries via Windows Prefetch
  artifacts (C:\Windows\Prefetch\*.pf)
detection:
  suspicious_paths:
    PrefetchFile|contains|any:
      - '\Temp\'
      - '\AppData\Roaming\'
      - '\Downloads\'
      - '.tmp.exe'
      - 'C$\'
  rare_extensions:
    ExecutableName|endswith|any:
      - '.vbs'
      - '.js'
      - '.hta'
      - '.cmd'
  condition: suspicious_paths or rare_extensions
level: medium

title: Human-readable detection name
id: UUID (unique identifier)
status: stable | experimental | deprecated
description: Multi-line explanation of what this rule detects
references:
  - https://attack.mitre.org/techniques/TXXXX/
author: Chronos-DFIR / Analyst Name
date: YYYY/MM/DD
tags:
  - attack.tactic_name
  - attack.tXXXX        # MITRE technique ID
logsource:
  category: process_creation | network_connection | file_event
  product: windows | linux | macos
  service: security | sysmon | auditd
detection:
  selection_1:
    FieldName: value
    FieldName|modifier: value
  selection_2:
    FieldName|contains|any:
      - string1
      - string2
  condition: selection_1 and not selection_2
falsepositives:
  - Known benign behavior causing false alerts
level: critical | high | medium | low
fields:
  - EventID
  - CommandLine
  - User
custom:
  mitre_tactic: "TA0002 – Execution"
  mitre_technique: "T1059 – Command and Scripting Interpreter"

# AND condition (all blocks must match)
condition: selection_1 and selection_2

# OR condition (any block matches)
condition: selection_1 or selection_2

# NOT condition (negation)
condition: selection_1 and not filter_benign

# Wildcards
condition: 1 of selection_*
condition: all of them

# Grouping (evaluated left-to-right)
condition: (selection_1 or selection_2) and not filter_whitelist

def load_sigma_rules(rules_dir: Optional[str] = None) -> list:
    """
    Walk rules/sigma directory and return all parsed YAML rules.
    Results are cached in-process after first load.
    """
    patterns = [
        os.path.join(base, "**", "*.yml"),
        os.path.join(base, "**", "*.yaml"),
    ]
    for pattern in patterns:
        for path in glob.glob(pattern, recursive=True):
            doc = yaml.safe_load(fh)
            rules.append(doc)
    return rules

def _build_field_condition(field_raw: str, values, columns: list[str]):
    """
    Parse 'Image|endswith|any' into Polars expression.
    Handles: plain field, field|modifier, field|not|modifier
    """
    parts = field_raw.split("|")
    field_name = parts[0]
    modifiers = [p.lower() for p in parts[1:]]
    
    negate = "not" in modifiers
    col_expr = _field_expr(field_name, columns)  # Case-insensitive lookup
    
    if modifier == "contains":
        expr = col_expr.str.contains(value, literal=True)
    elif modifier == "endswith":
        expr = col_expr.str.ends_with(value)
    # ... (continued for all modifiers)
    
    return (~expr) if negate else expr

def match_sigma_rules(df: pl.DataFrame, rules: list) -> list[dict]:
    """
    Evaluate all rules against DataFrame.
    Returns list of hits: {title, level, mitre_technique, matched_rows}
    """
    for rule in rules:
        detection = rule.get("detection", {})
        condition_str = detection.get("condition", "")
        
        # Build named expressions for each detection block
        named_exprs = {}
        for block_name, block_value in detection.items():
            if isinstance(block_value, dict):
                named_exprs[block_name] = _build_named_condition(block_value, columns)
        
        # Resolve condition string ("selection_1 or selection_2")
        final_expr = _parse_condition_string(condition_str, named_exprs)
        
        # Filter and count matches
        df_matched = df.filter(final_expr)
        if df_matched.height > 0:
            hits.append({...})
    return hits

# 27 forensic context columns automatically included if present
FORENSIC_CONTEXT_COLUMNS = [
    "UserName", "User", "AccountName", "TargetUserName",
    "ProcessName", "Image", "ParentImage",
    "SourceIP", "IpAddress", "ClientIP",
    "CommandLine", "ParentCommandLine",
    "Status", "LogonType", "ServiceName", ...
]

# Extract evidence: matched columns + forensic context (max 12 cols)
evidence_cols = ["_id", time_col] + matched_columns
for fc in FORENSIC_CONTEXT_COLUMNS:
    if len(evidence_cols) >= 12:
        break
    if fc in df_matched.columns:
        evidence_cols.append(fc)

sample_evidence = df_matched.head(150).select(evidence_cols).to_dicts()

detection:
  base_event:
    EventID: '4625'  # Failed logon
  timeframe: 5m
  correlation:
    type: event_count
    group-by: ['TargetUserName', 'IpAddress']
    timespan: 5m
    condition:
      gte: 10  # 10+ failed logins in 5 minutes = brute force

def _evaluate_temporal_correlation(df_matched, detection, rule):
    """
    Apply time-window grouping with threshold.
    Returns adjusted match count or None if no correlation.
    """
    timeframe = detection.get("timeframe")
    correlation = detection.get("correlation")
    
    if not timeframe and not correlation:
        return None  # No temporal logic
    
    time_col = _find_time_column(df_matched.columns)
    duration = _parse_timeframe("5m")  # "5m" -> Polars duration
    
    # Group by time window + specified fields
    windowed = (
        df_matched.sort("_sigma_ts")
        .group_by_dynamic("_sigma_ts", every=duration, group_by=resolved_groups)
        .agg(pl.len().alias("_event_count"))
    )
    
    # Count groups exceeding threshold
    hot_groups = windowed.filter(pl.col("_event_count") >= threshold)
    return hot_groups["_event_count"].sum()

def calculate_smart_risk_m4(df: pl.DataFrame, sigma_hits: list) -> dict:
    """
    Unified risk scoring: DF columns + Sigma detections
    """
    risk_score = 50  # Baseline
    
    # Sigma detection weighting
    for hit in sigma_hits:
        if hit["level"] == "critical":
            risk_score += 15
        elif hit["level"] == "high":
            risk_score += 10
        elif hit["level"] == "medium":
            risk_score += 5
    
    # Cap at 100
    return {"level": "Critical" if risk_score >= 85 else "High", ...}

// Frontend: TTP Summary Strip
const techniques = sigmaHits
  .flatMap(hit => hit.tags.filter(t => t.startsWith('attack.t')))
  .map(t => t.replace('attack.', '').toUpperCase());

// Render: CRITICAL: 3  HIGH: 12  |  T1003  T1059  T1070  T1218

<details class="sigma-rule-detail">
  <summary>🔴 HIGH: Anomalous Logon Patterns (18 matches)</summary>
  <table>
    <tr>
      <th>Time</th><th>User</th><th>IP</th><th>LogonType</th>
    </tr>
    <!-- sample_evidence rows rendered here -->
  </table>
  <button onclick="viewInGrid([...all_row_ids])">View all in Grid</button>
</details>

# BAD: Python loop (10,000x slower)
for row in df.iter_rows():
    if 'mimikatz' in row['CommandLine']:
        matches.append(row)

# GOOD: Polars vectorized (M4 ARM NEON optimized)
df_matched = df.filter(
    pl.col('CommandLine').str.contains('mimikatz', literal=True)
)

col_expr = pl.col(column_name).cast(pl.Utf8, strict=False)

title: Your Detection Name
id: $(uuidgen)  # Generate unique UUID
status: experimental
description: |
  Detailed explanation of threat behavior.
author: Your Name / Organization
date: $(date +%Y/%m/%d)
tags:
  - attack.tactic_name
  - attack.tXXXX
logsource:
  product: windows
  service: security
detection:
  selection:
    EventID: '4688'
    CommandLine|contains: 'suspicious_string'
  condition: selection
level: medium

# 1. Place YAML in rules/sigma/custom/
cp my_rule.yml rules/sigma/custom/

# 2. Validate YAML syntax
python -c "import yaml; yaml.safe_load(open('rules/sigma/custom/my_rule.yml'))"

# 3. Reload rules in Chronos (force cache refresh)
# Rules auto-reload on server restart

# 4. Test against sample dataset
# Load file in Chronos UI → Check Forensic Context modal for hits