Good intelligence separates a red team engagement that emulates a real adversary from one that just runs down a tool list. When you know which TTPs an actor actually uses, which infrastructure they stage from, and which people on the target’s staff are worth researching, the engagement stops being a checklist and starts resembling the thing the client is actually worried about.
Cyber threat intelligence (CTI) is the work of turning raw data into that context. DNS records, breach dumps, forum chatter, commercial feeds, an executive’s public Strava history: none of it is intelligence until you’ve collected it against a specific question, analyzed it, and handed it to someone who can act. This post walks that lifecycle from the operator’s seat, from setting collection requirements through OSINT and closed-source gathering, kill-chain and Diamond Model analysis, STIX/TAXII dissemination, and finally feeding real adversary behavior back into red team planning.
The code below leans Python, because collection and parsing is attacker-host work and Python is where that work usually lives. Treat the scripts as scaffolding for the technique rather than drop-in tools; several stub out the parsing so the structure stays readable.
Intelligence collection frameworks#
Collection starts with a question, not a tool. Gather everything and you drown in noise and leave a wide trail; the discipline is deciding what you actually need before you start pulling data, then collecting it without tipping off the target.
Intelligence collection planning#
Before collecting intelligence, establish clear objectives and requirements:
class IntelligenceRequirement:
def __init__(self, priority, scope, timeframe):
self.priority = priority # Critical, High, Medium, Low
self.scope = scope # Target organization, industry, threat actors
self.timeframe = timeframe # Immediate, Short-term, Long-term
self.collection_methods = []
self.success_criteria = []
def add_collection_method(self, method, tools, risks):
self.collection_methods.append({
'method': method,
'tools': tools,
'operational_risks': risks,
'collection_status': 'pending'
})
# Example intelligence requirement for red team operation
apt_campaign_intel = IntelligenceRequirement(
priority="Critical",
scope="Target organization APT simulation",
timeframe="Short-term"
)
apt_campaign_intel.add_collection_method(
method="Passive DNS reconnaissance",
tools=["PassiveTotal", "VirusTotal", "RiskIQ"],
risks=["IP attribution", "Query rate limits"]
)
apt_campaign_intel.add_collection_method(
method="Dark web monitoring",
tools=["OnionScan", "Ahmia", "Custom scrapers"],
risks=["Legal compliance", "OpSec exposure"]
)Open source intelligence (OSINT) collection#
OSINT is the foundation of most CTI work. It’s accessible, and it doesn’t carry the attribution risk that active reconnaissance does, since you’re reading public sources rather than touching the target.
Social media intelligence#
Social media platforms serve as inadvertent intelligence sources for threat actors and targets alike:
#!/usr/bin/env python3
"""
OSINT Social Media Intelligence Collector
Security Note: Only collect publicly available information
Comply with platform terms of service and local laws
"""
import requests
import json
import time
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import logging
class SocialMediaIntelligence:
def __init__(self, target_username: str, platforms: List[str]):
self.target_username = target_username
self.platforms = platforms
self.session = requests.Session()
self.session.headers.update({
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
})
self.logger = logging.getLogger('SocialMediaIntel')
def collect_twitter_intelligence(self) -> Dict:
"""Collect Twitter/X intelligence without API keys"""
try:
# Use Twitter's web interface (rate limited)
url = f"https://twitter.com/{self.target_username}"
response = self.session.get(url, timeout=10)
if response.status_code == 200:
# Extract publicly visible information
intel = {
'platform': 'twitter',
'username': self.target_username,
'profile_exists': True,
'bio': self.extract_twitter_bio(response.text),
'follower_count': self.extract_follower_count(response.text),
'recent_tweets': self.extract_recent_tweets(response.text),
'collection_timestamp': datetime.now().isoformat()
}
return intel
else:
return {
'platform': 'twitter',
'username': self.target_username,
'profile_exists': False,
'error': f'HTTP {response.status_code}'
}
except Exception as e:
self.logger.error(f"Twitter collection failed: {e}")
return {'error': str(e)}
def collect_github_intelligence(self) -> Dict:
"""Collect GitHub intelligence for developer targets"""
try:
# GitHub user profile
profile_url = f"https://api.github.com/users/{self.target_username}"
profile_response = self.session.get(profile_url, timeout=10)
if profile_response.status_code == 200:
profile_data = profile_response.json()
# Get repositories
repos_url = f"https://api.github.com/users/{self.target_username}/repos"
repos_response = self.session.get(repos_url, timeout=10)
repos = repos_response.json() if repos_response.status_code == 200 else []
intel = {
'platform': 'github',
'username': self.target_username,
'profile_exists': True,
'name': profile_data.get('name'),
'company': profile_data.get('company'),
'location': profile_data.get('location'),
'email': profile_data.get('email'),
'bio': profile_data.get('bio'),
'public_repos': profile_data.get('public_repos', 0),
'followers': profile_data.get('followers', 0),
'following': profile_data.get('following', 0),
'repositories': [{
'name': repo.get('name'),
'language': repo.get('language'),
'stars': repo.get('stargazers_count', 0),
'forks': repo.get('forks_count', 0),
'updated': repo.get('updated_at')
} for repo in repos[:10]], # Top 10 repos
'collection_timestamp': datetime.now().isoformat()
}
return intel
except Exception as e:
self.logger.error(f"GitHub collection failed: {e}")
return {'error': str(e)}
def collect_linkedin_intelligence(self) -> Dict:
"""Collect LinkedIn intelligence (public profiles only)"""
try:
# LinkedIn public profile URL
url = f"https://www.linkedin.com/in/{self.target_username}"
response = self.session.get(url, timeout=10)
intel = {
'platform': 'linkedin',
'username': self.target_username,
'url': url,
'accessible': response.status_code == 200,
'collection_timestamp': datetime.now().isoformat()
}
if response.status_code == 200:
# Extract public information (limited due to login walls)
intel.update({
'name': self.extract_linkedin_name(response.text),
'headline': self.extract_linkedin_headline(response.text),
'location': self.extract_linkedin_location(response.text)
})
return intel
except Exception as e:
self.logger.error(f"LinkedIn collection failed: {e}")
return {'error': str(e)}
def collect_discord_intelligence(self) -> Dict:
"""Collect Discord server intelligence"""
# Discord requires more specialized approaches
# This is a placeholder for more advanced techniques
return {
'platform': 'discord',
'username': self.target_username,
'note': 'Discord intelligence requires specialized tools and access',
'collection_timestamp': datetime.now().isoformat()
}
def run_full_collection(self) -> Dict:
"""Run comprehensive OSINT collection across all platforms"""
intelligence = {
'target_username': self.target_username,
'collection_start': datetime.now().isoformat(),
'platforms_collected': [],
'intelligence': {}
}
platform_collectors = {
'twitter': self.collect_twitter_intelligence,
'github': self.collect_github_intelligence,
'linkedin': self.collect_linkedin_intelligence,
'discord': self.collect_discord_intelligence
}
for platform in self.platforms:
if platform in platform_collectors:
self.logger.info(f"Collecting {platform} intelligence for {self.target_username}")
result = platform_collectors[platform]()
intelligence['intelligence'][platform] = result
intelligence['platforms_collected'].append(platform)
# Rate limiting to avoid detection
time.sleep(2)
intelligence['collection_end'] = datetime.now().isoformat()
return intelligence
# Helper methods for data extraction (simplified implementations)
def extract_twitter_bio(self, html: str) -> Optional[str]:
# Simplified HTML parsing - in practice use BeautifulSoup
return None # Implementation would parse actual HTML
def extract_follower_count(self, html: str) -> Optional[int]:
return None # Implementation would parse actual HTML
def extract_recent_tweets(self, html: str) -> List[Dict]:
return [] # Implementation would parse actual HTML
def extract_linkedin_name(self, html: str) -> Optional[str]:
return None # Implementation would parse actual HTML
def extract_linkedin_headline(self, html: str) -> Optional[str]:
return None # Implementation would parse actual HTML
def extract_linkedin_location(self, html: str) -> Optional[str]:
return None # Implementation would parse actual HTML
# Usage example
collector = SocialMediaIntelligence("target_username", ["twitter", "github", "linkedin"])
intelligence = collector.run_full_collection()
# Save intelligence report
with open(f"osint_report_{intelligence['target_username']}.json", 'w') as f:
json.dump(intelligence, f, indent=2)One caveat before you lean on this: the unauthenticated Twitter/X and LinkedIn scraping shown here stopped working around 2023, once both sites moved profile pages behind login walls. The GitHub collector still runs as-is. For Twitter/X and LinkedIn you now need authenticated API access or a paid OSINT service, so treat those two collectors as placeholders for whatever access you actually have; the surrounding structure still holds.
Technical infrastructure intelligence#
Infrastructure intelligence reveals the technical capabilities and attack surface of targets:
#!/usr/bin/env python3
"""
Infrastructure Intelligence Collection Framework
Security Note: Passive reconnaissance only - no active scanning
"""
import dns.resolver
import whois
import requests
import json
from datetime import datetime
import socket
import ssl
class InfrastructureIntelligence:
def __init__(self, target_domain: str):
self.target_domain = target_domain
self.intelligence = {
'domain': target_domain,
'collection_timestamp': datetime.now().isoformat(),
'dns_intelligence': {},
'whois_intelligence': {},
'ssl_intelligence': {},
'web_intelligence': {}
}
def collect_dns_intelligence(self) -> Dict:
"""Collect comprehensive DNS intelligence"""
dns_intel = {
'a_records': [],
'aaaa_records': [],
'mx_records': [],
'txt_records': [],
'cname_records': [],
'ns_records': [],
'soa_record': None,
'subdomains': []
}
try:
# A records (IPv4 addresses)
try:
answers = dns.resolver.resolve(self.target_domain, 'A')
dns_intel['a_records'] = [str(rdata) for rdata in answers]
except dns.resolver.NXDOMAIN:
dns_intel['a_records'] = ['NXDOMAIN']
except Exception as e:
dns_intel['a_records'] = [f'Error: {e}']
# AAAA records (IPv6 addresses)
try:
answers = dns.resolver.resolve(self.target_domain, 'AAAA')
dns_intel['aaaa_records'] = [str(rdata) for rdata in answers]
except dns.resolver.NoAnswer:
dns_intel['aaaa_records'] = []
except Exception as e:
dns_intel['aaaa_records'] = [f'Error: {e}']
# MX records (mail servers)
try:
answers = dns.resolver.resolve(self.target_domain, 'MX')
dns_intel['mx_records'] = [{'priority': rdata.preference,
'server': str(rdata.exchange)}
for rdata in answers]
except Exception as e:
dns_intel['mx_records'] = [f'Error: {e}']
# TXT records (various metadata)
try:
answers = dns.resolver.resolve(self.target_domain, 'TXT')
dns_intel['txt_records'] = [str(rdata) for rdata in answers]
except Exception as e:
dns_intel['txt_records'] = [f'Error: {e}']
# NS records (name servers)
try:
answers = dns.resolver.resolve(self.target_domain, 'NS')
dns_intel['ns_records'] = [str(rdata) for rdata in answers]
except Exception as e:
dns_intel['ns_records'] = [f'Error: {e}']
# SOA record (domain authority)
try:
answers = dns.resolver.resolve(self.target_domain, 'SOA')
soa = answers[0]
dns_intel['soa_record'] = {
'primary_ns': str(soa.mname),
'responsible_email': str(soa.rname),
'serial': soa.serial,
'refresh': soa.refresh,
'retry': soa.retry,
'expire': soa.expire,
'minimum': soa.minimum
}
except Exception as e:
dns_intel['soa_record'] = f'Error: {e}'
except Exception as e:
dns_intel['error'] = str(e)
return dns_intel
def collect_whois_intelligence(self) -> Dict:
"""Collect WHOIS registration intelligence"""
whois_intel = {}
try:
domain_info = whois.whois(self.target_domain)
whois_intel = {
'registrar': domain_info.get('registrar'),
'creation_date': str(domain_info.get('creation_date')),
'expiration_date': str(domain_info.get('expiration_date')),
'updated_date': str(domain_info.get('updated_date')),
'name_servers': domain_info.get('name_servers', []),
'status': domain_info.get('status', []),
'emails': domain_info.get('emails', []),
'name': domain_info.get('name'),
'org': domain_info.get('org'),
'address': domain_info.get('address'),
'city': domain_info.get('city'),
'state': domain_info.get('state'),
'zipcode': domain_info.get('zipcode'),
'country': domain_info.get('country')
}
except Exception as e:
whois_intel['error'] = str(e)
return whois_intel
def collect_ssl_intelligence(self) -> Dict:
"""Collect SSL certificate intelligence"""
ssl_intel = {
'has_ssl': False,
'certificate_info': {},
'chain_info': [],
'vulnerabilities': []
}
try:
# Create SSL context
context = ssl.create_default_context()
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
# Connect to get certificate
with socket.create_connection((self.target_domain, 443), timeout=10) as sock:
with context.wrap_socket(sock, server_hostname=self.target_domain) as ssock:
ssl_intel['has_ssl'] = True
# Get certificate
cert = ssock.getpeercert()
ssl_intel['certificate_info'] = {
'subject': dict(cert.get('subject', [])),
'issuer': dict(cert.get('issuer', [])),
'version': cert.get('version'),
'serial_number': str(cert.get('serialNumber')),
'not_before': cert.get('notBefore'),
'not_after': cert.get('notAfter'),
'signature_algorithm': cert.get('signatureAlgorithm'),
'public_key_bits': cert.get('publicKey', {}).get('bits'),
'subject_alt_names': cert.get('subjectAltName', [])
}
# Check for common vulnerabilities
ssl_intel['vulnerabilities'] = self.check_ssl_vulnerabilities(cert)
except Exception as e:
ssl_intel['error'] = str(e)
return ssl_intel
def collect_web_intelligence(self) -> Dict:
"""Collect web application intelligence"""
web_intel = {
'server_info': {},
'technologies': [],
'headers': {},
'response_analysis': {}
}
try:
response = requests.get(f"https://{self.target_domain}",
timeout=10, verify=False, allow_redirects=True)
web_intel['status_code'] = response.status_code
web_intel['final_url'] = response.url
web_intel['headers'] = dict(response.headers)
# Server information
server_header = response.headers.get('Server', '')
web_intel['server_info'] = {
'server_header': server_header,
'inferred_technology': self.infer_technology(server_header, response.text)
}
# Technology fingerprinting
web_intel['technologies'] = self.fingerprint_technologies(response.text, response.headers)
except Exception as e:
web_intel['error'] = str(e)
return web_intel
def run_full_collection(self) -> Dict:
"""Run comprehensive infrastructure intelligence collection"""
print(f"Collecting infrastructure intelligence for {self.target_domain}")
self.intelligence['dns_intelligence'] = self.collect_dns_intelligence()
print("✓ DNS intelligence collected")
self.intelligence['whois_intelligence'] = self.collect_whois_intelligence()
print("✓ WHOIS intelligence collected")
self.intelligence['ssl_intelligence'] = self.collect_ssl_intelligence()
print("✓ SSL intelligence collected")
self.intelligence['web_intelligence'] = self.collect_web_intelligence()
print("✓ Web intelligence collected")
return self.intelligence
# Helper methods (simplified implementations)
def check_ssl_vulnerabilities(self, cert) -> List[str]:
vulnerabilities = []
# Implementation would check for Heartbleed, POODLE, etc.
return vulnerabilities
def infer_technology(self, server_header: str, html: str) -> str:
# Simplified technology inference
if 'nginx' in server_header.lower():
return 'nginx'
elif 'apache' in server_header.lower():
return 'apache'
return 'unknown'
def fingerprint_technologies(self, html: str, headers: Dict) -> List[str]:
technologies = []
# Check for common technologies
if 'wp-content' in html:
technologies.append('WordPress')
if 'jquery' in html:
technologies.append('jQuery')
if headers.get('X-Powered-By'):
technologies.append(headers['X-Powered-By'])
return technologies
# Usage
collector = InfrastructureIntelligence("example.com")
intelligence = collector.run_full_collection()
# Save intelligence
with open(f"infrastructure_intel_{intelligence['domain']}.json", 'w') as f:
json.dump(intelligence, f, indent=2)Closed source intelligence (CSINT)#
CSINT is higher-fidelity than OSINT, but you pay for it in money and access. This is the world of commercial feeds and closed, vetted sharing communities.
Commercial threat intelligence platforms#
Recorded Future, Mandiant Advantage, and Cortex XSOAR are the platforms you’ll run into most. You don’t hand-roll authentication against them the way older writeups imply: Recorded Future takes an API token in a request header, Mandiant issues an OAuth-style bearer token you exchange your key and secret for, and the rest sit behind a vendor SDK. The part worth automating isn’t the request signing, it’s normalization. Pull each feed into a common indicator schema, deduplicate across sources, and reconcile the risk scores so one vendor’s “high” doesn’t quietly outvote another’s. Treat any snippet that invents a signing scheme for these APIs as a warning sign; the real ones don’t work that way.
Human intelligence (HUMINT)#
The frameworks above cover technical collection. HUMINT is the other half: intelligence gathered from people rather than systems. It’s often the most valuable intelligence you can get and the hardest to come by, because it depends on access and trust rather than a query.
One way to gather HUMINT is to participate in underground forums or chat rooms frequented by cybercriminals. By posing as a member of the community, you can pick up intelligence on upcoming attacks, new malware families, and other relevant chatter.
Another way to gather HUMINT is to build relationships with insiders or other individuals with access to sensitive information. This can be done through social engineering techniques or by leveraging existing relationships.
Underground collection extends past forums into dark web markets and leak sites, and the mechanics matter. Onion services are reached through Tor, so your OPSEC has to hold: a dedicated VM, no session reuse, nothing in your credentials or writing style that ties back to you, and fresh circuits between sites. What’s actually worth pulling is narrow. Credential dumps and combolists that name your target’s domain, initial-access-broker listings advertising a foothold into an organization that looks like your client, and ransomware leak-site posts showing who’s already been hit and how. Most of what scrolls past is noise or bait. The value is in the handful of posts that touch your engagement scope, and in seeing your target’s exposure before the real adversary acts on it.
Analysis techniques#
Once you have gathered intelligence, you have to analyze it and pull out something actionable. That takes technical skill, familiarity with how current threats actually operate, and the discipline to question what the data seems to be telling you.
One common analysis technique is to use a kill chain model to identify the various stages of an attack and the corresponding TTPs. The kill chain model consists of the following stages: reconnaissance, weaponization, delivery, exploitation, installation, command and control (C2), and actions on objectives.
By mapping intelligence to the different stages of the kill chain, you can identify potential vulnerabilities and develop effective countermeasures. For example, if you identify a new malware family being used in delivery or exploitation, you can develop signatures or behavioral rules to detect and block the malware.
Another analysis technique is to use a diamond model to identify the various actors involved in an attack and their relationships. The diamond model consists of four components: adversaries, infrastructure, capabilities, and victims. Working through each component of the diamond model gives you a fuller picture of who’s operating and how, which feeds directly into mitigation. For example, if you identify a new APT group targeting a particular industry, you can analyze their infrastructure and capabilities to identify potential vulnerabilities and develop effective defenses.
In addition to these techniques, machine learning and artificial intelligence (AI) can be used to analyze large volumes of data and identify patterns or anomalies. For example, machine learning can be used to analyze network traffic and identify potential indicators of compromise (IOCs) or suspicious behavior.
Dissemination techniques#
The final step in CTI is dissemination, or sharing intelligence with relevant stakeholders. Effective dissemination requires clear communication, actionable intelligence, and a deep understanding of the target audience.
One common technique for dissemination is to use a standard format, such as the Structured Threat Information Expression (STIX) or the Trusted Automated eXchange of Indicator Information (TAXII). These formats provide a common language for sharing intelligence and allow for automated ingestion and processing.
Another technique for dissemination is to use a threat intelligence platform (TIP) to aggregate and share intelligence with relevant stakeholders. A TIP can be used to store, analyze, and disseminate intelligence, as well as to integrate with other security tools and technologies.
Advanced CTI operations and red team integration#
CTI-driven red team operations#
Cyber Threat Intelligence transforms red team operations from tactical exercises into strategic campaigns informed by real-world adversary behaviors:
#!/usr/bin/env python3
"""
CTI-Driven Red Team Operations Framework
Security Note: Framework for integrating CTI into red team planning and execution
"""
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import json
import uuid
class CTIDrivenRedTeam:
def __init__(self, campaign_name: str):
self.campaign_name = campaign_name
self.threat_intelligence = []
self.campaign_objectives = []
self.execution_phases = []
self.success_metrics = {}
def ingest_threat_intelligence(self, intelligence_package: Dict):
"""Ingest and normalize threat intelligence for campaign planning"""
normalized_intel = {
'id': str(uuid.uuid4()),
'source': intelligence_package.get('source'),
'threat_actor': intelligence_package.get('threat_actor'),
'ttps': intelligence_package.get('ttps', []),
'indicators': intelligence_package.get('indicators', []),
'confidence': intelligence_package.get('confidence', 50),
'timestamp': intelligence_package.get('timestamp', datetime.now().isoformat())
}
self.threat_intelligence.append(normalized_intel)
def develop_campaign_objectives(self):
"""Develop campaign objectives based on ingested intelligence"""
objectives = []
# Analyze threat actor TTPs
actor_ttps = {}
for intel in self.threat_intelligence:
actor = intel['threat_actor']
if actor not in actor_ttps:
actor_ttps[actor] = []
actor_ttps[actor].extend(intel['ttps'])
# Generate objectives based on most common TTPs
for actor, ttps in actor_ttps.items():
if ttps:
primary_ttp = max(set(ttps), key=ttps.count)
objectives.append({
'actor': actor,
'objective': f'Emulate {primary_ttp} techniques used by {actor}',
'success_criteria': f'Successfully demonstrate {primary_ttp} without detection',
'risk_level': self.assess_ttp_risk(primary_ttp)
})
self.campaign_objectives = objectives
def plan_execution_phases(self):
"""Plan campaign execution phases using MITRE ATT&CK framework"""
phases = [
{
'phase': 'reconnaissance',
'objectives': ['Gather target intelligence', 'Identify attack vectors'],
'ttps': ['T1595', 'T1590', 'T1592'], # MITRE ATT&CK IDs
'tools': ['Maltego', 'theHarvester', 'Shodan'],
'duration_days': 7
},
{
'phase': 'initial_access',
'objectives': ['Gain initial foothold', 'Establish persistence'],
'ttps': ['T1566', 'T1190', 'T1133'],
'tools': ['Metasploit', 'Cobalt Strike', 'Custom implants'],
'duration_days': 3
},
{
'phase': 'execution',
'objectives': ['Execute malicious code', 'Move laterally'],
'ttps': ['T1059', 'T1204', 'T1570'],
'tools': ['PowerShell Empire', 'BloodHound', 'Mimikatz'],
'duration_days': 5
},
{
'phase': 'command_and_control',
'objectives': ['Establish C2 channel', 'Maintain access'],
'ttps': ['T1573', 'T1001', 'T1095'],
'tools': ['Covenant', 'Brute Ratel', 'DNS tunneling tools'],
'duration_days': 10
},
{
'phase': 'exfiltration',
'objectives': ['Steal sensitive data', 'Cover tracks'],
'ttps': ['T1041', 'T1020', 'T1070'],
'tools': ['Rclone', 'Exfil tools', 'Anti-forensic techniques'],
'duration_days': 3
}
]
# Customize phases based on intelligence
for phase in phases:
phase['adapted_ttps'] = self.adapt_ttps_to_intelligence(phase['ttps'])
self.execution_phases = phases
def adapt_ttps_to_intelligence(self, base_ttps: List[str]) -> List[str]:
"""Adapt TTPs based on available threat intelligence"""
adapted_ttps = base_ttps.copy()
# Add TTPs commonly used by identified threat actors
actor_ttps = set()
for intel in self.threat_intelligence:
actor_ttps.update(intel['ttps'])
# Include relevant TTPs from intelligence
adapted_ttps.extend(list(actor_ttps)[:3]) # Add up to 3 additional TTPs
return list(set(adapted_ttps)) # Remove duplicates
def assess_ttp_risk(self, ttp: str) -> str:
"""Assess risk level of TTP implementation"""
high_risk_ttps = ['T1003', 'T1485', 'T1490'] # Credential dumping, data destruction
medium_risk_ttps = ['T1059', 'T1570', 'T1021'] # Common lateral movement
if any(ttp.startswith(high) for high in high_risk_ttps):
return 'high'
elif any(ttp.startswith(medium) for medium in medium_risk_ttps):
return 'medium'
else:
return 'low'
def execute_campaign_phase(self, phase_name: str) -> Dict:
"""Execute a specific campaign phase"""
phase = next((p for p in self.execution_phases if p['phase'] == phase_name), None)
if not phase:
return {'error': f'Phase {phase_name} not found'}
execution_results = {
'phase': phase_name,
'start_time': datetime.now().isoformat(),
'objectives_completed': [],
'ttps_executed': [],
'tools_used': [],
'issues_encountered': [],
'lessons_learned': []
}
# Simulate phase execution (in real implementation, this would execute actual tools)
for ttp in phase['adapted_ttps'][:2]: # Execute first 2 TTPs as example
execution_results['ttps_executed'].append({
'ttp': ttp,
'status': 'completed',
'timestamp': datetime.now().isoformat()
})
execution_results['end_time'] = datetime.now().isoformat()
return execution_results
def generate_campaign_report(self) -> Dict:
"""Generate comprehensive campaign report"""
report = {
'campaign_name': self.campaign_name,
'execution_date': datetime.now().isoformat(),
'intelligence_sources': [intel['source'] for intel in self.threat_intelligence],
'campaign_objectives': self.campaign_objectives,
'phases_executed': [],
'overall_success_rate': 0.0,
'key_findings': [],
'recommendations': []
}
# Simulate phase execution results
total_phases = len(self.execution_phases)
successful_phases = 0
for phase in self.execution_phases:
phase_result = self.execute_campaign_phase(phase['phase'])
report['phases_executed'].append(phase_result)
if 'error' not in phase_result:
successful_phases += 1
report['overall_success_rate'] = successful_phases / total_phases if total_phases > 0 else 0
# Generate findings based on intelligence
if self.threat_intelligence:
report['key_findings'].append(
f"Successfully emulated TTPs from {len(set([i['threat_actor'] for i in self.threat_intelligence]))} threat actors"
)
report['recommendations'] = [
"Implement additional monitoring for emulated TTPs",
"Update detection rules based on campaign findings",
"Conduct lessons learned session with blue team"
]
return report
# Usage example
# red_team = CTIDrivenRedTeam("APT28 Emulation Campaign")
#
# # Ingest threat intelligence
# intel_package = {
# 'source': 'Mandiant',
# 'threat_actor': 'APT28',
# 'ttps': ['T1059.001', 'T1071.001', 'T1566.001'],
# 'confidence': 85
# }
#
# red_team.ingest_threat_intelligence(intel_package)
# red_team.develop_campaign_objectives()
# red_team.plan_execution_phases()
#
# report = red_team.generate_campaign_report()
# print(f"Campaign success rate: {report['overall_success_rate']:.1%}")Intelligence operations security (OPSEC)#
Maintaining operational security during intelligence collection:
#!/usr/bin/env python3
"""
Intelligence Operations Security Framework
Security Note: OPSEC principles for CTI collection and analysis
"""
import hashlib
import hmac
import secrets
from cryptography.fernet import Fernet
from datetime import datetime, timedelta
import logging
from typing import Dict, List, Optional
class IntelligenceOPSEC:
def __init__(self, encryption_key: Optional[bytes] = None):
self.encryption_key = encryption_key or Fernet.generate_key()
self.cipher = Fernet(self.encryption_key)
self.audit_log = []
self.logger = logging.getLogger('IntelOPSEC')
def sanitize_intelligence(self, intelligence: Dict) -> Dict:
"""Remove or mask sensitive information from intelligence"""
sanitized = intelligence.copy()
# Remove sensitive fields
sensitive_fields = ['api_keys', 'passwords', 'internal_ips', 'employee_names']
for field in sensitive_fields:
if field in sanitized:
sanitized[field] = '[REDACTED]'
# Mask partial sensitive data
if 'email' in sanitized:
email = sanitized['email']
if '@' in email:
local, domain = email.split('@', 1)
masked_local = local[:2] + '*' * (len(local) - 2) if len(local) > 2 else local
sanitized['email'] = f"{masked_local}@{domain}"
# Log sanitization action
self.audit_log.append({
'action': 'sanitization',
'timestamp': datetime.now().isoformat(),
'fields_modified': sensitive_fields
})
return sanitized
def encrypt_sensitive_data(self, data: str) -> str:
"""Encrypt sensitive intelligence data"""
encrypted = self.cipher.encrypt(data.encode())
return encrypted.decode('utf-8')
def decrypt_sensitive_data(self, encrypted_data: str) -> str:
"""Decrypt sensitive intelligence data"""
try:
decrypted = self.cipher.decrypt(encrypted_data.encode())
return decrypted.decode('utf-8')
except Exception as e:
self.logger.error(f"Decryption failed: {e}")
return "[DECRYPTION_FAILED]"
def generate_anonymized_report(self, intelligence: Dict) -> Dict:
"""Generate anonymized intelligence report"""
anonymized = {
'report_id': secrets.token_hex(16),
'generation_date': datetime.now().isoformat(),
'anonymized_intelligence': {},
'anonymization_metadata': {}
}
# Anonymize key intelligence
if 'indicators' in intelligence:
anonymized['anonymized_intelligence']['indicator_count'] = len(intelligence['indicators'])
anonymized['anonymized_intelligence']['indicator_types'] = \
list(set(ind['type'] for ind in intelligence['indicators']))
if 'threat_actors' in intelligence:
anonymized['anonymized_intelligence']['actor_count'] = len(intelligence['threat_actors'])
# Don't include actual actor names
anonymized['anonymization_metadata'] = {
'anonymization_method': 'field_removal_and_masking',
'anonymization_date': datetime.now().isoformat(),
'anonymizer_version': '1.0'
}
return anonymized
def validate_information_sharing(self, recipient: str, intelligence: Dict) -> bool:
"""Validate that intelligence can be safely shared with recipient"""
# Check classification level
classification = intelligence.get('classification', 'unclassified')
allowed_classifications = self.get_allowed_classifications(recipient)
if classification not in allowed_classifications:
self.logger.warning(f"Sharing blocked: {classification} not allowed for {recipient}")
return False
# Check need-to-know
required_clearance = self.get_required_clearance(intelligence)
recipient_clearance = self.get_recipient_clearance(recipient)
if recipient_clearance < required_clearance:
self.logger.warning(f"Sharing blocked: insufficient clearance for {recipient}")
return False
# Check data minimization
if not self.is_minimized(intelligence):
self.logger.warning(f"Sharing blocked: intelligence not properly minimized")
return False
return True
def get_allowed_classifications(self, recipient: str) -> List[str]:
"""Get classifications recipient is cleared to receive"""
# Simplified clearance matrix
clearances = {
'internal_soc': ['unclassified', 'confidential', 'secret'],
'external_partner': ['unclassified'],
'public': ['unclassified']
}
return clearances.get(recipient, ['unclassified'])
def get_required_clearance(self, intelligence: Dict) -> int:
"""Get required clearance level for intelligence"""
classification = intelligence.get('classification', 'unclassified')
clearance_levels = {
'unclassified': 1,
'confidential': 2,
'secret': 3,
'top_secret': 4
}
return clearance_levels.get(classification, 1)
def get_recipient_clearance(self, recipient: str) -> int:
"""Get recipient's clearance level"""
clearances = {
'internal_soc': 3,
'external_partner': 1,
'public': 1
}
return clearances.get(recipient, 1)
def is_minimized(self, intelligence: Dict) -> bool:
"""Check if intelligence follows data minimization principles"""
# Ensure only necessary fields are included
required_fields = {'type', 'value', 'confidence', 'source'}
optional_fields = {'context', 'tags', 'timestamp'}
for item in intelligence.get('indicators', []):
item_fields = set(item.keys())
if not required_fields.issubset(item_fields):
return False
# Check for excessive optional data
extra_fields = item_fields - required_fields - optional_fields
if len(extra_fields) > 2: # Allow max 2 extra fields
return False
return True
def log_intelligence_access(self, user: str, action: str, intelligence_id: str):
"""Log intelligence access for audit purposes"""
log_entry = {
'timestamp': datetime.now().isoformat(),
'user': user,
'action': action,
'intelligence_id': intelligence_id,
'ip_address': 'logged', # In real implementation, get actual IP
'user_agent': 'logged' # In real implementation, get actual UA
}
self.audit_log.append(log_entry)
self.logger.info(f"Intelligence access logged: {action} by {user}")
def generate_opsec_report(self) -> Dict:
"""Generate OPSEC compliance report"""
report = {
'generation_date': datetime.now().isoformat(),
'audit_entries': len(self.audit_log),
'opsec_violations': [],
'recommendations': []
}
# Analyze audit log for potential OPSEC issues
access_by_user = {}
for entry in self.audit_log:
user = entry['user']
if user not in access_by_user:
access_by_user[user] = []
access_by_user[user].append(entry)
# Check for suspicious access patterns
for user, accesses in access_by_user.items():
if len(accesses) > 100: # Arbitrary threshold
report['opsec_violations'].append({
'type': 'excessive_access',
'user': user,
'access_count': len(accesses)
})
report['recommendations'] = [
"Implement multi-factor authentication for intelligence access",
"Regular security awareness training for intelligence personnel",
"Automated alerting for suspicious access patterns",
"Regular OPSEC audits and compliance reviews"
]
return report
# Usage example
# opsec = IntelligenceOPSEC()
#
# # Sanitize intelligence before sharing
# sensitive_intel = {'api_keys': 'secret', 'indicators': [...]}
# sanitized = opsec.sanitize_intelligence(sensitive_intel)
#
# # Validate sharing
# if opsec.validate_information_sharing('external_partner', sanitized):
# share_intelligence(sanitized)
#
# # Generate OPSEC report
# opsec_report = opsec.generate_opsec_report()References#
Core CTI frameworks and standards#
- MITRE ATT&CK Framework: Comprehensive knowledge base of adversary tactics and techniques
- Cyber Kill Chain: Lockheed Martin’s model for understanding cyber attacks
- Diamond Model of Intrusion Analysis: Caltagirone, Pendergast, and Betz’s framework (2013) for analyzing intrusions as adversary, infrastructure, capability, and victim
- STIX (Structured Threat Information Expression): Standardized language for describing threat information
- TAXII (Trusted Automated eXchange of Indicator Information): Protocol for sharing threat intelligence
Threat intelligence platforms#
- MISP (MISP Threat Sharing, originally Malware Information Sharing Platform): Open source threat intelligence sharing platform
- OpenIOC: Open Indicators of Compromise format
- VERIS (Vocabulary for Event Recording and Incident Sharing): Standardized vocabulary for incident description
- CybOX (Cyber Observable Expression): Standardized language for cyber observables
Academic and research resources#
- “Intelligence-Driven Computer Network Defense”: 2011 paper by Hutchins, Cloppert, and Amin (Lockheed Martin) that introduced the Cyber Kill Chain
- “The Art of Deception”: Kevin Mitnick’s exploration of social engineering
- “Sandworm”: Andy Greenberg’s investigation of Russian cyber operations
- “This Is How They Tell Me the World Ends”: Nicole Perlroth’s analysis of zero-day markets
Professional certifications#
- Certified Threat Intelligence Analyst (CTIA): EC-Council certification
- GIAC Critical Infrastructure Protection (GCIP): GIAC certification
- Certified Information Forensics Investigator (CIFI): IISFA certification
- CREST Registered Threat Intelligence Analyst: CREST certification
Industry reports and databases#
- Verizon DBIR (Data Breach Investigations Report): Annual analysis of security incidents
- Mandiant M-Trends Report: Annual threat intelligence analysis
- CrowdStrike Global Threat Report: Comprehensive threat landscape analysis
- Microsoft Digital Defense Report: Microsoft’s view of the threat landscape
Tools and software#
- Maltego: Open source intelligence and forensics application
- theHarvester: Email, domain, and IP address harvesting tool
- Shodan: Search engine for internet-connected devices
- VirusTotal: Online virus/malware scanning service
- Recorded Future: Commercial threat intelligence platform
- Mandiant Advantage: Threat intelligence platform
- Cortex XSOAR: Security orchestration and automation platform
Where CTI earns its keep on engagement#
Strip away the frameworks and CTI comes down to a few habits. Collect against a stated requirement instead of hoovering up everything in reach. Run what you collect through a structured method (the kill chain, the Diamond Model) rather than eyeballing it. Hand it off in a format the recipient can actually ingest. And keep your own OPSEC intact while you do it, because the same collection techniques work just as well pointed back at you.
For a red team, the payoff is specific: emulating a real actor’s TTPs instead of running a generic playbook. When you’ve profiled how a group actually stages infrastructure and moves once it lands, the engagement surfaces the gaps the client would hit against the real thing, not the gaps that only exist against a scanner. That’s the difference between a report the blue team files and a report that changes what they monitor on Monday.
None of it requires a commercial feed or a dedicated intel team to start. A written collection plan, a couple of the scripts above adapted to your target, and a framework to organize what comes back will already put you ahead of most engagements.
