Install this skill

npx skills add mukul975/Anthropic-Cybersecurity-Skills

Framework mappings

MITRE ATT&CK

NIST CSF 2.0

Overview

Shodan is the world's first search engine for internet-connected devices, continuously scanning the IPv4 and IPv6 address space to catalog open ports, running services, SSL certificates, and known vulnerabilities. This skill covers using the Shodan API and InternetDB free API to enrich IP addresses from security alerts, assess threat levels based on exposed services and vulnerabilities, identify hosting infrastructure patterns, and integrate IP reputation data into SOC triage and threat intelligence workflows.

When to Use

When conducting security assessments that involve performing ip reputation analysis with shodan
When following incident response procedures for related security events
When performing scheduled security testing or auditing activities
When validating security controls through hands-on testing

Prerequisites

Python 3.9+ with shodan library (pip install shodan)
Shodan API key (free tier: limited queries; paid plans for higher limits and streaming)
Understanding of TCP/UDP ports, common services, and CVE identifiers
Familiarity with ASN, CIDR notation, and IP geolocation concepts
Network security knowledge for interpreting scan results

Key Concepts

Shodan Data Model

Each IP record in Shodan contains: open ports and protocols, banner data (service responses), SSL/TLS certificate details, known CVE vulnerabilities, hostname(s) and reverse DNS, ASN and ISP information, geographic location, operating system fingerprint, and historical scan data showing changes over time.

InternetDB API

Shodan's free InternetDB API (internetdb.shodan.io) provides quick IP lookups without authentication, returning open ports, hostnames, tags, CPEs, and known vulnerabilities. This is useful for high-volume enrichment where the full Shodan API would hit rate limits.

Reputation Scoring

IP reputation is assessed by combining: number and type of open ports (unusual ports indicate compromise), vulnerable services (unpatched software with known CVEs), hosting type (residential, cloud, VPN/proxy, bulletproof hosting), historical activity (past associations with malware, scanning, spam), and geographic context (countries known for specific threat activity).

Workflow

Step 1: Basic IP Enrichment with Shodan API

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))

Step 2: Batch IP Reputation Check

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)

Step 3: Infrastructure Correlation

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

Validation Criteria

Shodan API queried successfully with proper authentication
IP enrichment returns ports, services, vulnerabilities, and geolocation
Reputation scoring classifies IPs by threat level
Batch enrichment handles rate limiting correctly
Infrastructure correlation identifies related hosts
InternetDB free API used for high-volume lookups

References

Source materials

References and resources

Everything below is rendered for inspection. Script files are read-only and never run.

References 1

api-reference.md1.1 KB

API Reference — Performing IP Reputation Analysis with Shodan

Libraries Used

shodan: Shodan API client for IP host lookup, port scanning, vulnerability data
requests: HTTP client for AbuseIPDB API reputation checks

CLI Interface

python agent.py --shodan-key <key> [--abuseipdb-key <key>] lookup --ip <ip>
python agent.py --shodan-key <key> [--abuseipdb-key <key>] bulk --ips <ip1> <ip2>

Core Functions

`shodan_host_lookup(api_key, ip)` — Shodan host details

shodan.Shodan(api_key) / api.host(ip)
Returns: org, ASN, ISP, country, open ports, vulns, services

`abuseipdb_check(api_key, ip, max_age_days)` — AbuseIPDB reputation

GET https://api.abuseipdb.com/api/v2/check
Returns: abuse_confidence (0-100), total_reports, is_tor

`bulk_reputation(shodan_key, ips, abuseipdb_key)` — Multi-IP analysis with risk scoring

Risk Classification

Risk	Criteria
Critical	Abuse score >= 80 or 3+ known vulns
High	Abuse score >= 50 or 1+ known vulns
Medium	Abuse score >= 20
Low	Below thresholds

Dependencies

pip install shodan requests

Scripts 1

agent.py4.2 KB

Display-only source. This catalog never executes bundled scripts.

#!/usr/bin/env python3
"""Agent for performing IP reputation analysis using the Shodan API."""

import json
import argparse
from datetime import datetime

try:
    import shodan
    HAS_SHODAN = True
except ImportError:
    HAS_SHODAN = False

