09 - Testing Detection Rules: Unit Testing per Logica di Sicurezza

# Sigma Rule con test integrati (formato sigma-test)
# File: rules/windows/t1059_001_powershell_encoded.yml

title: PowerShell Encoded Command Execution
id: 5b4f6d89-1234-4321-ab12-fedcba987654
status: stable
description: >
  Rileva esecuzione PowerShell con parametri di encoding, frequentemente
  usati da malware per offuscare payload.
references:
  - https://attack.mitre.org/techniques/T1059/001/
author: Detection Team
date: 2025-01-15
tags:
  - attack.execution
  - attack.t1059.001
logsource:
  category: process_creation
  product: windows
detection:
  selection:
    Image|endswith:
      - '\powershell.exe'
      - '\pwsh.exe'
    CommandLine|contains:
      - ' -enc '
      - ' -EncodedCommand '
      - ' -ec '
  condition: selection
falsepositives:
  - Software legittimo enterprise che usa PowerShell con encoding
  - Script di deployment automatizzati
level: medium

# Test cases (formato sigma-test)
tests:
  - name: "TP: PowerShell con -EncodedCommand"
    should_match: true
    event:
      Image: 'C:\Windows\System32\WindowsPowerShell\v1.0\powershell.exe'
      CommandLine: 'powershell.exe -EncodedCommand SQBFAFgAKABOAGUAdAAgAC4AIAAuACkA'
      ParentImage: 'C:\Windows\System32\cmd.exe'

  - name: "TP: PowerShell con -enc (shorthand)"
    should_match: true
    event:
      Image: 'C:\Windows\System32\WindowsPowerShell\v1.0\powershell.exe'
      CommandLine: 'powershell.exe -enc SQBFAFgA'
      ParentImage: 'C:\Windows\Explorer.exe'

  - name: "TP: pwsh (PowerShell Core) con encoded"
    should_match: true
    event:
      Image: 'C:\Program Files\PowerShell\7\pwsh.exe'
      CommandLine: 'pwsh -EncodedCommand SQBFAFgA'
      ParentImage: 'C:\Windows\System32\services.exe'

  - name: "FP: PowerShell normale senza encoding"
    should_match: false
    event:
      Image: 'C:\Windows\System32\WindowsPowerShell\v1.0\powershell.exe'
      CommandLine: 'powershell.exe -ExecutionPolicy Bypass -File C:\scripts\deploy.ps1'
      ParentImage: 'C:\Windows\System32\svchost.exe'

  - name: "FP: PowerShell con parametro simile ma non encoding"
    should_match: false
    event:
      Image: 'C:\Windows\System32\WindowsPowerShell\v1.0\powershell.exe'
      CommandLine: 'powershell.exe Get-Content C:\scripts\encrypted-backup.zip'
      ParentImage: 'C:\Windows\System32\svchost.exe'

# Esecuzione sigma-test
# Installa: go install github.com/bradleyjkemp/sigma-test@latest

# Test singola regola
sigma-test rules/windows/t1059_001_powershell_encoded.yml

# Test di tutte le regole in una directory
sigma-test rules/windows/

# Output example:
# PASS rules/windows/t1059_001_powershell_encoded.yml
#   TP: PowerShell con -EncodedCommand ... PASS
#   TP: PowerShell con -enc (shorthand) ... PASS
#   TP: pwsh (PowerShell Core) con encoded ... PASS
#   FP: PowerShell normale senza encoding ... PASS
#   FP: PowerShell con parametro simile ... PASS

# Framework pytest per Detection Rules
# File: tests/test_detection_rules.py

import pytest
import yaml
from pathlib import Path
from sigma.rule import SigmaRule
from sigma.backends.splunk import SplunkBackend
from sigma.backends.elasticsearch import ElasticsearchQueryStringBackend
import re

RULES_DIR = Path("rules")
TESTS_DIR = Path("tests/test_data")

def load_all_rules() -> list[tuple[str, str]]:
    """Carica tutte le regole Sigma dalla directory rules/."""
    rules = []
    for rule_file in RULES_DIR.glob("**/*.yml"):
        content = rule_file.read_text()
        rules.append((str(rule_file), content))
    return rules

