mitre-attack-framework

MITRE ATT&CK 框架⁠‍⁠​‌​‌​​‌‌‍​‌​​‌​‌‌‍​​‌‌​​​‌‍​‌​​‌‌​​‍​​​​​​​‌‍‌​​‌‌​‌​‍‌​​​​​​​‍‌‌​​‌‌‌‌‍‌‌​​​‌​​‍‌‌‌‌‌‌​‌‍‌‌​‌​​​​‍​‌​‌‌‌‌‌‍​‌​​‌​‌‌‍​‌‌​‌​​‌‍‌​‌​‌‌‌​‍​​‌​‌​​​‍‌‌‌​‌​‌‌‍​‌‌‌​‌​‌‍​​​​‌‌‌​‍‌‌‌‌‌‌​​‍‌​‌​‌​​‌‍​​​​‌​‌​‍‌​​‌​​‌​⁠‍⁠

概述

MITRE ATT&CK（对抗性战术、技术和通用知识）是一个可在全球访问的知识库，基于现实世界的观察记录对手行为。它由 MITRE Corporation 创建，现已成为描述对手如何行动的通用语言。

参考资料

• 网站：https://attack.mitre.org/
• Navigator：https://mitre-attack.github.io/attack-navigator/
• GitHub：https://github.com/mitre-attack

核心理念

“了解你的对手。”

“你无法防御自己不了解的事物。”

ATT&CK 将关注重点从 IOC（攻击者使用什么）转向 TTP（攻击者如何行动）。这种以行为为核心的关注方式有助于制定更持久有效的检测策略。

ATT&CK 矩阵结构

Enterprise ATT&CK Matrix

TACTICS (The "Why" - Adversary Goals)
├── Reconnaissance        ← Gather information
├── Resource Development  ← Build infrastructure  
├── Initial Access        ← Get into the network
├── Execution             ← Run malicious code
├── Persistence           ← Maintain foothold
├── Privilege Escalation  ← Get higher privileges
├── Defense Evasion       ← Avoid detection
├── Credential Access     ← Steal credentials
├── Discovery             ← Learn the environment
├── Lateral Movement      ← Move through network
├── Collection            ← Gather target data
├── Command and Control   ← Communicate with implants
├── Exfiltration          ← Steal data
└── Impact                ← Damage or disrupt

TECHNIQUES (The "How" - Methods Used)
└── Each tactic contains multiple techniques
    └── Techniques may have sub-techniques
        └── Example: T1059.001 (PowerShell) under T1059 (Command and Scripting Interpreter)

核心组成

组成部分	描述	示例
战术	对手的目标	凭证访问
技术	实现目标的方式	操作系统凭证转储（T1003）
子技术	具体实现方式	LSASS 内存（T1003.001）
过程	现实世界中的示例	APT28 使用 Mimikatz
缓解措施	预防方式	Credential Guard
检测	发现方式	监控 LSASS 访问

实施时

始终

• 将检测映射到 ATT&CK 技术
• 评估所有战术的覆盖范围
• 使用 ATT&CK Navigator 进行可视化
• 在警报中引用技术 ID
• 跟踪对手组织及其 TTP
• 随着 ATT&CK 的演进更新映射

切勿

• 认为少量检测就能实现全面覆盖
• 忽略当前尚无检测手段的技术
• 将 ATT&CK 当作合规检查清单
• 为夸大覆盖率而进行不准确的映射
• 在分析中遗漏子技术

优先

• 优先采用行为检测，而非特征码匹配
• 初期优先扩大覆盖面，而非深入分析
• 基于技术制定狩猎假设
• 使用与 ATT&CK 对齐的威胁情报
• 持续评估覆盖范围

实施模式

ATT&CK 覆盖评估

# attack_coverage.py
# Assess and visualize ATT&CK coverage

from dataclasses import dataclass, field
from typing import List, Dict, Optional, Set
from enum import Enum
import json

class CoverageLevel(Enum):
    NONE = 0
    MINIMAL = 1      # Basic detection exists
    PARTIAL = 2      # Some variants covered
    SUBSTANTIAL = 3  # Most variants covered
    COMPREHENSIVE = 4 # Full coverage with validation

