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)

undefined

# 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

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

undefined

undefined

# 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"""

undefined

undefined

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

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

undefined

undefined

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

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

undefined