class SigmaRuleSimulator:
    """Simula il matching di una regola Sigma su eventi."""

    def __init__(self, rule_yaml: str):
        self.rule_dict = yaml.safe_load(rule_yaml)
        self.detection = self.rule_dict.get('detection', {})

    def matches(self, event: dict) -> bool:
        """Verifica se l'evento matcha la regola (simulazione semplificata)."""
        condition = self.detection.get('condition', '')
        selectors = {k: v for k, v in self.detection.items() if k != 'condition'}

        # Valuta ogni selettore
        selector_results = {}
        for selector_name, criteria in selectors.items():
            if selector_name.startswith('filter'):
                selector_results[selector_name] = self._eval_criteria(event, criteria)
            else:
                selector_results[selector_name] = self._eval_criteria(event, criteria)

        # Valuta la condition
        return self._eval_condition(condition, selector_results)

    def _eval_criteria(self, event: dict, criteria) -> bool:
        """Valuta un selettore contro l'evento."""
        if not isinstance(criteria, dict):
            return False

        for field, values in criteria.items():
            # Estrai nome campo e modifier
            parts = field.split('|')
            field_name = parts[0]
            modifier = parts[1] if len(parts) > 1 else 'exact'

            event_value = str(event.get(field_name, ''))

            # Controlla lista di valori (OR implicito)
            if isinstance(values, list):
                if not any(self._apply_modifier(event_value, str(v), modifier)
                           for v in values):
                    return False
            else:
                if not self._apply_modifier(event_value, str(values), modifier):
                    return False

        return True

    def _apply_modifier(self, event_val: str,
                         pattern: str, modifier: str) -> bool:
        """Applica il modifier Sigma."""
        ev = event_val.lower()
        pat = pattern.lower().replace('*', '')

        if modifier == 'contains':
            return pat in ev
        elif modifier == 'endswith':
            return ev.endswith(pat)
        elif modifier == 'startswith':
            return ev.startswith(pat)
        elif modifier == 're':
            return bool(re.search(pattern, event_val, re.IGNORECASE))
        else:  # exact
            return ev == pat

    def _eval_condition(self, condition: str,
                         results: dict[str, bool]) -> bool:
        """Valuta la condition Sigma (logica base)."""
        # Gestisce condizioni semplici comuni
        condition = condition.strip()

        # "selection" semplice
        if condition in results:
            return results[condition]

        # "selection and not filter"
        if ' and not ' in condition:
            parts = condition.split(' and not ')
            left = self._eval_condition(parts[0].strip(), results)
            right = self._eval_condition(parts[1].strip(), results)
            return left and not right

        # "selection or selection2"
        if ' or ' in condition:
            parts = condition.split(' or ')
            return any(self._eval_condition(p.strip(), results) for p in parts)

        # "selection and selection2"
        if ' and ' in condition:
            parts = condition.split(' and ')
            return all(self._eval_condition(p.strip(), results) for p in parts)

        # "not selection"
        if condition.startswith('not '):
            inner = condition[4:].strip()
            return not self._eval_condition(inner, results)

        return results.get(condition, False)

# ===== TEST CLASSES =====

class TestSigmaRuleSyntax:
    """Test sintattici: tutte le regole devono essere YAML valido."""

    @pytest.mark.parametrize("rule_path,rule_content", load_all_rules())
    def test_valid_yaml(self, rule_path: str, rule_content: str):
        """Ogni regola deve essere YAML valido e parsabile."""
        try:
            rule_dict = yaml.safe_load(rule_content)
            assert rule_dict is not None, f"YAML vuoto: {rule_path}"
        except yaml.YAMLError as e:
            pytest.fail(f"YAML invalido in {rule_path}: {e}")

    @pytest.mark.parametrize("rule_path,rule_content", load_all_rules())
    def test_required_fields(self, rule_path: str, rule_content: str):
        """Ogni regola deve avere i campi obbligatori."""
        rule_dict = yaml.safe_load(rule_content)
        required = ['title', 'description', 'logsource', 'detection']

        for field in required:
            assert field in rule_dict, \
                f"Campo '{field}' mancante in {rule_path}"

    @pytest.mark.parametrize("rule_path,rule_content", load_all_rules())
    def test_detection_has_condition(self, rule_path: str, rule_content: str):
        """Ogni regola deve avere una 'condition' in detection."""
        rule_dict = yaml.safe_load(rule_content)
        detection = rule_dict.get('detection', {})
        assert 'condition' in detection, \
            f"'condition' mancante in detection di {rule_path}"

    @pytest.mark.parametrize("rule_path,rule_content", load_all_rules())
    def test_valid_level(self, rule_path: str, rule_content: str):
        """Il level deve essere uno dei valori standard."""
        rule_dict = yaml.safe_load(rule_content)
        valid_levels = {'informational', 'low', 'medium', 'high', 'critical'}
        level = rule_dict.get('level', '')
        if level:
            assert level in valid_levels, \
                f"Level '{level}' non valido in {rule_path}"

    @pytest.mark.parametrize("rule_path,rule_content", load_all_rules())
    def test_pysigma_parseable(self, rule_path: str, rule_content: str):
        """Ogni regola deve essere parsabile da pySigma."""
        from sigma.exceptions import SigmaError
        try:
            SigmaRule.from_yaml(rule_content)
        except SigmaError as e:
            pytest.fail(f"pySigma non riesce a parsare {rule_path}: {e}")