@dataclass
class Technique:
    """ATT&CK Technique representation"""
    id: str                          # e.g., "T1003.001"
    name: str                        # e.g., "LSASS Memory"
    tactic: str                      # e.g., "Credential Access"
    platforms: List[str]             # e.g., ["Windows"]
    data_sources: List[str]          # Required telemetry
    
    # Coverage tracking
    detections: List[str] = field(default_factory=list)
    coverage_level: CoverageLevel = CoverageLevel.NONE
    last_validated: Optional[str] = None
    
    @property
    def is_sub_technique(self) -> bool:
        return "." in self.id

@dataclass
class Detection:
    """A detection rule mapped to ATT&CK"""
    name: str
    techniques: List[str]           # ATT&CK technique IDs
    query: str
    platform: str                   # SIEM/EDR platform
    false_positive_rate: float
    validated: bool = False

class ATTACKCoverageAnalyzer:
    """Analyze detection coverage against ATT&CK"""
    
    def __init__(self):
        self.techniques: Dict[str, Technique] = {}
        self.detections: List[Detection] = []
        self.tactics = [
            "Reconnaissance", "Resource Development", "Initial Access",
            "Execution", "Persistence", "Privilege Escalation",
            "Defense Evasion", "Credential Access", "Discovery",
            "Lateral Movement", "Collection", "Command and Control",
            "Exfiltration", "Impact"
        ]
    
    def load_attack_data(self, attack_json_path: str):
        """Load ATT&CK STIX data"""
        # In practice, load from MITRE's STIX bundle
        pass
    
    def add_detection(self, detection: Detection):
        """Add detection and update coverage"""
        self.detections.append(detection)
        
        for tech_id in detection.techniques:
            if tech_id in self.techniques:
                self.techniques[tech_id].detections.append(detection.name)
                self._update_coverage_level(tech_id)
    
    def _update_coverage_level(self, tech_id: str):
        """Recalculate coverage level for technique"""
        tech = self.techniques[tech_id]
        detection_count = len(tech.detections)
        
        if detection_count == 0:
            tech.coverage_level = CoverageLevel.NONE
        elif detection_count == 1:
            tech.coverage_level = CoverageLevel.MINIMAL
        elif detection_count < 3:
            tech.coverage_level = CoverageLevel.PARTIAL
        elif detection_count < 5:
            tech.coverage_level = CoverageLevel.SUBSTANTIAL
        else:
            tech.coverage_level = CoverageLevel.COMPREHENSIVE
    
    def coverage_by_tactic(self) -> Dict[str, dict]:
        """Get coverage statistics per tactic"""
        results = {}
        
        for tactic in self.tactics:
            tactic_techs = [t for t in self.techniques.values() 
                          if t.tactic == tactic]
            
            if not tactic_techs:
                continue
            
            covered = sum(1 for t in tactic_techs 
                         if t.coverage_level != CoverageLevel.NONE)
            
            results[tactic] = {
                'total_techniques': len(tactic_techs),
                'covered': covered,
                'coverage_percent': (covered / len(tactic_techs)) * 100,
                'gaps': [t.id for t in tactic_techs 
                        if t.coverage_level == CoverageLevel.NONE]
            }
        
        return results
    
    def identify_gaps(self) -> List[Technique]:
        """Find techniques with no detection coverage"""
        return [t for t in self.techniques.values() 
                if t.coverage_level == CoverageLevel.NONE]
    
    def prioritize_gaps(self) -> List[Technique]:
        """Prioritize gaps by adversary usage"""
        gaps = self.identify_gaps()
        
        # Prioritize by:
        # 1. Commonly used by threat groups
        # 2. Data sources already available
        # 3. Tactic importance
        
        high_priority_tactics = [
            "Initial Access", "Execution", "Persistence",
            "Credential Access", "Lateral Movement"
        ]
        
        prioritized = sorted(
            gaps,
            key=lambda t: (
                t.tactic in high_priority_tactics,  # Prioritize key tactics
                len(t.data_sources)                  # Easier to detect
            ),
            reverse=True
        )
        
        return prioritized
    
    def generate_navigator_layer(self) -> dict:
        """Generate ATT&CK Navigator layer JSON"""
        layer = {
            "name": "Detection Coverage",
            "version": "4.4",
            "domain": "enterprise-attack",
            "description": "Current detection coverage",
            "techniques": []
        }
        
        color_map = {
            CoverageLevel.NONE: "#ffffff",
            CoverageLevel.MINIMAL: "#ffcccc",
            CoverageLevel.PARTIAL: "#ffff99",
            CoverageLevel.SUBSTANTIAL: "#99ff99",
            CoverageLevel.COMPREHENSIVE: "#00ff00"
        }
        
        for tech in self.techniques.values():
            layer["techniques"].append({
                "techniqueID": tech.id,
                "color": color_map[tech.coverage_level],
                "comment": f"Detections: {len(tech.detections)}"
            })
        
        return layer

