performing-ip-reputation-analysis-with-shodan

Shodanを使用したIPレピュテーション分析

概要

Shodanは、インターネットに接続されたデバイスを対象とする世界初の検索エンジンです。IPv4およびIPv6のアドレス空間を継続的にスキャンし、開いているポート、稼働中のサービス、SSL証明書、既知の脆弱性をカタログ化します。このスキルでは、Shodan APIとInternetDBの無料APIを使用して、セキュリティアラートに含まれるIPアドレスの情報を拡充し、公開サービスと脆弱性に基づいて脅威レベルを評価し、ホスティングインフラのパターンを特定し、IPレピュテーションデータをSOCのトリアージや脅威インテリジェンスのワークフローに組み込む方法を説明します。

使用する場面

• Shodanを使用したIPレピュテーション分析を伴うセキュリティ評価を実施する場合
• 関連するセキュリティイベントに対するインシデント対応手順に従う場合
• 定期的なセキュリティテストや監査活動を実施する場合
• 実地テストを通じてセキュリティ制御を検証する場合

前提条件

• shodanライブラリをインストールしたPython 3.9以降（pip install shodan）
• Shodan APIキー（無料プランではクエリ数に制限があります。より高い上限やストリーミングには有料プランが必要です）
• TCP/UDPポート、一般的なサービス、CVE識別子に関する知識
• ASN、CIDR表記、IPジオロケーションの概念に関する知識
• スキャン結果を解釈するためのネットワークセキュリティの知識

主な概念

Shodanのデータモデル

Shodanの各IPレコードには、開いているポートとプロトコル、バナーデータ（サービス応答）、SSL/TLS証明書の詳細、既知のCVE脆弱性、ホスト名と逆引きDNS、ASNとISPの情報、地理的位置、オペレーティングシステムのフィンガープリント、経時的な変化を示す過去のスキャンデータが含まれます。

InternetDB API

Shodanの無料InternetDB API（internetdb.shodan.io）は、認証なしでIPアドレスをすばやく検索でき、開いているポート、ホスト名、タグ、CPE、既知の脆弱性を返します。これは、Shodan APIのレート制限に達する可能性がある大量の情報拡充に便利です。

レピュテーションスコアリング

IPレピュテーションは、開いているポートの数と種類（通常とは異なるポートは侵害を示す場合があります）、脆弱なサービス（既知のCVEがある未パッチのソフトウェア）、ホスティングの種類（住宅向け、クラウド、VPN／プロキシ、弾丸ホスティング）、過去の活動（マルウェア、スキャン、スパムとの過去の関連）、地理的な状況（特定の脅威活動で知られる国）を組み合わせて評価します。

ワークフロー

ステップ 1：Shodan APIによる基本的なIP情報の拡充

import shodan
import json
from datetime import datetime

class ShodanEnricher:
    def __init__(self, api_key):
        self.api = shodan.Shodan(api_key)
        self.info = self.api.info()
        print(f"[+] Shodan API initialized. Credits: {self.info.get('scan_credits', 0)}")

    def enrich_ip(self, ip_address):
        """Full enrichment of an IP address via Shodan."""
        try:
            host = self.api.host(ip_address)
            enrichment = {
                "ip": ip_address,
                "organization": host.get("org", ""),
                "asn": host.get("asn", ""),
                "isp": host.get("isp", ""),
                "country": host.get("country_name", ""),
                "country_code": host.get("country_code", ""),
                "city": host.get("city", ""),
                "latitude": host.get("latitude"),
                "longitude": host.get("longitude"),
                "os": host.get("os", ""),
                "ports": host.get("ports", []),
                "hostnames": host.get("hostnames", []),
                "domains": host.get("domains", []),
                "vulns": host.get("vulns", []),
                "tags": host.get("tags", []),
                "last_update": host.get("last_update", ""),
                "services": [],
            }

            for service in host.get("data", []):
                svc = {
                    "port": service.get("port", 0),
                    "transport": service.get("transport", "tcp"),
                    "product": service.get("product", ""),
                    "version": service.get("version", ""),
                    "module": service.get("_shodan", {}).get("module", ""),
                    "banner": service.get("data", "")[:200],
                }
                if "ssl" in service:
                    svc["ssl_subject"] = service["ssl"].get("cert", {}).get("subject", {})
                    svc["ssl_issuer"] = service["ssl"].get("cert", {}).get("issuer", {})
                    svc["ssl_expires"] = service["ssl"].get("cert", {}).get("expires", "")
                enrichment["services"].append(svc)

            # Calculate reputation score
            enrichment["reputation"] = self._calculate_reputation(enrichment)
            print(f"[+] {ip_address}: {len(enrichment['ports'])} ports, "
                  f"{len(enrichment['vulns'])} vulns, "
                  f"reputation: {enrichment['reputation']['level']}")
            return enrichment

        except shodan.APIError as e:
            print(f"[-] Shodan error for {ip_address}: {e}")
            return None

    def _calculate_reputation(self, data):
        """Calculate IP reputation score based on Shodan data."""
        score = 0
        factors = []

        # Vulnerability assessment
        vuln_count = len(data.get("vulns", []))
        if vuln_count > 10:
            score += 40
            factors.append(f"{vuln_count} known vulnerabilities")
        elif vuln_count > 5:
            score += 25
            factors.append(f"{vuln_count} known vulnerabilities")
        elif vuln_count > 0:
            score += 10
            factors.append(f"{vuln_count} known vulnerabilities")

        # Suspicious port analysis
        suspicious_ports = {4444, 5555, 6666, 8888, 9090, 1234, 31337,
                           6667, 6697, 8080, 8443, 3128, 1080}
        open_ports = set(data.get("ports", []))
        sus_found = open_ports.intersection(suspicious_ports)
        if sus_found:
            score += 15
            factors.append(f"suspicious ports: {sus_found}")

        # Tag-based assessment
        malicious_tags = {"self-signed", "cloud", "vpn", "proxy", "tor"}
        tags = set(data.get("tags", []))
        mal_tags = tags.intersection(malicious_tags)
        if mal_tags:
            score += 10
            factors.append(f"tags: {mal_tags}")

        # Too many open ports
        port_count = len(data.get("ports", []))
        if port_count > 20:
            score += 15
            factors.append(f"excessive open ports ({port_count})")

        level = (
            "critical" if score >= 50
            else "high" if score >= 35
            else "medium" if score >= 15
            else "low"
        )

        return {"score": score, "level": level, "factors": factors}

    def enrich_ip_free(self, ip_address):
        """Quick IP enrichment using free InternetDB API."""
        import requests
        resp = requests.get(f"https://internetdb.shodan.io/{ip_address}", timeout=10)
        if resp.status_code == 200:
            data = resp.json()
            print(f"[+] InternetDB: {ip_address} -> "
                  f"{len(data.get('ports', []))} ports, "
                  f"{len(data.get('vulns', []))} vulns")
            return data
        return None

