建構攻擊者基礎設施追蹤系統 是做什麼的？
概述
攻擊者基礎設施追蹤使用被動 DNS 記錄、憑證透明度記錄、WHOIS 註冊資料和 IP 強化，來發現、繪製和監控威脅行為者的命令與控制（C2）網路。攻擊者經常在多次行動中重複使用託管供應商、註冊商、SSL 憑證和命名模式，使分析師能夠從已知指標進行樞紐分析，以發現新的基礎設施。本技能涵蓋建立一個自動化追蹤系統，該系統能識別基礎設施之間的關係、偵測符合攻擊者模式的新註冊網域，並維護一個持續更新的威脅行為者網路地圖。
使用時機
- 在您的環境中部署或設定攻擊者基礎設施追蹤系統功能時
- 建立符合合規要求的安全控制時
- 為此領域建立或改善安全架構時
- 進行需要此實作的安全性評估時
先決條件
- Python 3.9+ 與
requests、dnspython、python-whois、shodan、networkx函式庫 - API 金鑰：SecurityTrails、PassiveTotal/RiskIQ、Shodan、VirusTotal
- 存取被動 DNS 資料來源
- 了解 DNS 基礎設施、託管和網域註冊
- 圖形資料庫（Neo4j）或 NetworkX 用於關係視覺化
關鍵概念
被動 DNS
被動 DNS 擷取歷史 DNS 解析資料，記錄哪些網域在何時解析到哪些 IP。與主動 DNS 查詢不同，被動 DNS 即使在記錄變更後仍保留歷史關係，使分析師能夠追蹤基礎設施變更、識別共享託管模式，並發現過去解析到相同 IP 的相關網域。
基礎設施樞紐分析
樞紐分析透過追蹤連線來識別相關基礎設施：IP 樞紐（尋找某個 IP 上的所有網域）、網域樞紐（尋找某網域解析過的所有 IP）、WHOIS 樞紐（尋找相同註冊人的網域）、憑證樞紐（尋找共享 SSL 憑證的主機）以及 NS/MX 樞紐（尋找使用相同名稱伺服器或郵件伺服器的網域）。
攻擊者基礎設施模式
威脅行為者會表現出一些模式：偏好的註冊商（Namecheap、REG.RU、Tucows）、偏好的託管（防彈託管供應商、雲端服務）、網域生成演算法（DGA）、一致的命名模式以及跨行動的憑證重用。
工作流程
步驟 1：被動 DNS 基礎設施發現
import requests
import json
from collections import defaultdict
from datetime import datetime
class InfrastructureTracker:
def __init__(self, securitytrails_key=None, vt_key=None, shodan_key=None):
self.st_key = securitytrails_key
self.vt_key = vt_key
self.shodan_key = shodan_key
self.infrastructure_graph = defaultdict(lambda: {"nodes": set(), "edges": []})
def passive_dns_lookup(self, domain):
"""查詢被動 DNS 以取得網域解析歷史。"""
headers = {"apikey": self.st_key}
url = f"https://api.securitytrails.com/v1/history/{domain}/dns/a"
resp = requests.get(url, headers=headers, timeout=30)
if resp.status_code == 200:
records = resp.json().get("records", [])
history = []
for record in records:
for value in record.get("values", []):
history.append({
"domain": domain,
"ip": value.get("ip", ""),
"first_seen": record.get("first_seen", ""),
"last_seen": record.get("last_seen", ""),
"type": record.get("type", "a"),
})
print(f"[+] 被動 DNS 查詢 {domain}：{len(history)} 筆記錄")
return history
return []
def reverse_ip_lookup(self, ip_address):
"""尋找託管在特定 IP 位址上的所有網域。"""
headers = {"apikey": self.st_key}
url = f"https://api.securitytrails.com/v1/ips/nearby/{ip_address}"
resp = requests.get(url, headers=headers, timeout=30)
if resp.status_code == 200:
blocks = resp.json().get("blocks", [])
domains = []
for block in blocks:
for site in block.get("sites", []):
domains.append(site)
print(f"[+] 反向 IP 查詢 {ip_address}：{len(domains)} 個網域")
return domains
return []
def whois_lookup(self, domain):
"""取得 WHOIS 註冊資料以進行樞紐分析。"""
headers = {"apikey": self.st_key}
url = f"https://api.securitytrails.com/v1/domain/{domain}/whois"
resp = requests.get(url, headers=headers, timeout=30)
if resp.status_code == 200:
data = resp.json()
whois_data = {
"domain": domain,
"registrar": data.get("registrar", ""),
"registrant_org": data.get("registrant_org", ""),
"registrant_email": data.get("registrant_email", ""),
"name_servers": data.get("nameServers", []),
"created_date": data.get("createdDate", ""),
"updated_date": data.get("updatedDate", ""),
"expires_date": data.get("expiresDate", ""),
}
return whois_data
return {}
def pivot_from_seed(self, seed_indicator, indicator_type="domain", depth=2):
"""從種子指標遞迴樞紐分析以發現基礎設施。"""
discovered = {"domains": set(), "ips": set(), "relationships": []}
if indicator_type == "domain":
discovered["domains"].add(seed_indicator)
# 取得網域的 IP
pdns = self.passive_dns_lookup(seed_indicator)
for record in pdns:
ip = record["ip"]
discovered["ips"].add(ip)
discovered["relationships"].append({
"source": seed_indicator, "target": ip,
"type": "resolves_to",
"first_seen": record["first_seen"],
"last_seen": record["last_seen"],
})
if depth > 1:
# 對發現的 IP 進行反向查詢
reverse_domains = self.reverse_ip_lookup(ip)
for rd in reverse_domains[:20]:
discovered["domains"].add(rd)
discovered["relationships"].append({
"source": rd, "target": ip,
"type": "hosted_on",
})
elif indicator_type == "ip":
discovered["ips"].add(seed_indicator)
domains = self.reverse_ip_lookup(seed_indicator)
for domain in domains[:20]:
discovered["domains"].add(domain)
discovered["relationships"].append({
"source": domain, "target": seed_indicator,
"type": "hosted_on",
})
print(f"[+] 從 {seed_indicator} 樞紐分析結果："
f"{len(discovered['domains'])} 個網域，"
f"{len(discovered['ips'])} 個 IP，"
f"{len(discovered['relationships'])} 個關係")
return discovered
tracker = InfrastructureTracker(
securitytrails_key="YOUR_ST_KEY",
vt_key="YOUR_VT_KEY",
)
步驟 2：建構基礎設施圖
import networkx as nx
class InfrastructureGraph:
def __init__(self):
self.graph = nx.Graph()
def add_discovery(self, discovery_data):
"""將發現的基礎設施加入圖中。"""
for domain in discovery_data["domains"]:
self.graph.add_node(domain, type="domain")
for ip in discovery_data["ips"]:
self.graph.add_node(ip, type="ip")
for rel in discovery_data["relationships"]:
self.graph.add_edge(
rel["source"], rel["target"],
relationship=rel["type"],
first_seen=rel.get("first_seen", ""),
last_seen=rel.get("last_seen", ""),
)
def find_clusters(self):
"""識別基礎設施叢集。"""
components = list(nx.connected_components(self.graph))
clusters = []
for component in components:
domains = [n for n in component if self.graph.nodes[n].get("type") == "domain"]
ips = [n for n in component if self.graph.nodes[n].get("type") == "ip"]
clusters.append({
"size": len(component),
"domains": sorted(domains),
"ips": sorted(ips),
"domain_count": len(domains),
"ip_count": len(ips),
})
clusters.sort(key=lambda x: x["size"], reverse=True)
print(f"[+] 基礎設施叢集：{len(clusters)} 個")
return clusters
def find_hub_nodes(self, top_n=10):
"""尋找高中心性節點（共享基礎設施）。"""
centrality = nx.degree_centrality(self.graph)
top_nodes = sorted(centrality.items(), key=lambda x: x[1], reverse=True)[:top_n]
hubs = []
for node, score in top_nodes:
hubs.append({
"node": node,
"type": self.graph.nodes[node].get("type", "unknown"),
"centrality": round(score, 4),
"connections": self.graph.degree(node),
})
return hubs
def export_graph(self, output_file="infrastructure_graph.json"):
data = nx.node_link_data(self.graph)
with open(output_file, "w") as f:
json.dump(data, f, indent=2)
print(f"[+] 圖表已匯出：{self.graph.number_of_nodes()} 個節點，"
f"{self.graph.number_of_edges()} 條邊")
infra_graph = InfrastructureGraph()
discovery = tracker.pivot_from_seed("evil-domain.com", depth=2)
infra_graph.add_discovery(discovery)
clusters = infra_graph.find_clusters()
hubs = infra_graph.find_hub_nodes()
infra_graph.export_graph()
步驟 3：監控新基礎設施
import time
class InfrastructureMonitor:
def __init__(self, tracker, known_indicators):
self.tracker = tracker
self.known = set(known_indicators)
self.alerts = []
def check_new_registrations(self, patterns):
"""檢查符合攻擊者模式的新註冊網域。"""
import re
new_domains = []
for pattern in patterns:
# 查詢 SecurityTrails 以取得符合模式的新網域
headers = {"apikey": self.tracker.st_key}
url = "https://api.securitytrails.com/v1/domains/list"
params = {"include_ips": "true", "page": 1}
body = {"filter": {"keyword": pattern}}
resp = requests.post(url, headers=headers, json=body, timeout=30)
if resp.status_code == 200:
records = resp.json().get("records", [])
for record in records:
domain = record.get("hostname", "")
if domain not in self.known:
new_domains.append({
"domain": domain,
"pattern_matched": pattern,
"first_seen": datetime.now().isoformat(),
})
self.known.add(domain)
if new_domains:
print(f"[警報] {len(new_domains)} 個符合模式的新網域")
self.alerts.extend(new_domains)
return new_domains
def generate_infrastructure_report(self, clusters, hubs):
report = f"""# 攻擊者基礎設施追蹤報告
產生時間：{datetime.now().isoformat()}
## 摘要
- 已識別基礎設施叢集數量：{len(clusters)}
- 追蹤中網域總數：{sum(c['domain_count'] for c in clusters)}
- 追蹤中 IP 總數：{sum(c['ip_count'] for c in clusters)}
- 偵測到的新網域：{len(self.alerts)}
## 主要基礎設施樞紐
| 節點 | 類型 | 連線數 | 中心性 |
|------|------|--------|--------|
"""
for hub in hubs[:10]:
report += (f"| {hub['node']} | {hub['type']} "
f"| {hub['connections']} | {hub['centrality']} |\n")
report += "\n## 基礎設施叢集\n"
for i, cluster in enumerate(clusters[:5], 1):
report += f"\n### 叢集 {i}（{cluster['size']} 個節點）\n"
report += f"- 網域：{', '.join(cluster['domains'][:5])}\n"
report += f"- IP：{', '.join(cluster['ips'][:5])}\n"
with open("infrastructure_report.md", "w") as f:
f.write(report)
print("[+] 基礎設施報告已儲存")
monitor = InfrastructureMonitor(tracker, known_indicators=set())
驗證標準
- 被動 DNS 查詢能成功回傳歷史解析資料
- 反向 IP 查詢能發現共同託管的網域
- 基礎設施樞紐分析能從種子指標進行擴展
- 圖形分析能識別叢集和樞紐節點
- 透過模式監控能偵測到新的基礎設施
- 能產生包含可操作建議的報告