performing-ip-reputation-analysis-with-shodan

name	performing-ip-reputation-analysis-with-shodan
description	使用 Shodan API 分析 IP 地址声誉，识别开放端口、运行服务、已知漏洞和托管上下文，用于威胁情报富化和事件分类。
domain	cybersecurity
subdomain	threat-intelligence
tags	["shodan","ip-reputation","enrichment","threat-intelligence","reconnaissance","vulnerability","api","internet-scanning"]
version	1.0
author	mahipal
license	Apache-2.0

使用 Shodan 执行 IP 声誉分析

概述

Shodan 是全球首个互联网设备搜索引擎，持续扫描 IPv4 和 IPv6 地址空间，记录开放端口、运行服务、SSL 证书和已知漏洞。本技能涵盖使用 Shodan API 和 InternetDB 免费 API 富化来自安全告警的 IP 地址、根据暴露服务和漏洞评估威胁级别、识别托管基础设施模式，以及将 IP 声誉数据集成到 SOC 分类和威胁情报工作流中。

前置条件

• Python 3.9+，安装 shodan 库（pip install shodan）
• Shodan API 密钥（免费版：查询受限；付费版支持更高限额和流式传输）
• 了解 TCP/UDP 端口、常见服务和 CVE 标识符
• 熟悉 ASN、CIDR 表示法和 IP 地理定位概念
• 具备解读扫描结果的网络安全知识

核心概念

Shodan 数据模型

Shodan 中的每条 IP 记录包含：开放端口和协议、Banner 数据（服务响应）、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 已初始化。剩余扫描积分：{self.info.get('scan_credits', 0)}")

    def enrich_ip(self, ip_address):
        """通过 Shodan 对 IP 地址进行完整富化。"""
        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(, ),
                : host.get(, []),
                : host.get(, []),
                : host.get(, []),
                : host.get(, []),
                : host.get(, []),
                : host.get(, ),
                : [],
            }

             service  host.get(, []):
                svc = {
                    : service.get(, ),
                    : service.get(, ),
                    : service.get(, ),
                    : service.get(, ),
                    : service.get(, {}).get(, ),
                    : service.get(, )[:],
                }
                   service:
                    svc[] = service[].get(, {}).get(, {})
                    svc[] = service[].get(, {}).get(, {})
                    svc[] = service[].get(, {}).get(, )
                enrichment[].append(svc)

            
            enrichment[] = ._calculate_reputation(enrichment)
            (
                  
                  )
             enrichment

         shodan.APIError  e:
            ()
             

     ():
        
        score = 
        factors = []

        
        vuln_count = (data.get(, []))
         vuln_count > :
            score += 
            factors.append()
         vuln_count > :
            score += 
            factors.append()
         vuln_count > :
            score += 
            factors.append()

        
        suspicious_ports = {, , , , , , ,
                           , , , , , }
        open_ports = (data.get(, []))
        sus_found = open_ports.intersection(suspicious_ports)
         sus_found:
            score += 
            factors.append()

        
        malicious_tags = {, , , , }
        tags = (data.get(, []))
        mal_tags = tags.intersection(malicious_tags)
         mal_tags:
            score += 
            factors.append()

        
        port_count = (data.get(, []))
         port_count > :
            score += 
            factors.append()

        level = (
              score >= 
               score >= 
               score >= 
             
        )

         {: score, : level, : factors}

     ():
        
         requests
        resp = requests.get(, timeout=)
         resp.status_code == :
            data = resp.json()
            (
                  
                  )
             data
         

enricher = ShodanEnricher()
result = enricher.enrich_ip()
(json.dumps(result, indent=, default=))

步骤 2：批量 IP 声誉检查

import time

def batch_ip_reputation(enricher, ip_list, output_file="ip_reputation.json"):
    """检查 IP 地址列表的声誉。"""
    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)}] 已处理")
            time.sleep(1)  # 速率限制

    # 按声誉评分降序排列（最高风险在前）
    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)

    # 汇总
    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=== 批量声誉汇总 ===")
    ()
     level, count  levels.items():
        ()

     results

suspicious_ips = [, , ]
results = batch_ip_reputation(enricher, suspicious_ips)

步骤 3：基础设施关联

def correlate_infrastructure(enricher, ip_address):
    """根据共享属性发现相关基础设施。"""
    host_data = enricher.enrich_ip(ip_address)
    if not host_data:
        return {}

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

    # 搜索同一组织
    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

    # 搜索相同 SSL 证书
    for service in host_data.get("services", []):
        ssl_subject = service.get("ssl_subject", {})
        if ssl_subject:
            cn = ssl_subject.get("CN", "")
            if cn:
                :
                    results = enricher.api.search(, limit=)
                       results.get(, []):
                        correlations[].append({
                            : .get(, ),
                            : cn,
                        })
                 shodan.APIError:
                    

    ()
    ()
    ()
     correlations

验收标准

• Shodan API 经正确身份验证后查询成功
• IP 富化返回端口、服务、漏洞和地理定位信息
• 声誉评分按威胁级别对 IP 进行分类
• 批量富化正确处理速率限制
• 基础设施关联识别相关主机
• InternetDB 免费 API 用于高容量查询

参考资料

• Shodan 开发者 API
• Shodan InternetDB API
• Shodan Python 库
• Query.ai：利用 Shodan 进行安全研究
• Torq：Shodan IP 富化工作流
• Recorded Future：Shodan 集成