基于技术的检测

# technique_detection.py
# Build detections aligned to ATT&CK techniques

from dataclasses import dataclass
from typing import List, Dict

@dataclass
class ATTACKDetection:
    """Detection rule with full ATT&CK context"""
    
    # ATT&CK mapping
    technique_id: str
    technique_name: str
    tactic: str
    
    # Detection details
    name: str
    description: str
    query: str
    platform: str
    
    # Quality metrics
    severity: str
    confidence: str
    false_positive_guidance: str
    
    # Data requirements
    data_sources: List[str]
    
    # Response
    recommended_response: List[str]
    
    def to_sigma(self) -> str:
        """Export as Sigma rule"""
        return f"""
title: {self.name}
id: {self.technique_id.lower().replace('.', '-')}-detection
status: experimental
description: {self.description}
references:
    - https://attack.mitre.org/techniques/{self.technique_id}/
author: Security Team
date: 2024/01/01
tags:
    - attack.{self.tactic.lower().replace(' ', '_')}
    - attack.{self.technique_id.lower()}
logsource:
    category: process_creation
    product: windows
detection:
    selection:
        # Detection logic here
    condition: selection
falsepositives:
    - {self.false_positive_guidance}
level: {self.severity.lower()}
"""


# Example: T1003.001 - LSASS Memory
lsass_detection = ATTACKDetection(
    technique_id="T1003.001",
    technique_name="LSASS Memory",
    tactic="Credential Access",
    name="LSASS Memory Access via Suspicious Process",
    description="Detects processes accessing LSASS memory, "
                "commonly used for credential dumping",
    query="""
        event.code:10 AND 
        winlog.event_data.TargetImage:*lsass.exe AND
        NOT winlog.event_data.SourceImage:(
            *MsMpEng.exe OR *csrss.exe OR *wininit.exe
        )
    """,
    platform="Elastic SIEM",
    severity="High",
    confidence="Medium",
    false_positive_guidance="Security tools may legitimately access LSASS",
    data_sources=["Process: Process Access (Sysmon EID 10)"],
    recommended_response=[
        "Isolate affected host",
        "Capture memory dump for analysis",
        "Reset potentially compromised credentials",
        "Hunt for lateral movement"
    ]
)

威胁组织跟踪

# threat_groups.py
# Track adversary groups and their TTPs

from dataclasses import dataclass
from typing import List, Set, Dict

@dataclass
class ThreatGroup:
    """Known adversary group with TTPs"""
    
    id: str                          # e.g., "G0007"
    name: str                        # e.g., "APT28"
    aliases: List[str]               # e.g., ["Fancy Bear", "Sofacy"]
    
    # Attribution
    suspected_origin: str
    target_sectors: List[str]
    target_regions: List[str]
    
    # TTPs
    techniques: List[str]            # ATT&CK technique IDs
    
    # Intelligence
    first_seen: str
    last_seen: str
    active: bool
    
    def technique_overlap(self, other: 'ThreatGroup') -> Set[str]:
        """Find common techniques with another group"""
        return set(self.techniques) & set(other.techniques)
    
    def unique_techniques(self, all_groups: List['ThreatGroup']) -> Set[str]:
        """Find techniques unique to this group"""
        other_techniques = set()
        for group in all_groups:
            if group.id != self.id:
                other_techniques.update(group.techniques)
        
        return set(self.techniques) - other_techniques