try:
    import requests
    HAS_REQUESTS = True
except ImportError:
    HAS_REQUESTS = False

ABUSEIPDB_URL = "https://api.abuseipdb.com/api/v2/check"


def shodan_host_lookup(api_key, ip):
    """Look up IP details from Shodan."""
    api = shodan.Shodan(api_key)
    result = api.host(ip)
    return {
        "ip": result.get("ip_str"),
        "org": result.get("org"),
        "asn": result.get("asn"),
        "isp": result.get("isp"),
        "os": result.get("os"),
        "country": result.get("country_code"),
        "city": result.get("city"),
        "open_ports": result.get("ports", []),
        "vulns": result.get("vulns", []),
        "hostnames": result.get("hostnames", []),
        "last_update": result.get("last_update"),
        "services": [
            {"port": s.get("port"), "transport": s.get("transport"),
             "product": s.get("product"), "version": s.get("version")}
            for s in result.get("data", [])
        ][:20],
    }


def abuseipdb_check(api_key, ip, max_age_days=90):
    """Check IP reputation on AbuseIPDB."""
    resp = requests.get(ABUSEIPDB_URL, headers={"Key": api_key, "Accept": "application/json"},
                        params={"ipAddress": ip, "maxAgeInDays": max_age_days}, timeout=15)
    resp.raise_for_status()
    data = resp.json().get("data", {})
    return {
        "ip": data.get("ipAddress"),
        "abuse_confidence": data.get("abuseConfidenceScore"),
        "total_reports": data.get("totalReports"),
        "country": data.get("countryCode"),
        "isp": data.get("isp"),
        "domain": data.get("domain"),
        "is_tor": data.get("isTor"),
        "is_whitelisted": data.get("isWhitelisted"),
        "last_reported": data.get("lastReportedAt"),
    }


def bulk_reputation(shodan_key, ips, abuseipdb_key=None):
    """Analyze reputation of multiple IPs."""
    results = []
    for ip in ips:
        ip = ip.strip()
        if not ip:
            continue
        entry = {"ip": ip}
        try:
            entry["shodan"] = shodan_host_lookup(shodan_key, ip)
        except Exception as e:
            entry["shodan"] = {"error": str(e)}
        if abuseipdb_key:
            try:
                entry["abuseipdb"] = abuseipdb_check(abuseipdb_key, ip)
            except Exception as e:
                entry["abuseipdb"] = {"error": str(e)}
        risk = "low"
        abuse_score = entry.get("abuseipdb", {}).get("abuse_confidence", 0) or 0
        vulns = entry.get("shodan", {}).get("vulns", [])
        if abuse_score >= 80 or len(vulns) >= 3:
            risk = "critical"
        elif abuse_score >= 50 or len(vulns) >= 1:
            risk = "high"
        elif abuse_score >= 20:
            risk = "medium"
        entry["risk"] = risk
        results.append(entry)
    return {"timestamp": datetime.utcnow().isoformat(), "total": len(results), "results": results}


def main():
    parser = argparse.ArgumentParser(description="IP Reputation Analysis Agent")
    parser.add_argument("--shodan-key", required=True, help="Shodan API key")
    parser.add_argument("--abuseipdb-key", help="AbuseIPDB API key")
    sub = parser.add_subparsers(dest="command")
    l = sub.add_parser("lookup", help="Look up single IP")
    l.add_argument("--ip", required=True)
    b = sub.add_parser("bulk", help="Analyze multiple IPs")
    b.add_argument("--ips", nargs="+", required=True)
    args = parser.parse_args()
    if not HAS_SHODAN:
        print(json.dumps({"error": "shodan library not installed"}))
        return
    if args.command == "lookup":
        result = {"shodan": shodan_host_lookup(args.shodan_key, args.ip)}
        if args.abuseipdb_key and HAS_REQUESTS:
            result["abuseipdb"] = abuseipdb_check(args.abuseipdb_key, args.ip)
    elif args.command == "bulk":
        result = bulk_reputation(args.shodan_key, args.ips, args.abuseipdb_key)
    else:
        parser.print_help()
        return
    print(json.dumps(result, indent=2, default=str))


if __name__ == "__main__":
    main()

Keep exploring