01 - Sigma Rules: Logica di Detection Universale e Conversione SIEM

title: Mimikatz Detection via Windows Security Events
id: 0d82df6e-0e4e-4fbb-a12b-6e7a00a9c7c3
status: stable
description: Detects mimikatz usage patterns via LSASS access events
  and command-line indicators
references:
  - https://github.com/gentilkiwi/mimikatz
  - https://attack.mitre.org/techniques/T1003/001/
author: Federico Calo
date: 2026/03/01
modified: 2026/03/09
tags:
  - attack.credential_access
  - attack.t1003.001
  - attack.defense_evasion
logsource:
  category: process_access
  product: windows
detection:
  selection_lsass:
    TargetImage|endswith: '\lsass.exe'
    GrantedAccess|contains:
      - '0x1010'
      - '0x1038'
      - '0x1400'
      - '0x143a'
  selection_mimikatz_cli:
    CommandLine|contains|all:
      - 'sekurlsa'
      - 'logonpasswords'
  filter_legitimate:
    SourceImage|startswith:
      - 'C:\Windows\System32\'
      - 'C:\Program Files\Windows Defender\'
  condition: (selection_lsass and not filter_legitimate) or selection_mimikatz_cli
falsepositives:
  - Legitimate security tools performing LSASS inspection
  - Antivirus and EDR solutions
level: high

# Esempio 1: Windows Security Events
logsource:
  category: process_creation
  product: windows

# Esempio 2: Linux Authentication
logsource:
  category: process_creation
  product: linux

# Esempio 3: Web Server Logs
logsource:
  category: webserver
  # nessun product specifico: generico

# Esempio 4: Log specifico di un servizio
logsource:
  product: windows
  service: security

# Esempio 5: Log di rete
logsource:
  category: network_connection
  product: windows

# Esempio 6: Cloud Trail
logsource:
  product: aws
  service: cloudtrail

detection:
  # Selection con match esatto su campo singolo
  selection_exact:
    EventID: 4625

  # Selection con lista di valori (OR implicito)
  selection_multi_value:
    CommandLine|contains:
      - 'net user'
      - 'net localgroup'
      - 'whoami'

  # Selection con match su tutti i valori (AND implicito con |all)
  selection_and_all:
    CommandLine|contains|all:
      - '-enc'
      - 'powershell'

  # Selection con regex
  selection_regex:
    CommandLine|re: '(?i)(invoke-mimikatz|sekurlsa::)'

  # Selection con wildcard
  selection_wildcard:
    TargetFilename|startswith: 'C:\Users\'
    TargetFilename|endswith: '.exe'

  # Filter per ridurre i falsi positivi
  filter_system_processes:
    ParentImage|startswith: 'C:\Windows\System32\'

  # Condition: logica booleana
  condition: selection_exact and (
    selection_multi_value or selection_and_all
  ) and not filter_system_processes

# Rilevare brute force: più di 10 login falliti in 60 secondi
title: Windows Brute Force Login Attempt
logsource:
  product: windows
  service: security
detection:
  selection:
    EventID: 4625
    LogonType: 3
  condition: selection | count() > 10
timeframe: 60s
falsepositives:
  - Users with expired passwords
level: medium

---

# Rilevare data exfiltration: upload anomalo per destinazione
title: Potential Data Exfiltration by Volume
logsource:
  category: network_connection
  product: windows
detection:
  selection:
    Initiated: 'true'
    DestinationPort:
      - 443
      - 80
  condition: selection | count(DestinationHostname) by SourceIp > 50
timeframe: 1h
level: high

---

# Port scan detection: molte porte diverse dallo stesso sorgente
title: Network Port Scan Detection
logsource:
  category: network_connection
detection:
  selection:
    EventID: 5156
  condition: selection | count(DestinationPort) by SourceAddress > 30
timeframe: 10m
level: medium

# Installazione
pip install sigma-cli
pip install pySigma-backend-splunk
pip install pySigma-backend-elasticsearch
pip install pySigma-backend-microsoft365defender

# Verifica backend disponibili
sigma list backends

# Conversione base: Sigma -> Splunk SPL
sigma convert -t splunk -p splunk_windows rules/mimikatz.yml

# Output atteso:
# (source="WinEventLog:Microsoft-Windows-Sysmon/Operational"
#  EventCode=10 TargetImage="*\\lsass.exe"
#  GrantedAccess IN ("0x1010", "0x1038", "0x1400", "0x143a"))
# NOT (SourceImage="C:\\Windows\\System32\\*"
#      OR SourceImage="C:\\Program Files\\Windows Defender\\*")

# Conversione con pipeline personalizzata
sigma convert \
  -t splunk \
  -p splunk_windows \
  -p my_custom_field_mapping.yml \
  rules/

