敵対者インフラストラクチャ追跡システムの構築

敵対者インフラストラクチャ追跡システムの構築 は何をしますか？

概要

敵対者インフラストラクチャ追跡では、受動的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）、好むホスティング（ bulletproof ホスティングプロバイダー、クラウドサービス）、ドメイン生成アルゴリズム（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):
        """Query passive DNS for domain resolution history."""
        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"[+] Passive DNS for {domain}: {len(history)} records")
            return history
        return []

    def reverse_ip_lookup(self, ip_address):
        """Find all domains hosted on an IP address."""
        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"[+] Reverse IP for {ip_address}: {len(domains)} domains")
            return domains
        return []

    def whois_lookup(self, domain):
        """Get WHOIS registration data for pivoting."""
        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):
        """Recursively pivot from a seed indicator to discover infrastructure."""
        discovered = {"domains": set(), "ips": set(), "relationships": []}

        if indicator_type == "domain":
            discovered["domains"].add(seed_indicator)
            # Get IPs for domain
            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:
                    # Reverse lookup on discovered IPs
                    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"[+] Pivot from {seed_indicator}: "
              f"{len(discovered['domains'])} domains, "
              f"{len(discovered['ips'])} IPs, "
              f"{len(discovered['relationships'])} 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):
        """Add discovered infrastructure to graph."""
        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):
        """Identify infrastructure clusters."""
        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"[+] Infrastructure clusters: {len(clusters)}")
        return clusters

    def find_hub_nodes(self, top_n=10):
        """Find high-centrality nodes (shared infrastructure)."""
        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"[+] Graph exported: {self.graph.number_of_nodes()} nodes, "
              f"{self.graph.number_of_edges()} 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):
        """Check for newly registered domains matching adversary patterns."""
        import re
        new_domains = []
        for pattern in patterns:
            # Query SecurityTrails for new domains matching pattern
            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"[ALERT] {len(new_domains)} new domains matching patterns")
            self.alerts.extend(new_domains)
        return new_domains

    def generate_infrastructure_report(self, clusters, hubs):
        report = f"""# Adversary Infrastructure Tracking Report
Generated: {datetime.now().isoformat()}

## Summary
- Infrastructure clusters identified: {len(clusters)}
- Total domains tracked: {sum(c['domain_count'] for c in clusters)}
- Total IPs tracked: {sum(c['ip_count'] for c in clusters)}
- New domains detected: {len(self.alerts)}

## Top Infrastructure Hubs
| Node | Type | Connections | Centrality |
|------|------|-------------|------------|
"""
        for hub in hubs[:10]:
            report += (f"| {hub['node']} | {hub['type']} "
                       f"| {hub['connections']} | {hub['centrality']} |\n")

        report += "\n## Infrastructure Clusters\n"
        for i, cluster in enumerate(clusters[:5], 1):
            report += f"\n### Cluster {i} ({cluster['size']} nodes)\n"
            report += f"- Domains: {', '.join(cluster['domains'][:5])}\n"
            report += f"- IPs: {', '.join(cluster['ips'][:5])}\n"

        with open("infrastructure_report.md", "w") as f:
            f.write(report)
        print("[+] Infrastructure report saved")

monitor = InfrastructureMonitor(tracker, known_indicators=set())

検証基準

• 受動的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