# Generatore di Log Sintetici
# File: tests/log_generator.py

from dataclasses import dataclass
from datetime import datetime, timedelta
import random
import string
import uuid

@dataclass
class WindowsProcessEvent:
    """Evento di process_creation Windows (Sysmon Event ID 1)."""
    EventID: str = "1"
    ComputerName: str = "WORKSTATION01"
    User: str = "DOMAIN\\user"
    Image: str = "C:\\Windows\\System32\\cmd.exe"
    CommandLine: str = "cmd.exe"
    ParentImage: str = "C:\\Windows\\Explorer.exe"
    ParentCommandLine: str = "explorer.exe"
    ProcessId: str = "1234"
    ParentProcessId: str = "5678"
    MD5: str = ""
    SHA256: str = ""
    Hashes: str = ""
    UtcTime: str = ""

    def __post_init__(self):
        if not self.UtcTime:
            self.UtcTime = datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S.%f")
        if not self.MD5:
            self.MD5 = ''.join(random.choices(string.hexdigits, k=32)).upper()

    def to_dict(self) -> dict:
        return {k: v for k, v in self.__dict__.items()}

class SyntheticLogGenerator:
    """Genera log sintetici per scenari di testing specifici."""

    # Template per tecniche ATT&CK comuni
    TEMPLATES = {
        'T1059.001_encoded': [
            # True Positives
            {
                'should_match': True,
                'name': 'PS encoded via cmd',
                'event': WindowsProcessEvent(
                    Image='C:\\Windows\\System32\\WindowsPowerShell\\v1.0\\powershell.exe',
                    CommandLine='powershell.exe -EncodedCommand SQBFAFgAKABOAGUAdAAgAC4AIAAuACkA',
                    ParentImage='C:\\Windows\\System32\\cmd.exe'
                )
            },
            {
                'should_match': True,
                'name': 'PS enc shorthand',
                'event': WindowsProcessEvent(
                    Image='C:\\Windows\\System32\\WindowsPowerShell\\v1.0\\powershell.exe',
                    CommandLine='powershell -ec SQBFAFgA',
                    ParentImage='C:\\Windows\\Explorer.exe'
                )
            },
            # False Positives
            {
                'should_match': False,
                'name': 'PS script normale',
                'event': WindowsProcessEvent(
                    Image='C:\\Windows\\System32\\WindowsPowerShell\\v1.0\\powershell.exe',
                    CommandLine='powershell.exe -ExecutionPolicy Bypass -File deploy.ps1',
                    ParentImage='C:\\Windows\\System32\\svchost.exe'
                )
            },
            {
                'should_match': False,
                'name': 'PS word encrypted (FP trap)',
                'event': WindowsProcessEvent(
                    Image='C:\\Windows\\System32\\WindowsPowerShell\\v1.0\\powershell.exe',
                    CommandLine='powershell.exe Get-Content C:\\backup\\encrypted.zip',
                    ParentImage='C:\\Windows\\System32\\TaskScheduler.exe'
                )
            },
        ],

        'T1003.001_lsass_dump': [
            {
                'should_match': True,
                'name': 'Procdump lsass',
                'event': WindowsProcessEvent(
                    Image='C:\\Tools\\procdump.exe',
                    CommandLine='procdump.exe -ma lsass.exe lsass_dump.dmp',
                    ParentImage='C:\\Windows\\System32\\cmd.exe'
                )
            },
            {
                'should_match': True,
                'name': 'Task Manager lsass dump',
                'event': WindowsProcessEvent(
                    Image='C:\\Windows\\System32\\taskmgr.exe',
                    CommandLine='taskmgr.exe',
                    ParentImage='C:\\Windows\\Explorer.exe'
                )
            },
            {
                'should_match': False,
                'name': 'Normal lsass activity',
                'event': WindowsProcessEvent(
                    Image='C:\\Windows\\System32\\lsass.exe',
                    CommandLine='lsass.exe',
                    ParentImage='C:\\Windows\\System32\\wininit.exe'
                )
            },
        ],

        'T1053.005_scheduled_task': [
            {
                'should_match': True,
                'name': 'schtasks create con cmd',
                'event': WindowsProcessEvent(
                    Image='C:\\Windows\\System32\\schtasks.exe',
                    CommandLine='schtasks.exe /create /tn "Windows Update" /tr "cmd.exe /c evil.exe" /sc daily',
                    ParentImage='C:\\Windows\\System32\\cmd.exe'
                )
            },
            {
                'should_match': False,
                'name': 'schtasks query legittimo',
                'event': WindowsProcessEvent(
                    Image='C:\\Windows\\System32\\schtasks.exe',
                    CommandLine='schtasks.exe /query /fo LIST',
                    ParentImage='C:\\Windows\\System32\\svchost.exe'
                )
            },
        ]
    }

    def get_test_cases(self, technique_id: str) -> list[dict]:
        """Restituisce i test case per una tecnica ATT&CK."""
        key = technique_id.replace('.', '_').replace('T', 'T', 1)
        # Cerca per prefix (es. 'T1059_001')
        for template_key, cases in self.TEMPLATES.items():
            if template_key.startswith(key.replace('T', 'T', 1)):
                return [{
                    'should_match': c['should_match'],
                    'name': c['name'],
                    'event': c['event'].to_dict()
                } for c in cases]
        return []

    def generate_random_events(self, count: int = 100) -> list[dict]:
        """Genera eventi casuali per stress testing (tutti FP)."""
        events = []
        legitimate_processes = [
            'C:\\Windows\\System32\\cmd.exe',
            'C:\\Windows\\System32\\svchost.exe',
            'C:\\Program Files\\Microsoft Office\\root\\Office16\\EXCEL.EXE',
            'C:\\Windows\\System32\\notepad.exe',
            'C:\\Windows\\explorer.exe',
        ]
        for _ in range(count):
            events.append(WindowsProcessEvent(
                Image=random.choice(legitimate_processes),
                CommandLine=f"process.exe {''.join(random.choices(string.ascii_lowercase, k=20))}",
                ParentImage=random.choice(legitimate_processes),
                ComputerName=f"WS-{random.randint(1000, 9999)}"
            ).to_dict())
        return events