class ThreatIntelligence:
    """Manage threat group intelligence"""
    
    def __init__(self):
        self.groups: Dict[str, ThreatGroup] = {}
    
    def relevant_groups(self, sector: str, region: str) -> List[ThreatGroup]:
        """Find groups targeting specific sector/region"""
        return [
            g for g in self.groups.values()
            if (sector in g.target_sectors or "All" in g.target_sectors) and
               (region in g.target_regions or "Global" in g.target_regions) and
               g.active
        ]
    
    def priority_techniques(self, sector: str, region: str) -> Dict[str, int]:
        """Rank techniques by threat relevance"""
        relevant = self.relevant_groups(sector, region)
        
        technique_counts = {}
        for group in relevant:
            for tech in group.techniques:
                technique_counts[tech] = technique_counts.get(tech, 0) + 1
        
        # Sort by frequency across relevant groups
        return dict(sorted(
            technique_counts.items(),
            key=lambda x: x[1],
            reverse=True
        ))
    
    def detection_priorities(self, 
                            sector: str, 
                            region: str,
                            current_coverage: Set[str]) -> List[str]:
        """Prioritize detections based on threat landscape"""
        priority_techs = self.priority_techniques(sector, region)
        
        # Find high-priority techniques without coverage
        gaps = [
            tech for tech, count in priority_techs.items()
            if tech not in current_coverage and count >= 2
        ]
        
        return gaps[:10]  # Top 10 priorities

狩猎假设生成

# hypothesis_generator.py
# Generate hunt hypotheses from ATT&CK

from dataclasses import dataclass
from typing import List

@dataclass
class HuntHypothesis:
    """ATT&CK-based hunt hypothesis"""
    
    technique_id: str
    technique_name: str
    tactic: str
    
    hypothesis: str
    rationale: str
    
    data_requirements: List[str]
    hunt_query: str
    
    expected_findings: List[str]
    success_criteria: str


def generate_hypotheses(technique_id: str, 
                       technique_data: dict) -> List[HuntHypothesis]:
    """Generate hunt hypotheses for a technique"""
    
    hypotheses = []
    
    # Hypothesis 1: Direct technique detection
    hypotheses.append(HuntHypothesis(
        technique_id=technique_id,
        technique_name=technique_data['name'],
        tactic=technique_data['tactic'],
        hypothesis=f"Adversaries are using {technique_data['name']} "
                   f"in our environment",
        rationale=f"This technique is commonly used by threat groups "
                  f"targeting our sector",
        data_requirements=technique_data['data_sources'],
        hunt_query=technique_data.get('detection_query', ''),
        expected_findings=[
            "Legitimate use by IT/security tools",
            "Developer/admin testing",
            "Potential adversary activity"
        ],
        success_criteria="Identify all instances, triage findings, "
                        "create detection for confirmed malicious patterns"
    ))
    
    # Hypothesis 2: Technique chaining
    if technique_data.get('related_techniques'):
        hypotheses.append(HuntHypothesis(
            technique_id=technique_id,
            technique_name=technique_data['name'],
            tactic=technique_data['tactic'],
            hypothesis=f"Adversaries using {technique_data['name']} "
                       f"are also using related techniques",
            rationale="Techniques are often used in combination",
            data_requirements=technique_data['data_sources'],
            hunt_query="Correlated query across related techniques",
            expected_findings=["Attack chains", "Lateral movement paths"],
            success_criteria="Map complete attack chain if present"
        ))
    
    return hypotheses

思维模型

MITRE ATT&CK 从业者会提出以下问题：

1. 这属于什么战术？ 了解对手的目标
2. 是什么技术？ 识别具体方法
3. 我们能检测到吗？ 评估覆盖范围
4. 谁会使用它？ 归因威胁组织
5. 差距在哪里？ 确定改进优先级

典型 ATT&CK 做法

• 在所有检测和警报中注明技术 ID
• 使用 Navigator 图层可视化覆盖范围
• 跟踪威胁组织的 TTP
• 按战术进行差距分析
• 根据技术生成假设
• 持续评估覆盖范围