Monitoring Your Personal Attack Surface with Shodan and n8n

import socket
import requests


DYNDNS_DOMAIN = "dyndns.example.de"
SHODAN_API_KEY = 'abcdef12345'

try:
    # 2. Resolve the DynDNS Domain to an IP
    # We use socket to look up the 'A' record (IPv4) for your domain
    target_ip = socket.gethostbyname(DYNDNS_DOMAIN)

    # 3. Lookup IP on Shodan (Passive Scan)
    shodan_url = f"https://api.shodan.io/shodan/host/{target_ip}?key={SHODAN_API_KEY}"
    shodan_response = requests.get(shodan_url, timeout=10)
    
    # Check results
    if shodan_response.status_code == 200:
        scan_data = shodan_response.json()
        found = True
    elif shodan_response.status_code == 404:
        scan_data = {"message": "IP not found in Shodan database."}
        found = False
    else:
        scan_data = {"error": f"Shodan API Error: {shodan_response.status_code}"}
        found = False

    # 4. Return to n8n
    return [{
        "json": {
            "domain": DYNDNS_DOMAIN,
            "resolved_ip": target_ip,
            "shodan_found": found,
            "shodan_data": scan_data
        }
    }]

except socket.gaierror:
    # Handle DNS resolution errors (e.g., if the domain doesn't exist)
    return [{
        "json": {
            "error": "Could not resolve hostname. Check your DynDNS domain.",
            "domain": DYNDNS_DOMAIN
        }
    }]
except Exception as e:
    return [{
        "json": {
            "error": str(e)
        }
    }]

from datetime import datetime

# 1. Get the input data using your preferred syntax
data = _items[0]["json"]

domain = data.get("domain", "Unknown Domain")
ip = data.get("resolved_ip", "Unknown IP")
is_found = data.get("shodan_found", False)
shodan_data = data.get("shodan_data", {})

# 2. Initialize Embed Variables
embed_fields = []

if is_found:
    # --- STATUS: WARNING (Red/Orange) ---
    color = 16744192  # 0xFF7700 (Orange)
    title = f"⚠️ Shodan Alert: {domain}"
    description = f"**Target IP:** `{ip}`\n**Status:** Exposed services found."

    # Extract specific details from your JSON structure
    # Use 'get' with defaults to prevent crashes
    org = shodan_data.get("org", "Unknown Org")
    city = shodan_data.get("city", "Unknown City")
    country = shodan_data.get("country_name", "Unknown Country")
    
    # Extract Ports
    # In your JSON, 'ports' is a list [8089]
    ports = shodan_data.get("ports", [])
    ports_str = ", ".join(map(str, ports)) if ports else "None detected"

    # Extract Service Banners (from the 'data' list in your JSON)
    # This shows *what* is running on the port (e.g., HTTP 404)
    service_details = []
    for service in shodan_data.get("data", []):
        p = service.get("port")
        # Try to get a product name, otherwise grab the raw data banner
        banner = service.get("product") or service.get("data", "").strip()
        # Truncate banner if it's too long for Discord
        if len(banner) > 50: 
            banner = banner[:47] + "..."
        service_details.append(f"**Port {p}:** `{banner}`")
    
    services_str = "\n".join(service_details) if service_details else "No service banners captured."

    # Build the Fields
    embed_fields = [
        {"name": "🏢 Organization", "value": org, "inline": True},
        {"name": "🌍 Location", "value": f"{city}, {country}", "inline": True},
        {"name": "🔓 Open Ports", "value": ports_str, "inline": True},
        {"name": "🔎 Service Details", "value": services_str, "inline": False}
    ]

else:
    # --- STATUS: SAFE (Green) ---
    color = 5763719  # 0x57F287 (Green)
    title = f"✅ Clean Scan: {domain}"
    description = f"**Target IP:** `{ip}`\n**Status:** No open ports found on Shodan."
    embed_fields = [
        {"name": "Result", "value": "IP address is not indexed or has no public services.", "inline": False}
    ]

# 3. Construct the Final Discord Webhook Payload
discord_payload = {
    "username": "Shodan Bot",
    "avatar_url": "https://static-00.iconduck.com/assets.00/shodan-icon-512x512-j6i1s0d4.png",
    "embeds": [{
        "title": title,
        "description": description,
        "color": color,
        "fields": embed_fields,
        "footer": {
            "text": f"Scan Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
        }
    }]
}

# 4. Return formatted for the next node
return [{"json": discord_payload}]

# n8n Code Node: Post to Discord
import requests

DISCORD_WEBHOOK_URL = "https://discord.com/api/webhooks/"

# Get the input data (which is the Discord payload itself)
payload = _items[0]["json"]

try:
    resp = requests.post(
        DISCORD_WEBHOOK_URL,
        json=payload,
        timeout=(5, 10),
    )
    resp.raise_for_status()
    
    # Return success status
    return [{
        "json": {
            "status": resp.status_code,
            "sent": True,
            "target": payload.get('embeds', [{}])[0].get('title', 'Unknown Target')
        }
    }]

except Exception as e:
    # Return error details
    return [{
        "json": {
            "error": str(e),
            "sent": False,
            "failed_payload": payload
        }
    }]