# Test completi con pytest
# File: tests/test_rule_logic.py

class TestRuleLogicWithSyntheticLogs:
    """Test della logica di detection con log sintetici."""

    @pytest.fixture
    def generator(self) -> SyntheticLogGenerator:
        return SyntheticLogGenerator()

    def test_powershell_encoded_true_positives(self, generator):
        """Verifica che tutti i TP vengano rilevati."""
        rule_content = Path(
            "rules/windows/t1059_001_powershell_encoded.yml"
        ).read_text()
        simulator = SigmaRuleSimulator(rule_content)

        test_cases = generator.get_test_cases('T1059.001')
        tp_cases = [tc for tc in test_cases if tc['should_match']]

        assert len(tp_cases) > 0, "Nessun test case TP trovato"

        for tc in tp_cases:
            result = simulator.matches(tc['event'])
            assert result, \
                f"FALSE NEGATIVE: '{tc['name']}' non ha triggerato la regola\n" \
                f"Event: {tc['event']}"

    def test_powershell_encoded_false_positives(self, generator):
        """Verifica che gli eventi legittimi NON vengano rilevati."""
        rule_content = Path(
            "rules/windows/t1059_001_powershell_encoded.yml"
        ).read_text()
        simulator = SigmaRuleSimulator(rule_content)

        test_cases = generator.get_test_cases('T1059.001')
        fp_cases = [tc for tc in test_cases if not tc['should_match']]

        for tc in fp_cases:
            result = simulator.matches(tc['event'])
            assert not result, \
                f"FALSE POSITIVE: '{tc['name']}' ha triggerato la regola inaspettatamente\n" \
                f"Event: {tc['event']}"

    def test_stress_no_false_positives(self, generator):
        """Stress test: 100 eventi casuali non devono triggherare."""
        rule_content = Path(
            "rules/windows/t1059_001_powershell_encoded.yml"
        ).read_text()
        simulator = SigmaRuleSimulator(rule_content)

        random_events = generator.generate_random_events(100)
        fp_count = sum(1 for ev in random_events if simulator.matches(ev))

        # Accettiamo max 2% di false positive su eventi casuali
        fp_rate = fp_count / len(random_events)
        assert fp_rate <= 0.02, \
            f"Tasso FP troppo alto: {fp_rate:.1%} ({fp_count}/{len(random_events)})"

