Shodan and jobs

cloud/APIs/shodan.py

@@ -0,0 +1,134 @@
+import os
+import requests
+from typing import Dict, Any, Optional
+
+
+class ShodanAuditor:
+    """Service class to handle Shodan API security audits and IP profiling."""
+
+    def __init__(self, api_key: Optional[str] = None):
+        self.api_key = api_key or os.getenv("SHODAN_API_KEY", "")
+        self.base_url = "https://api.shodan.io"
+
+    def get_public_ip(self) -> Optional[str]:
+        """Utility method to discover the server's public IPv4 address."""
+        try:
+            res = requests.get("https://api.ipify.org?format=json", timeout=5)
+            return res.json().get("ip")
+        except Exception:
+            return None
+
+    def _query_host(self, ip: str) -> Dict[str, Any]:
+        """Internal helper to query Shodan's Host API endpoint."""
+        if not self.api_key:
+            return {"error": "Missing Shodan API key"}
+
+        url = f"{self.base_url}/shodan/host/{ip}?key={self.api_key}"
+        try:
+            response = requests.get(url, timeout=8)
+            if response.status_code == 404:
+                return {"indexed": False, "ip": ip}
+            if response.status_code != 200:
+                return {"error": f"Shodan API error HTTP {response.status_code}"}
+            
+            data = response.json()
+            data["indexed"] = True
+            return data
+        except requests.RequestException as e:
+            return {"error": str(e)}
+
+    def audit_server_vulnerabilities(self, target_ip: Optional[str] = None) -> Dict[str, Any]:
+        """
+        Audits the server for reported CVEs, exposed services, and SSL details.
+        
+        :param target_ip: IP to scan (defaults to server's own public IP)
+        """
+        ip = target_ip or self.get_public_ip()
+        if not ip:
+            return {"status": "error", "message": "Unable to resolve target IP"}
+
+        data = self._query_host(ip)
+
+        if "error" in data:
+            return {"status": "error", "message": data["error"], "target_ip": ip}
+
+        if not data.get("indexed"):
+            return {
+                "status": "PASS",
+                "target_ip": ip,
+                "message": "IP is not indexed by Shodan (low public exposure)",
+                "vulnerabilities": [],
+                "open_ports": []
+            }
+
+        # Extract CVEs
+        vulns = data.get("vulns", [])
+        
+        # Extract SSL/TLS info from HTTP services if available
+        ssl_info = []
+        for service in data.get("data", []):
+            if "ssl" in service:
+                ssl_details = service["ssl"]
+                ssl_info.append({
+                    "port": service.get("port"),
+                    "cert_expired": ssl_details.get("cert", {}).get("expired", False),
+                    "cipher": ssl_details.get("cipher", {}).get("name"),
+                    "version": ssl_details.get("versions")
+                })
+
+        return {
+            "status": "FAIL" if vulns else "PASS",
+            "target_ip": ip,
+            "hostnames": data.get("hostnames", []),
+            "open_ports": data.get("ports", []),
+            "vulnerabilities": vulns,
+            "vuln_count": len(vulns),
+            "ssl_certificates": ssl_info,
+            "last_shodan_update": data.get("last_update")
+        }
+
+    def profile_client_ip(self, client_ip: str) -> Dict[str, Any]:
+        """
+        Profiles a incoming client (microwave) public IP address.
+        Checks if the client network has exposed admin panels, Telnet, or CVEs.
+        
+        :param client_ip: Public IP of the connecting device/router
+        """
+        # Skip local/private network IPs
+        if client_ip in ["127.0.0.1", "localhost"] or client_ip.startswith(("10.", "192.168.", "172.16.")):
+            return {"status": "skipped", "reason": "Local/Private IP"}
+
+        data = self._query_host(client_ip)
+
+        if "error" in data or not data.get("indexed"):
+            return {
+                "client_ip": client_ip,
+                "risk_level": "LOW",
+                "indexed": False,
+                "summary": "Client IP not indexed on Shodan"
+            }
+
+        ports = data.get("ports", [])
+        vulns = data.get("vulns", [])
+
+        # Check for high-risk exposed router/device ports
+        high_risk_ports = [23, 80, 8080, 8443, 7547]  # Telnet, Web UI, TR-069
+        flagged_ports = [p for p in ports if p in high_risk_ports]
+
+        # Calculate basic risk level
+        if vulns or 23 in ports:
+            risk_level = "HIGH"
+        elif flagged_ports:
+            risk_level = "MEDIUM"
+        else:
+            risk_level = "LOW"
+
+        return {
+            "client_ip": client_ip,
+            "risk_level": risk_level,
+            "indexed": True,
+            "open_ports": ports,
+            "flagged_router_ports": flagged_ports,
+            "vulnerabilities": vulns,
+            "org": data.get("org", "Unknown ISP")
+        }