# Conversione per Elasticsearch con ECS
sigma convert \
  -t elasticsearch \
  -p ecs_windows \
  -f lucene \
  rules/credential_access/

# Conversione per Microsoft Sentinel (KQL)
sigma convert \
  -t microsoft365defender \
  -p microsoft365defender \
  rules/lateral_movement/

# Output in formato JSON per automazione
sigma convert \
  -t splunk \
  -p splunk_windows \
  --format-output json \
  rules/ > converted_rules.json

# custom_pipeline.yml
name: Custom Windows Pipeline
priority: 20
transformations:
  # Rinominare campo Sigma -> campo reale nel SIEM
  - id: field_mapping_process
    type: field_name_mapping
    mapping:
      CommandLine: process.command_line
      Image: process.executable
      ParentImage: process.parent.executable
      User: user.name
      TargetImage: target.process.executable
      GrantedAccess: target.process.granted_access
    rule_conditions:
      - type: logsource
        category: process_creation

  # Aggiungere condizioni all'index di destinazione
  - id: index_condition
    type: detection_item_keyword
    keyword: 'index=windows_events'
    rule_conditions:
      - type: logsource
        product: windows

  # Sostituire valori nei campi
  - id: winlog_channel_map
    type: field_value_mapping
    field_name: winlog.channel
    mapping:
      'Security': 'security'
      'System': 'system'
      'Application': 'application'

title: Pass-the-Hash Attack Pattern
id: 6571d552-4c16-4f50-b5f6-11ae9d1b0a38
status: stable
description: Detects Pass-the-Hash attacks via anomalous NTLM authentication
tags:
  - attack.lateral_movement
  - attack.t1550.002
logsource:
  product: windows
  service: security
detection:
  selection:
    EventID: 4624
    LogonType: 3
    AuthenticationPackageName: NTLM
    KeyLength: 0
  filter_local:
    IpAddress:
      - '127.0.0.1'
      - '::1'
  filter_machine:
    SubjectUserName|endswith: '$'
  condition: selection and not filter_local and not filter_machine
falsepositives:
  - NTLMv1 authentication in legacy environments
level: high

title: Suspicious Scheduled Task Creation via CLI
id: 92a7b4f1-e6c9-4e1d-b7df-3c8e9a0b5d17
status: stable
description: Detects creation of scheduled tasks via command line
tags:
  - attack.persistence
  - attack.t1053.005
logsource:
  category: process_creation
  product: windows
detection:
  selection_schtasks:
    Image|endswith: '\schtasks.exe'
    CommandLine|contains|all:
      - '/create'
      - '/tr'
  selection_suspicious_location:
    CommandLine|contains:
      - 'C:\Users\Public\'
      - 'C:\ProgramData\'
      - '%TEMP%'
      - 'AppData\Local\Temp'
  filter_admin_tools:
    ParentImage|startswith:
      - 'C:\Windows\System32\'
      - 'C:\Program Files\Microsoft Deployment Toolkit\'
  condition: selection_schtasks and selection_suspicious_location
    and not filter_admin_tools
falsepositives:
  - Software installers creating scheduled maintenance tasks
level: high

title: Potential DNS Tunneling via High Subdomain Count
id: b7c3a9f1-d4e5-4a7b-9f2c-8d0e1b3c5a7f
status: experimental
description: Detects potential DNS tunneling via unusually long DNS queries
  or high number of unique subdomains
tags:
  - attack.exfiltration
  - attack.t1048.001
  - attack.command_and_control
  - attack.t1071.004
logsource:
  category: dns
detection:
  selection_long_query:
    QueryName|re: '^([a-zA-Z0-9]{32,}\.)+'
  selection_high_entropy:
    QueryName|re: '^[a-zA-Z0-9]{20,}\.[a-zA-Z0-9]{20,}\.'
  condition: selection_long_query or selection_high_entropy
  timeframe: 5m
falsepositives:
  - CDN providers with hashed subdomains
  - Legitimate DynamicDNS services
level: medium

sigma-rules/
├── .github/
│   └── workflows/
│       ├── validate.yml          # Lint e validazione YAML
│       └── convert-deploy.yml    # Conversione e deploy
├── rules/
│   ├── windows/
│   │   ├── process_creation/
│   │   ├── security/
│   │   └── sysmon/
│   ├── linux/
│   │   ├── auth/
│   │   └── process/
│   ├── cloud/
│   │   ├── aws/
│   │   ├── azure/
│   │   └── gcp/
│   └── network/
│       ├── dns/
│       └── firewall/
├── pipelines/
│   ├── splunk_custom.yml
│   ├── elastic_ecs.yml
│   └── sentinel_kql.yml
├── tests/
│   ├── positive/               # Log che DEVONO triggherare
│   └── negative/               # Log che NON devono triggherare
└── scripts/
    ├── validate_rules.py
    ├── convert_all.py
    └── deploy_to_siem.py

