building-adversary-infrastructure-tracking-system

name	building-adversary-infrastructure-tracking-system
description	构建自动化系统，利用被动 DNS、证书透明度、WHOIS 数据和 IP 富化来映射和监控威胁行为者的命令与控制（C2）网络，追踪对手基础设施。
domain	cybersecurity
subdomain	threat-intelligence
tags	["infrastructure-tracking","passive-dns","c2","whois","threat-actor","pivoting","threat-intelligence","domain-analysis"]
version	1.0
author	mahipal
license	Apache-2.0

构建对手基础设施追踪系统

概述

对手基础设施追踪利用被动 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(, ),
                        : record.get(, ),
                        : record.get(, ),
                    })
            ()
             history
         []

     ():
        
        headers = {: .st_key}
        url = 
        resp = requests.get(url, headers=headers, timeout=)
         resp.status_code == :
            blocks = resp.json().get(, [])
            domains = []
             block  blocks:
                 site  block.get(, []):
                    domains.append(site)
            ()
             domains
         []

     ():
        
        headers = {: .st_key}
        url = 
        resp = requests.get(url, headers=headers, timeout=)
         resp.status_code == :
            data = resp.json()
            whois_data = {
                : domain,
                : data.get(, ),
                : data.get(, ),
                : data.get(, ),
                : data.get(, []),
                : data.get(, ),
                : data.get(, ),
                : data.get(, ),
            }
             whois_data
         {}

     ():
        
        discovered = {: (), : (), : []}

         indicator_type == :
            discovered[].add(seed_indicator)
            
            pdns = .passive_dns_lookup(seed_indicator)
             record  pdns:
                ip = record[]
                discovered[].add(ip)
                discovered[].append({
                    : seed_indicator, : ip,
                    : ,
                    : record[],
                    : record[],
                })

                 depth > :
                    
                    reverse_domains = .reverse_ip_lookup(ip)
                     rd  reverse_domains[:]:
                        discovered[].add(rd)
                        discovered[].append({
                            : rd, : ip,
                            : ,
                        })

         indicator_type == :
            discovered[].add(seed_indicator)
            domains = .reverse_ip_lookup(seed_indicator)
             domain  domains[:]:
                discovered[].add(domain)
                discovered[].append({
                    : domain, : seed_indicator,
                    : ,
                })

        (
              
              
              )
         discovered

tracker = InfrastructureTracker(
    securitytrails_key=,
    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 .graph.nodes[n].get() == ]
            ips = [n  n  component  .graph.nodes[n].get() == ]
            clusters.append({
                : (component),
                : (domains),
                : (ips),
                : (domains),
                : (ips),
            })
        clusters.sort(key= x: x[], reverse=)
        ()
         clusters

     ():
        
        centrality = nx.degree_centrality(.graph)
        top_nodes = (centrality.items(), key= x: x[], reverse=)[:top_n]
        hubs = []
         node, score  top_nodes:
            hubs.append({
                : node,
                : .graph.nodes[node].get(, ),
                : (score, ),
                : .graph.degree(node),
            })
         hubs

     ():
        data = nx.node_link_data(.graph)
         (output_file, )  f:
            json.dump(data, f, indent=)
        (
              )

infra_graph = InfrastructureGraph()
discovery = tracker.pivot_from_seed(, depth=)
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(),
                        })
                        .known.add(domain)

         new_domains:
            ()
            .alerts.extend(new_domains)
         new_domains

     ():
        report = 
         hub  hubs[:]:
            report += (
                       )

        report += 
         i, cluster  (clusters[:], ):
            report += 
            report += 
            report += 

         (, )  f:
            f.write(report)
        ()

monitor = InfrastructureMonitor(tracker, known_indicators=())

验收标准

• 被动 DNS 查询返回历史解析数据
• 反向 IP 查询发现共同托管域名
• 基础设施枢纽从种子指标扩展
• 图分析识别集群和枢纽节点
• 通过模式监控检测到新的基础设施
• 生成包含可执行建议的报告

参考资料

• Juniper: Threat Hunting with Passive DNS
• Censys: Advanced Persistent Infrastructure Tracking
• Embee Research: Malware Infrastructure with DNS Pivoting
• Validin: Passive DNS Threat Hunting
• SecurityTrails API
• Hunt.io: Uncovering Malicious Infrastructure