Sector-specific intelligence comes from ISACs (Information Sharing and Analysis Centers), government advisories (CISA, FBI, NSA), vendor threat reports (CrowdStrike Annual Threat Report, Mandiant M-Trends, Verizon DBIR), and academic research on sector-specific attacks.
from attackcti import attack_client
import json
class SectorThreatAssessment:
SECTOR_GROUPS = {
"financial": ["FIN7", "FIN8", "FIN11", "Carbanak", "Lazarus Group",
"Cobalt Group", "TA505", "GOLD SOUTHFIELD"],
"healthcare": ["FIN12", "Ryuk", "Conti", "Wizard Spider",
"GOLD ULRICK", "Vice Society"],
"energy": ["TEMP.Veles", "Sandworm Team", "Dragonfly",
"XENOTIME", "ERYTHRITE", "Berserk Bear"],
"government": ["APT29", "APT28", "Turla", "Gamaredon Group",
"Mustang Panda", "APT41", "Lazarus Group"],
"manufacturing": ["APT41", "TEMP.Veles", "Dragonfly",
"HEXANE", "MAGNALLIUM"],
"technology": ["APT41", "Lazarus Group", "APT10",
"HAFNIUM", "Winnti Group"],
}
def __init__(self, sector):
self.sector = sector.lower()
self.lift = attack_client()
self.groups = self.lift.get_groups()
self.assessment = {
"sector": sector,
"threat_actors": [],
"common_techniques": {},
"attack_vectors": {},
"risk_summary": {},
}
def analyze_sector_actors(self):
"""Analyze threat actors known to target this sector."""
target_groups = self.SECTOR_GROUPS.get(self.sector, [])
actor_profiles = []
for group_name in target_groups:
group = next(
(g for g in self.groups
if g.get("name", "").lower() == group_name.lower()
or group_name.lower() in [a.lower() for a in g.get("aliases", [])]),
None
)
if group:
group_id = ""
for ref in group.get("external_references", []):
if ref.get("source_name") == "mitre-attack":
group_id = ref.get("external_id", "")
break
techniques = []
if group_id:
techs = self.lift.get_techniques_used_by_group(group_id)
for t in techs:
for ref in t.get("external_references", []):
if ref.get("source_name") == "mitre-attack":
techniques.append({
"id": ref.get("external_id", ""),
"name": t.get("name", ""),
})
break
profile = {
"name": group.get("name", ""),
"aliases": group.get("aliases", []),
"description": group.get("description", "")[:300],
"attack_id": group_id,
"technique_count": len(techniques),
"techniques": techniques[:20],
}
actor_profiles.append(profile)
print(f" [+] {group.get('name')}: {len(techniques)} techniques")
self.assessment["threat_actors"] = actor_profiles
print(f"[+] Profiled {len(actor_profiles)} threat actors for {self.sector}")
return actor_profiles
def identify_common_techniques(self):
"""Find the most commonly used techniques across sector actors."""
from collections import Counter
technique_counter = Counter()
for actor in self.assessment["threat_actors"]:
for tech in actor.get("techniques", []):
technique_counter[f"{tech['id']}:{tech['name']}"] += 1
common = technique_counter.most_common(20)
self.assessment["common_techniques"] = [
{
"technique": tech.split(":")[0],
"name": tech.split(":")[1] if ":" in tech else "",
"actor_count": count,
"actors_using": [
a["name"] for a in self.assessment["threat_actors"]
if any(t["id"] == tech.split(":")[0] for t in a.get("techniques", []))
],
}
for tech, count in common
]
print(f"\n=== Top Techniques for {self.sector.upper()} ===")
for entry in self.assessment["common_techniques"][:10]:
print(f" {entry['technique']} {entry['name']}: "
f"used by {entry['actor_count']} groups")
return self.assessment["common_techniques"]
assessment = SectorThreatAssessment("financial")
assessment.analyze_sector_actors()
assessment.identify_common_techniques()
def analyze_attack_vectors(assessment):
"""Analyze initial access vectors common for the sector."""
initial_access_techniques = [
t for t in assessment.assessment["common_techniques"]
if t["technique"].startswith("T1566") or t["technique"].startswith("T1190")
or t["technique"].startswith("T1133") or t["technique"].startswith("T1078")
or t["technique"].startswith("T1195")
]
sector_vectors = {
"financial": {
"primary": ["Spearphishing (T1566)", "Exploit Public-Facing App (T1190)",
"Valid Accounts (T1078)", "Supply Chain Compromise (T1195)"],
"emerging": ["MFA Fatigue/Push Bombing", "QR Code Phishing (Quishing)",
"Business Email Compromise", "API Key Theft"],
},
"healthcare": {
"primary": ["Spearphishing (T1566)", "Exploit Public-Facing App (T1190)",
"External Remote Services (T1133)", "Valid Accounts (T1078)"],
"emerging": ["IoMT Device Exploitation", "Telehealth Platform Attacks",
"Medical Device Firmware Attacks", "Supply Chain via EHR Vendors"],
},
"energy": {
"primary": ["Spearphishing (T1566)", "Exploit Public-Facing App (T1190)",
"External Remote Services (T1133)", "Supply Chain Compromise (T1195)"],
"emerging": ["OT/ICS Protocol Exploitation", "Remote Access to SCADA",
"Engineering Workstation Compromise", "Vendor VPN Exploitation"],
},
}
vectors = sector_vectors.get(assessment.sector, {})
assessment.assessment["attack_vectors"] = vectors
return vectors
def generate_sector_report(assessment):
data = assessment.assessment
report = f"""# {data['sector'].title()} Sector Threat Landscape Assessment
Generated: {datetime.datetime.now().isoformat()}
## Executive Summary
This assessment analyzes the cyber threat landscape for the {data['sector']} sector,
identifying {len(data['threat_actors'])} active threat groups, their preferred techniques,
and recommended defensive priorities.
## Threat Actor Summary
| Actor | ATT&CK ID | Techniques | Key Focus |
|-------|-----------|------------|-----------|
"""
for actor in data["threat_actors"]:
report += (f"| {actor['name']} | {actor['attack_id']} "
f"| {actor['technique_count']} | {actor['description'][:60]}... |\n")
report += f"""
## Most Common Techniques
| Rank | Technique | Name | Groups Using |
|------|-----------|------|-------------|
"""
for i, tech in enumerate(data.get("common_techniques", [])[:15], 1):
actors = ", ".join(tech["actors_using"][:3])
report += f"| {i} | {tech['technique']} | {tech['name']} | {actors} |\n"
vectors = data.get("attack_vectors", {})
report += f"""
## Attack Vectors
This section covers attack vectors for performing threat landscape assessment for sector.
- Ensure all prerequisites are met before proceeding
- Follow the documented workflow steps in sequence
- Record results and any anomalies encountered during this phase
### Primary Vectors
"""
for v in vectors.get("primary", []):
report += f"- {v}\n"
report += "\n### Emerging Vectors\n"
for v in vectors.get("emerging", []):
report += f"- {v}\n"
report += """
## Recommendations
1. Prioritize detections for the top 10 techniques used by sector-targeting groups
2. Conduct threat-informed red team exercises mimicking identified actors
3. Join sector ISAC for real-time threat sharing
4. Implement controls for identified initial access vectors
5. Review supply chain security posture for sector-specific risks
"""
with open(f"threat_landscape_{data['sector']}.md", "w") as f:
f.write(report)
print(f"[+] Sector report saved: threat_landscape_{data['sector']}.md")
generate_sector_report(assessment)