enricher = ShodanEnricher("YOUR_SHODAN_API_KEY")
result = enricher.enrich_ip("8.8.8.8")
print(json.dumps(result, indent=2, default=str))

ステップ 2：IPレピュテーションの一括チェック

import time

def batch_ip_reputation(enricher, ip_list, output_file="ip_reputation.json"):
    """Check reputation for a list of IP addresses."""
    results = []
    for i, ip in enumerate(ip_list):
        result = enricher.enrich_ip(ip)
        if result:
            results.append(result)
        if (i + 1) % 10 == 0:
            print(f"  [{i+1}/{len(ip_list)}] Processed")
            time.sleep(1)  # Rate limiting

    # Sort by reputation score (highest risk first)
    results.sort(key=lambda x: x.get("reputation", {}).get("score", 0), reverse=True)

    with open(output_file, "w") as f:
        json.dump(results, f, indent=2, default=str)

    # Summary
    levels = {"critical": 0, "high": 0, "medium": 0, "low": 0}
    for r in results:
        level = r.get("reputation", {}).get("level", "low")
        levels[level] += 1

    print(f"\n=== Batch Reputation Summary ===")
    print(f"Total IPs: {len(results)}")
    for level, count in levels.items():
        print(f"  {level.upper()}: {count}")

    return results

suspicious_ips = ["203.0.113.1", "198.51.100.5", "192.0.2.100"]
results = batch_ip_reputation(enricher, suspicious_ips)

ステップ 3：インフラの相関分析

def correlate_infrastructure(enricher, ip_address):
    """Find related infrastructure based on shared attributes."""
    host_data = enricher.enrich_ip(ip_address)
    if not host_data:
        return {}

    correlations = {
        "same_org": [],
        "same_asn": [],
        "shared_ssl": [],
    }

    # Search for same organization
    org = host_data.get("organization", "")
    if org:
        try:
            results = enricher.api.search(f'org:"{org}"', limit=20)
            for match in results.get("matches", []):
                correlations["same_org"].append({
                    "ip": match.get("ip_str", ""),
                    "port": match.get("port", 0),
                    "product": match.get("product", ""),
                })
        except shodan.APIError:
            pass

    # Search for same SSL certificate
    for service in host_data.get("services", []):
        ssl_subject = service.get("ssl_subject", {})
        if ssl_subject:
            cn = ssl_subject.get("CN", "")
            if cn:
                try:
                    results = enricher.api.search(f'ssl.cert.subject.CN:"{cn}"', limit=20)
                    for match in results.get("matches", []):
                        correlations["shared_ssl"].append({
                            "ip": match.get("ip_str", ""),
                            "cn": cn,
                        })
                except shodan.APIError:
                    pass

    print(f"[+] Infrastructure correlations for {ip_address}:")
    print(f"  Same org: {len(correlations['same_org'])} hosts")
    print(f"  Shared SSL: {len(correlations['shared_ssl'])} hosts")
    return correlations

検証基準

• 適切な認証を使用してShodan APIへのクエリが正常に実行される
• IP情報の拡充により、ポート、サービス、脆弱性、地理的位置が取得される
• レピュテーションスコアリングによって、IPが脅威レベル別に分類される
• 一括での情報拡充でレート制限が適切に処理される
• インフラの相関分析によって関連ホストが特定される
• 大量の検索にInternetDBの無料APIが使用される

参考資料

• Shodan開発者向けAPI
• Shodan InternetDB API
• Shodan Pythonライブラリ
• Query.ai：セキュリティ調査でのShodan活用
• Torq：ShodanによるIP情報拡充ワークフロー
• Recorded Future：Shodan連携