#!/usr/bin/env python3
"""
Sigma Rule Validator and Converter
Valida la sintassi YAML e converte le regole per tutti i SIEM target.
"""
import os
import sys
import json
import yaml
import uuid
import subprocess
from pathlib import Path
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class ConversionResult:
    rule_path: str
    success: bool
    target: str
    output: str = ""
    error: str = ""

@dataclass
class ValidationResult:
    rule_path: str
    valid: bool
    errors: list[str] = field(default_factory=list)
    warnings: list[str] = field(default_factory=list)

REQUIRED_FIELDS = ['title', 'id', 'status', 'description', 'logsource', 'detection', 'level']
VALID_STATUSES = ['stable', 'test', 'experimental', 'deprecated', 'unsupported']
VALID_LEVELS = ['informational', 'low', 'medium', 'high', 'critical']

SIEM_TARGETS = [
    {"name": "splunk", "pipeline": "splunk_windows", "flag": "-t splunk"},
    {"name": "elastic", "pipeline": "ecs_windows", "flag": "-t elasticsearch"},
    {"name": "sentinel", "pipeline": "microsoft365defender",
     "flag": "-t microsoft365defender"},
]

def validate_sigma_rule(rule_path: Path) -> ValidationResult:
    """Valida una singola regola Sigma."""
    result = ValidationResult(rule_path=str(rule_path), valid=True)

    try:
        with open(rule_path, 'r', encoding='utf-8') as f:
            rule = yaml.safe_load(f)
    except yaml.YAMLError as e:
        result.valid = False
        result.errors.append(f"YAML parse error: {e}")
        return result

    # Check campi obbligatori
    for field_name in REQUIRED_FIELDS:
        if field_name not in rule:
            result.valid = False
            result.errors.append(f"Missing required field: {field_name}")

    # Validate UUID
    if 'id' in rule:
        try:
            uuid.UUID(str(rule['id']))
        except ValueError:
            result.errors.append(f"Invalid UUID format: {rule['id']}")
            result.valid = False

    # Validate status
    if 'status' in rule and rule['status'] not in VALID_STATUSES:
        result.warnings.append(
            f"Unknown status: {rule['status']}. "
            f"Valid: {VALID_STATUSES}"
        )

    # Validate level
    if 'level' in rule and rule['level'] not in VALID_LEVELS:
        result.valid = False
        result.errors.append(
            f"Invalid level: {rule['level']}. Valid: {VALID_LEVELS}"
        )

    # Check logsource ha almeno category o service
    if 'logsource' in rule:
        logsource = rule['logsource']
        if not any(k in logsource for k in ['category', 'service', 'product']):
            result.errors.append("logsource must have at least one of: category, service, product")
            result.valid = False

    # Check detection ha condition
    if 'detection' in rule and 'condition' not in rule['detection']:
        result.errors.append("detection block missing 'condition' field")
        result.valid = False

    # Warn su falsepositives mancanti per regole stable/high
    if rule.get('status') == 'stable' and rule.get('level') in ['high', 'critical']:
        if 'falsepositives' not in rule:
            result.warnings.append("Stable high-severity rule missing falsepositives documentation")

    return result

def convert_rule(rule_path: Path, target_config: dict) -> ConversionResult:
    """Converte una regola Sigma per un SIEM target."""
    cmd = [
        'sigma', 'convert',
        '-t', target_config['name'],
        '-p', target_config['pipeline'],
        str(rule_path)
    ]
    try:
        proc = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
        if proc.returncode == 0:
            return ConversionResult(
                rule_path=str(rule_path),
                success=True,
                target=target_config['name'],
                output=proc.stdout.strip()
            )
        else:
            return ConversionResult(
                rule_path=str(rule_path),
                success=False,
                target=target_config['name'],
                error=proc.stderr.strip()
            )
    except subprocess.TimeoutExpired:
        return ConversionResult(
            rule_path=str(rule_path),
            success=False,
            target=target_config['name'],
            error="Conversion timeout"
        )

def process_repository(rules_dir: str) -> dict:
    """Processa l'intero repository di regole Sigma."""
    rules_path = Path(rules_dir)
    results = {"validation": [], "conversion": {}, "summary": {}}

    rule_files = list(rules_path.rglob("*.yml"))
    print(f"Found {len(rule_files)} rules")

    valid_count = 0
    invalid_count = 0

    for rule_file in rule_files:
        val_result = validate_sigma_rule(rule_file)
        results["validation"].append({
            "path": str(rule_file.relative_to(rules_path)),
            "valid": val_result.valid,
            "errors": val_result.errors,
            "warnings": val_result.warnings
        })

        if val_result.valid:
            valid_count += 1
            for target in SIEM_TARGETS:
                if target['name'] not in results["conversion"]:
                    results["conversion"][target['name']] = []
                conv = convert_rule(rule_file, target)
                results["conversion"][target['name']].append({
                    "path": str(rule_file.relative_to(rules_path)),
                    "success": conv.success,
                    "output": conv.output if conv.success else None,
                    "error": conv.error if not conv.success else None
                })
        else:
            invalid_count += 1
            print(f"INVALID: {rule_file.name}")
            for err in val_result.errors:
                print(f"  ERROR: {err}")

    results["summary"] = {
        "total": len(rule_files),
        "valid": valid_count,
        "invalid": invalid_count,
        "conversion_rate": valid_count / len(rule_files) * 100 if rule_files else 0
    }

    return results

