| name | tracking-threat-actor-infrastructure |
| description | Threat actor infrastructure tracking involves monitoring and mapping adversary-controlled assets including command-and-control (C2) servers, phishing domains, exploit kit hosts, bulletproof hosting, a |
| domain | cybersecurity |
| subdomain | threat-intelligence |
| tags | ["threat-intelligence","cti","ioc","mitre-attack","stix","infrastructure-tracking","shodan","censys","passive-dns"] |
| version | 1.0 |
| author | mahipal |
| license | Apache-2.0 |
Tracking Threat Actor Infrastructure
Overview
Threat actor infrastructure tracking involves monitoring and mapping adversary-controlled assets including command-and-control (C2) servers, phishing domains, exploit kit hosts, bulletproof hosting, and staging servers. This skill covers using passive DNS, certificate transparency logs, Shodan/Censys scanning, WHOIS analysis, and network fingerprinting to discover, track, and pivot across threat actor infrastructure over time.
Prerequisites
- Python 3.9+ with
shodan, censys, requests, stix2 libraries
- API keys: Shodan, Censys, VirusTotal, SecurityTrails, PassiveTotal
- Understanding of DNS, TLS/SSL certificates, IP allocation, ASN structure
- Familiarity with passive DNS and certificate transparency concepts
- Access to domain registration (WHOIS) lookup services
Key Concepts
Infrastructure Pivoting
Pivoting is the technique of using one known indicator to discover related infrastructure. Starting from a known C2 IP address, analysts can pivot via: passive DNS (find domains), reverse WHOIS (find related registrations), SSL certificates (find shared certs), SSH key fingerprints, HTTP response fingerprints, JARM/JA3S hashes, and WHOIS registrant data.
Passive DNS
Passive DNS databases record DNS query/response data observed at recursive resolvers. This allows analysts to find historical domain-to-IP mappings, discover domains hosted on a known C2 IP, and identify fast-flux or domain generation algorithm (DGA) behavior.
Certificate Transparency
Certificate Transparency (CT) logs publicly record all SSL/TLS certificates issued by CAs. Monitoring CT logs reveals new certificates registered for suspicious domains, helping identify phishing sites and C2 infrastructure before they become active.
Network Fingerprinting
- JARM: Active TLS server fingerprint (hash of TLS handshake responses)
- JA3S: Passive TLS server fingerprint (hash of Server Hello)
- HTTP Headers: Server banners, custom headers, response patterns
- Favicon Hash: Hash of HTTP favicon for server identification
Practical Steps
Step 1: Shodan Infrastructure Discovery
import shodan
api = shodan.Shodan("YOUR_SHODAN_API_KEY")
def discover_infrastructure(ip_address):
"""Discover services and metadata for a target IP."""
try:
host = api.host(ip_address)
return {
"ip": host["ip_str"],
"org": host.get("org", ""),
"asn": host.get("asn", ""),
"isp": host.get("isp", ""),
"country": host.get("country_name", ""),
"city": host.get("city", ""),
"os": host.get("os"),
"ports": host.get("ports", []),
"vulns": host.get("vulns", []),
"hostnames": host.get("hostnames", []),
"domains": host.get("domains", []),
"tags": host.get("tags", []),
"services": [
{
"port": svc.get("port"),
"transport": svc.get("transport"),
"product": svc.get("product", ""),
"version": svc.get("version", ),
: svc.get(, {}).get(, {}).get(, {}),
: svc.get(, {}).get(, ),
}
svc host.get(, [])
],
}
shodan.APIError e:
()
():
c2_queries = {
: ,
: ,
: ,
: ,
: ,
}
query = c2_queries.get(framework_name.lower(), framework_name)
results = api.search(query, limit=)
hosts = []
results.get(, []):
hosts.append({
: [],
: [],
: .get(, ),
: .get(, {}).get(, ),
: .get(, ),
: .get(, ),
})
hosts
Step 2: Passive DNS Pivoting
import requests
def passive_dns_lookup(indicator, api_key, indicator_type="ip"):
"""Query SecurityTrails for passive DNS records."""
base_url = "https://api.securitytrails.com/v1"
headers = {"APIKEY": api_key, "Accept": "application/json"}
if indicator_type == "ip":
url = f"{base_url}/search/list"
payload = {
"filter": {"ipv4": indicator}
}
resp = requests.post(url, json=payload, headers=headers, timeout=30)
else:
url = f"{base_url}/domain/{indicator}/subdomains"
resp = requests.get(url, headers=headers, timeout=30)
if resp.status_code == 200:
return resp.json()
return None
def query_passive_total(indicator, user, api_key):
"""Query PassiveTotal for passive DNS and WHOIS data."""
base_url = "https://api.passivetotal.org/v2"
auth = (user, api_key)
pdns_resp = requests.get(
f"{base_url}/dns/passive",
params={"query": indicator},
auth=auth,
timeout=30,
)
whois_resp = requests.get(
f"{base_url}/whois",
params={"query": indicator},
auth=auth,
timeout=30,
)
results = {}
pdns_resp.status_code == :
results[] = pdns_resp.json().get(, [])
whois_resp.status_code == :
results[] = whois_resp.json()
results
Step 3: Certificate Transparency Monitoring
import requests
def search_ct_logs(domain):
"""Search Certificate Transparency logs via crt.sh."""
resp = requests.get(
f"https://crt.sh/?q=%.{domain}&output=json",
timeout=30,
)
if resp.status_code == 200:
certs = resp.json()
unique_domains = set()
cert_info = []
for cert in certs:
name_value = cert.get("name_value", "")
for name in name_value.split("\n"):
unique_domains.add(name.strip())
cert_info.append({
"id": cert.get("id"),
"issuer": cert.get("issuer_name", ""),
"common_name": cert.get("common_name", ""),
"name_value": name_value,
"not_before": cert.get("not_before", ""),
"not_after": cert.get("not_after", ""),
"serial_number": cert.get("serial_number", ""),
})
return {
"domain": domain,
"total_certificates": len(certs),
"unique_domains": sorted(unique_domains),
"certificates": cert_info[:],
}
():
datetime datetime, timedelta
cutoff = (datetime.utcnow() - timedelta(hours=interval_hours)).isoformat()
new_certs = []
domain domains:
result = search_ct_logs(domain)
result:
cert result.get(, []):
cert.get(, ) > cutoff:
new_certs.append({
: domain,
: cert,
})
new_certs
Step 4: Infrastructure Correlation and Timeline
from datetime import datetime
def build_infrastructure_timeline(indicators):
"""Build a timeline of infrastructure changes."""
timeline = []
for ind in indicators:
if "passive_dns" in ind:
for record in ind["passive_dns"]:
timeline.append({
"timestamp": record.get("firstSeen", ""),
"event": "dns_resolution",
"source": record.get("resolve", ""),
"target": record.get("value", ""),
"record_type": record.get("recordType", ""),
})
if "certificates" in ind:
for cert in ind["certificates"]:
timeline.append({
"timestamp": cert.get("not_before", ""),
"event": "certificate_issued",
"domain": cert.get("common_name", ""),
"issuer": cert.get("issuer", ""),
})
timeline.sort(key=lambda x: x.get(, ))
timeline
Validation Criteria
- Shodan/Censys queries return infrastructure details for target IPs
- Passive DNS reveals historical domain-IP mappings
- Certificate transparency search finds associated domains
- Infrastructure pivoting discovers new related indicators
- Timeline shows infrastructure evolution over time
- Results are exportable as STIX 2.1 Infrastructure objects
References