class TestRuleCoverage:
    """Test di coverage: ogni tecnica ATT&CK deve avere almeno una regola."""

    CRITICAL_TECHNIQUES = [
        'T1059.001',  # PowerShell
        'T1003.001',  # LSASS Dump
        'T1055',      # Process Injection
        'T1053.005',  # Scheduled Task
        'T1078',      # Valid Accounts
        'T1021.002',  # SMB/Windows Admin Shares
        'T1562.001',  # Disable Security Tools
        'T1070.004',  # File Deletion
    ]

    def test_critical_techniques_have_rules(self):
        """Verifica che tutte le tecniche critiche abbiano almeno una regola."""
        # Carica tutti i tag dalle regole
        covered_techniques = set()
        for rule_file in RULES_DIR.glob("**/*.yml"):
            content = yaml.safe_load(rule_file.read_text())
            tags = content.get('tags', [])
            for tag in tags:
                if tag.startswith('attack.t'):
                    # Converti tag.t1059.001 -> T1059.001
                    technique = tag.replace('attack.', '').upper()
                    covered_techniques.add(technique)

        uncovered = [t for t in self.CRITICAL_TECHNIQUES
                     if t not in covered_techniques]

        assert not uncovered, \
            f"Tecniche critiche senza copertura: {uncovered}"

    def test_coverage_report(self):
        """Genera report di coverage ATT&CK (informativo, non fail)."""
        covered = set()
        for rule_file in RULES_DIR.glob("**/*.yml"):
            content = yaml.safe_load(rule_file.read_text())
            for tag in content.get('tags', []):
                if tag.startswith('attack.t'):
                    covered.add(tag.replace('attack.', '').upper())

        print(f"\n=== ATT&CK Coverage Report ===")
        print(f"Tecniche coperte: {len(covered)}")
        print(f"Tecniche critiche coperte: "
              f"{len([t for t in self.CRITICAL_TECHNIQUES if t in covered])}"
              f"/{len(self.CRITICAL_TECHNIQUES)}")
        print("Dettaglio critico:")
        for tech in self.CRITICAL_TECHNIQUES:
            status = "OK" if tech in covered else "MANCANTE"
            print(f"  {tech}: {status}")

# GitHub Actions CI/CD per Detection Rules
# File: .github/workflows/test-detection-rules.yml

"""
name: Detection Rules CI

on:
  pull_request:
    paths:
      - 'rules/**'
      - 'tests/**'
  push:
    branches: [main]

jobs:
  syntax-validation:
    name: Syntax Validation
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'
          cache: pip

      - name: Install dependencies
        run: |
          pip install pySigma pySigma-backend-splunk pyyaml pytest pytest-cov

      - name: Run syntax tests
        run: pytest tests/test_detection_rules.py::TestSigmaRuleSyntax -v

  logic-testing:
    name: Logic Testing
    runs-on: ubuntu-latest
    needs: syntax-validation
    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'
          cache: pip

      - name: Install dependencies
        run: pip install pySigma pyyaml pytest pytest-cov

      - name: Run logic tests with coverage
        run: |
          pytest tests/test_rule_logic.py -v --tb=short \
            --cov=rules --cov-report=xml --cov-report=term

      - name: Coverage gate
        run: |
          # Verifica che almeno l'80% delle regole abbia test
          python -c "
          import xml.etree.ElementTree as ET
          tree = ET.parse('coverage.xml')
          root = tree.getroot()
          rate = float(root.attrib.get('line-rate', 0))
          print(f'Coverage: {rate:.1%}')
          assert rate >= 0.8, f'Coverage {rate:.1%} sotto la soglia 80%'
          "

  sigma-test:
    name: sigma-test Inline Tests
    runs-on: ubuntu-latest
    needs: syntax-validation
    steps:
      - uses: actions/checkout@v4

      - name: Install sigma-test
        run: go install github.com/bradleyjkemp/sigma-test@latest

      - name: Run sigma-test on all rules
        run: sigma-test rules/ --exit-on-failure

  coverage-check:
    name: ATT&CK Coverage Check
    runs-on: ubuntu-latest
    needs: [syntax-validation, logic-testing]
    steps:
      - uses: actions/checkout@v4

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

      - name: Check critical technique coverage
        run: |
          pytest tests/test_rule_logic.py::TestRuleCoverage -v

  notify-on-failure:
    name: Notify on Failure
    runs-on: ubuntu-latest
    needs: [syntax-validation, logic-testing, sigma-test, coverage-check]
    if: failure()
    steps:
      - name: Notify Slack
        run: |
          curl -X POST ${{ secrets.SLACK_WEBHOOK }} \
            -H 'Content-type: application/json' \
            --data '{"text": "Detection Rule CI fallita! PR: ${{ github.event.pull_request.html_url }}"}'
"""