if __name__ == "__main__":
    rules_dir = sys.argv[1] if len(sys.argv) > 1 else "./rules"
    results = process_repository(rules_dir)

    output_file = "sigma_report.json"
    with open(output_file, 'w') as f:
        json.dump(results, f, indent=2)

    summary = results["summary"]
    print(f"\n=== SIGMA VALIDATION REPORT ===")
    print(f"Total rules: {summary['total']}")
    print(f"Valid: {summary['valid']} ({summary['conversion_rate']:.1f}%)")
    print(f"Invalid: {summary['invalid']}")
    print(f"Full report: {output_file}")

    sys.exit(0 if summary['invalid'] == 0 else 1)

# .github/workflows/sigma-pipeline.yml
name: Sigma Rules CI/CD

on:
  push:
    paths:
      - 'rules/**/*.yml'
      - 'pipelines/**/*.yml'
  pull_request:
    paths:
      - 'rules/**/*.yml'

jobs:
  validate:
    name: Validate Sigma Rules
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'

      - name: Install dependencies
        run: |
          pip install sigma-cli pySigma-backend-splunk \
            pySigma-backend-elasticsearch \
            pySigma-backend-microsoft365defender \
            pyyaml

      - name: Run validation
        run: python scripts/validate_rules.py ./rules

      - name: Upload validation report
        if: always()
        uses: actions/upload-artifact@v4
        with:
          name: sigma-validation-report
          path: sigma_report.json

  convert-and-deploy:
    name: Convert and Deploy Rules
    needs: validate
    if: github.ref == 'refs/heads/main'
    runs-on: ubuntu-latest
    environment: production
    steps:
      - uses: actions/checkout@v4

      - name: Install sigma-cli
        run: pip install sigma-cli pySigma-backend-splunk

      - name: Convert rules for Splunk
        run: |
          sigma convert \
            -t splunk \
            -p splunk_windows \
            -p pipelines/splunk_custom.yml \
            --format-output savedsearches \
            rules/ > converted/splunk_rules.conf

      - name: Deploy to Splunk
        env:
          SPLUNK_HOST: ${{ secrets.SPLUNK_HOST }}
          SPLUNK_TOKEN: ${{ secrets.SPLUNK_TOKEN }}
        run: |
          python scripts/deploy_to_splunk.py \
            --rules converted/splunk_rules.conf \
            --host "$SPLUNK_HOST" \
            --token "$SPLUNK_TOKEN"

import yaml
import json
from pathlib import Path
from collections import defaultdict

def build_attck_coverage(rules_dir: str) -> dict:
    """
    Analizza il repository Sigma e genera statistiche
    di copertura per tecnica MITRE ATT&CK.
    """
    coverage = defaultdict(list)

    for rule_file in Path(rules_dir).rglob("*.yml"):
        with open(rule_file) as f:
            rule = yaml.safe_load(f)

        tags = rule.get('tags', [])
        rule_info = {
            "title": rule.get('title', ''),
            "status": rule.get('status', ''),
            "level": rule.get('level', ''),
            "file": str(rule_file)
        }

        for tag in tags:
            # Estrae solo i tag di tecnica ATT&CK: attack.t1234.001
            if tag.startswith('attack.t') and len(tag) > 9:
                technique_id = tag.replace('attack.', '').upper()
                coverage[technique_id].append(rule_info)

    # Calcola statistiche per livello di severita
    stats = {}
    for technique, rules in coverage.items():
        level_counts = defaultdict(int)
        for r in rules:
            level_counts[r['level']] += 1

        stats[technique] = {
            "rule_count": len(rules),
            "stable_count": sum(1 for r in rules if r['status'] == 'stable'),
            "levels": dict(level_counts),
            "rules": [r['title'] for r in rules]
        }

    return {
        "total_techniques_covered": len(coverage),
        "total_rules": sum(len(r) for r in coverage.values()),
        "techniques": stats
    }

coverage = build_attck_coverage("./rules")
print(f"Tecniche coperte: {coverage['total_techniques_covered']}")
print(f"Regole totali: {coverage['total_rules']}")

with open("attck_coverage.json", "w") as f:
    json.dump(coverage, f, indent=2)