# Performing Ip Reputation Analysis With Shodan > 使用 Shodan API 分析 IP 地址声誉，识别开放端口、运行服务、已知漏洞和托管上下文，用于威胁情报富化和事件分类。 - Skill: `killvxk/performing-ip-reputation-analysis-with-shodan` (Agent Skill, multi-file: 4 files) - Install (CLI): `npx skillmds@latest add killvxk/performing-ip-reputation-analysis-with-shodan` - Raw SKILL.md: https://api.skillmd.com/api/skills/killvxk/performing-ip-reputation-analysis-with-shodan/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Integrations & APIs - License: Apache-2.0 - Author: killvxk (https://skillmd.com/u/killvxk) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/killvxk/performing-ip-reputation-analysis-with-shodan --- # 使用 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 富化 ```python 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("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) # 计算声誉评分 enrichment["reputation"] = self._calculate_reputation(enrichment) print(f"[+] {ip_address}：{len(enrichment['ports'])} 个端口，" f"{len(enrichment['vulns'])} 个漏洞，" f"声誉等级：{enrichment['reputation']['level']}") return enrichment except shodan.APIError as e: print(f"[-] {ip_address} 的 Shodan 查询出错：{e}") return None def _calculate_reputation(self, data): """根据 Shodan 数据计算 IP 声誉评分。""" score = 0 factors = [] # 漏洞评估 vuln_count = len(data.get("vulns", [])) if vuln_count > 10: score += 40 factors.append(f"{vuln_count} 个已知漏洞") elif vuln_count > 5: score += 25 factors.append(f"{vuln_count} 个已知漏洞") elif vuln_count > 0: score += 10 factors.append(f"{vuln_count} 个已知漏洞") # 可疑端口分析 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"可疑端口：{sus_found}") # 基于标签的评估 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"标签：{mal_tags}") # 开放端口过多 port_count = len(data.get("ports", [])) if port_count > 20: score += 15 factors.append(f"开放端口过多（{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): """使用免费 InternetDB API 快速富化 IP。""" 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', []))} 个端口，" f"{len(data.get('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 声誉检查 ```python 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=== 批量声誉汇总 ===") print(f"IP 总数：{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：基础设施关联 ```python 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: 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"[+] {ip_address} 的基础设施关联：") print(f" 同一组织：{len(correlations['same_org'])} 台主机") print(f" 共享 SSL：{len(correlations['shared_ssl'])} 台主机") return correlations ``` ## 验收标准 - Shodan API 经正确身份验证后查询成功 - IP 富化返回端口、服务、漏洞和地理定位信息 - 声誉评分按威胁级别对 IP 进行分类 - 批量富化正确处理速率限制 - 基础设施关联识别相关主机 - InternetDB 免费 API 用于高容量查询 ## 参考资料 - [Shodan 开发者 API](https://developer.shodan.io/api) - [Shodan InternetDB API](https://internetdb.shodan.io/) - [Shodan Python 库](https://github.com/achillean/shodan-python) - [Query.ai：利用 Shodan 进行安全研究](https://www.query.ai/resources/blogs/leveraging-shodan-for-security-research/) - [Torq：Shodan IP 富化工作流](https://kb.torq.io/en/articles/9350284-shodan-ip-address-enrichment-with-cache-workflow-template) - [Recorded Future：Shodan 集成](https://support.recordedfuture.com/hc/en-us/articles/115001403928-Shodan)