389 lines
11 KiB
Python
389 lines
11 KiB
Python
import base64
|
|
import binascii
|
|
import os.path
|
|
import re
|
|
import socket
|
|
import struct
|
|
import sys
|
|
import time
|
|
import zlib
|
|
import requests
|
|
from tqdm import tqdm
|
|
from hashlib import sha256
|
|
from tabulate import tabulate
|
|
|
|
import xxtea
|
|
from struct import pack
|
|
from ecdsa import BRAINPOOLP256r1, SigningKey, VerifyingKey
|
|
|
|
|
|
FRAM_MD5 = 'aadc292fe4063a7ac392e3c3dde51e84'
|
|
FRAM_OFFSETS = [3120, 3204, 3206, 3207]
|
|
RESP_PAT = re.compile(rb'admin:\n(.+)\nOK\n', re.MULTILINE | re.DOTALL)
|
|
SERIAL_PAT = re.compile(rb'RIGOL TECHNOLOGIES\$(.+?)\$(.+?)\$(.+?)\n\$(.+?)\$(.+?)\$(.+?)\$(.+?)$')
|
|
OPTIONS = ['BW1T2', 'BW1T3', 'BW1T5', 'BW2T3', 'BW2T5',
|
|
'BW3T5', 'MSO', '2RL', '5RL', 'BND', 'COMP',
|
|
'EMBD', 'AUTO', 'FLEX', 'AUDIO', 'SENSOR',
|
|
'AERO', 'ARINC', 'DG', 'JITTER', 'MASK',
|
|
'PWR', 'DVM', 'CTR', 'EDK', '4CH', 'BW07T1',
|
|
'BW07T2', 'BW07T3', 'BW07T5']
|
|
PRIV_PATH = 'priv.pem'
|
|
KEY1 = b''.join(pack('<I', x) for x in [0x03920001, 0x08410841, 0x18C32104, 0x318639C7])
|
|
KEY2 = b''.join(pack('<I', x) for x in [0x478AA887, 0x99A85895, 0x1770078, 0x87888798])
|
|
|
|
|
|
def decrypt_xxtea1(buf):
|
|
dec = xxtea.decrypt(buf, KEY1, padding=False)
|
|
return dec
|
|
|
|
|
|
def decrypt_xxtea2(buf):
|
|
dec = xxtea.decrypt(buf, KEY2, padding=False)
|
|
return dec
|
|
|
|
|
|
def encrypt_xxtea1(buf):
|
|
delta = len(buf) % 4
|
|
|
|
if delta:
|
|
buf += b'\x00' * (4 - delta)
|
|
|
|
enc = xxtea.encrypt(buf, KEY1, padding=False)
|
|
return enc
|
|
|
|
|
|
def encrypt_xxtea2(buf):
|
|
delta = len(buf) % 4
|
|
|
|
if delta:
|
|
buf += b'\x00' * (4 - delta)
|
|
|
|
enc = xxtea.encrypt(buf, KEY2, padding=False)
|
|
return enc
|
|
|
|
|
|
def sign_option(opt):
|
|
bb = bytearray()
|
|
bb.extend(opt['model'].encode())
|
|
bb.extend(opt['serial'].encode())
|
|
bb.extend(opt['option'].encode())
|
|
bb.extend(opt['version'].encode())
|
|
bb.append(0x00)
|
|
bb.append(0x00)
|
|
buf = bytes(bb)
|
|
dig = sha256(buf).digest()
|
|
|
|
prev_key = True
|
|
if not os.path.exists(PRIV_PATH):
|
|
sk = SigningKey.generate(curve=BRAINPOOLP256r1, hashfunc=sha256)
|
|
with open(PRIV_PATH, 'wb') as w:
|
|
w.write(sk.to_pem())
|
|
|
|
prev_key = False
|
|
else:
|
|
with open(PRIV_PATH) as f:
|
|
sk = SigningKey.from_pem(f.read())
|
|
|
|
sign = sk.sign_digest_deterministic(dig)
|
|
vk = sk.verifying_key
|
|
vkk = b'04' + vk.to_string().hex().upper().encode()
|
|
vk = VerifyingKey.from_string(binascii.unhexlify(vkk), curve=BRAINPOOLP256r1)
|
|
|
|
assert vk.verify_digest(sign, dig)
|
|
return binascii.hexlify(sign).upper(), vkk, prev_key
|
|
|
|
|
|
BLOCK_HDR_FMT = '<IiIiI'
|
|
|
|
|
|
def calc_crc32(buf):
|
|
return zlib.crc32(buf) & 0xFFFFFFFF
|
|
|
|
|
|
def get_dw(buf, off):
|
|
dw = struct.unpack_from('<I', buf, off)[0]
|
|
return dw, off + 4
|
|
|
|
|
|
def get_dws(buf, off):
|
|
dw = struct.unpack_from('<i', buf, off)[0]
|
|
return dw, off + 4
|
|
|
|
|
|
def get_data(buf, off, size):
|
|
block = buf[off:off + size]
|
|
return block, off + size
|
|
|
|
|
|
def read_block(buf, off):
|
|
start = off
|
|
id_, id_neg, data_size, data_size_neg, crc32 = struct.unpack_from(BLOCK_HDR_FMT, buf, off)
|
|
off += struct.calcsize(BLOCK_HDR_FMT)
|
|
|
|
assert ((id_ + id_neg) == 0) and ((data_size + data_size_neg) == 0)
|
|
|
|
block_data, off = get_data(buf, off, data_size)
|
|
crc32_real = calc_crc32(block_data)
|
|
|
|
assert crc32_real == crc32
|
|
|
|
return {
|
|
'offset': start,
|
|
'id': id_,
|
|
'data': block_data,
|
|
'crc32': crc32_real
|
|
}, off
|
|
|
|
|
|
def neg(val):
|
|
return -1 * val
|
|
|
|
|
|
def write_block(buf, off, block, block_data):
|
|
block_len = len(block_data)
|
|
assert len(block['data']) == block_len
|
|
crc32 = calc_crc32(block_data)
|
|
struct.pack_into(BLOCK_HDR_FMT, buf, off + block['offset'], block['id'], neg(block['id']), block_len,
|
|
neg(block_len), crc32)
|
|
block_data_off = off + block['offset'] + struct.calcsize(BLOCK_HDR_FMT)
|
|
buf[block_data_off:block_data_off + block_len] = block_data
|
|
|
|
|
|
def replace_cfram_key(cfram, new_key):
|
|
buf = cfram[0x100:]
|
|
|
|
off = 0
|
|
full_size, off = get_dw(buf, off)
|
|
full_size_neg, off = get_dws(buf, off)
|
|
|
|
assert (full_size + full_size_neg) == 0
|
|
|
|
items = {}
|
|
|
|
while off < full_size:
|
|
block, off = read_block(buf, off)
|
|
# print('%04X: id=%04d, data_sz=%04d, data=%s, crc32=%08X' % (
|
|
# block['offset'], block['id'], len(block['data']), binascii.hexlify(block['data']).decode(), block['crc32']))
|
|
items[block['id']] = block
|
|
|
|
# print('last_off: 0x%04X' % off)
|
|
|
|
data = bytearray(cfram)
|
|
pub_key = items[4512]
|
|
write_block(data, 0x100, pub_key, new_key)
|
|
|
|
return bytes(data)
|
|
|
|
|
|
def exec_rigol_cmd(ip_addr, cmd, need_res=True):
|
|
while True:
|
|
res = requests.post('http://%s/cgi-bin/changepwd.cgi' % ip_addr, data={'pass0': '', 'pass1': '; %s # "' % cmd})
|
|
|
|
if res.status_code == 500:
|
|
continue
|
|
|
|
body = res.content
|
|
m = RESP_PAT.match(body)
|
|
|
|
if m is None and need_res:
|
|
continue
|
|
|
|
if need_res:
|
|
grp = m.group(1)
|
|
return grp.decode()
|
|
else:
|
|
return None
|
|
|
|
|
|
def read_cfram_data(ip_addr):
|
|
print('Reading CFRAM...')
|
|
cfram = bytearray()
|
|
i = 0
|
|
|
|
with tqdm(total=0x2000) as pb:
|
|
while i < 0x2000:
|
|
cmd = '/rigol/tools/fram -r %0x' % i
|
|
res = exec_rigol_cmd(ip_addr, cmd)
|
|
|
|
if res is None:
|
|
continue
|
|
|
|
bb = binascii.unhexlify(res.replace(',', ''))
|
|
cfram.extend(bb)
|
|
i += 0x10
|
|
pb.update(0x10)
|
|
|
|
print('Reading CFRAM done.\n')
|
|
|
|
return bytes(cfram)
|
|
|
|
|
|
def read_rigol_model_serial(ip_addr):
|
|
res = requests.post('http://%s/cgi-bin/welcome.cgi' % ip_addr)
|
|
body = res.content
|
|
m = SERIAL_PAT.match(body)
|
|
|
|
if m is None:
|
|
return None
|
|
|
|
model = m.group(1).decode()
|
|
ser = m.group(2).decode()
|
|
ver = m.group(3).decode()
|
|
mac = m.group(4).decode()
|
|
|
|
print('Model: %s\nSerial: %s\nVersion: %s\nMAC: %s\n' % (model, ser, ver, mac))
|
|
|
|
return model, ser
|
|
|
|
|
|
def apply_new_key(ip_addr, new_cfram, new_key):
|
|
print('Applying new CFRAM...')
|
|
exec_rigol_cmd(ip_addr, 'echo -n -e \'\\x03\' > /tmp/byte1', need_res=False)
|
|
exec_rigol_cmd(ip_addr, 'echo -n -e \'\\x3d\' > /tmp/byte2', need_res=False)
|
|
exec_rigol_cmd(ip_addr, 'echo -n -e \'\\x5b\' > /tmp/byte3', need_res=False)
|
|
exec_rigol_cmd(ip_addr, 'echo -n -e \'\\xe5\' > /tmp/byte4', need_res=False)
|
|
|
|
exec_rigol_cmd(ip_addr, 'cp /rigol/tools/fram /rigol/tools/fram01', need_res=False)
|
|
exec_rigol_cmd(ip_addr, 'chmod +x /rigol/tools/fram01', need_res=False)
|
|
exec_rigol_cmd(ip_addr, 'dd if=/tmp/byte1 of=/rigol/tools/fram01 obs=1 seek=%d conv=notrunc' % FRAM_OFFSETS[0], need_res=False)
|
|
exec_rigol_cmd(ip_addr, 'dd if=/tmp/byte2 of=/rigol/tools/fram01 obs=1 seek=%d conv=notrunc' % FRAM_OFFSETS[1], need_res=False)
|
|
exec_rigol_cmd(ip_addr, 'dd if=/tmp/byte3 of=/rigol/tools/fram01 obs=1 seek=%d conv=notrunc' % FRAM_OFFSETS[2], need_res=False)
|
|
exec_rigol_cmd(ip_addr, 'dd if=/tmp/byte4 of=/rigol/tools/fram01 obs=1 seek=%d conv=notrunc' % FRAM_OFFSETS[3], need_res=False)
|
|
|
|
with tqdm(total=len(new_cfram)) as pb:
|
|
for i, b in enumerate(new_cfram):
|
|
exec_rigol_cmd(ip_addr, '/rigol/tools/fram01 -w %x %02x' % (i, b), need_res=False)
|
|
pb.update(1)
|
|
|
|
print('New CFRAM applied.\n')
|
|
|
|
exec_rigol_cmd(ip_addr, 'cp /rigol/data/Key.data /rigol/data/Key.data.bak', need_res=False)
|
|
print('Key.data backup created.')
|
|
exec_rigol_cmd(ip_addr, 'echo -n %s | base64 -d > /rigol/data/Key.data' % base64.b64encode(new_key).decode(), need_res=False)
|
|
print('New Key.data applied.')
|
|
|
|
|
|
def check_fram_tool(ip_addr):
|
|
res = exec_rigol_cmd(ip_addr, 'md5sum /rigol/tools/fram')
|
|
res = res.split(' ')[0]
|
|
|
|
if res != FRAM_MD5:
|
|
print('Different /rigol/tools/fram hash. You have to recalc FRAM_OFFSETS!')
|
|
exit(-1)
|
|
|
|
print('/rigol/tools/fram is OK!\n')
|
|
|
|
|
|
def deactivate_option(ip_addr, code, line):
|
|
print('Deactivating: %s...' % code, end=' ')
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
s.connect((ip_addr, 5555))
|
|
s.sendall(b':SYST:OPT:UNIN %s\n' % line)
|
|
print('deactivated.')
|
|
s.close()
|
|
|
|
|
|
def activate_option(ip_addr, code, line):
|
|
print('Activating: %s...' % code, end=' ')
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
s.settimeout(1)
|
|
s.connect((ip_addr, 5555))
|
|
s.sendall(b':SYST:OPT:INST %s\n' % line)
|
|
s.sendall(b':SYST:OPT:STAT? %s\n' % code.encode())
|
|
|
|
try:
|
|
res = s.recv(2)
|
|
|
|
res = res.rstrip(b'\n')
|
|
|
|
if res == b'0':
|
|
print('not', end=' ')
|
|
|
|
print('activated.')
|
|
except TimeoutError:
|
|
print('unavailable.')
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def get_unavail_options(ip_addr):
|
|
items = []
|
|
while True:
|
|
res = requests.post('http://%s/cgi-bin/options.cgi' % ip_addr)
|
|
|
|
if res.status_code == 500:
|
|
continue
|
|
|
|
body = res.content.decode()
|
|
items = body.split('#')
|
|
break
|
|
|
|
table = [['Code', 'Status', 'Description']]
|
|
actives = []
|
|
|
|
for item in items:
|
|
row = item.split('$')
|
|
table.append(row)
|
|
|
|
if row[1] == 'Forever':
|
|
actives.append(row[0])
|
|
|
|
print(tabulate(table, headers='firstrow', tablefmt='fancy_grid'))
|
|
|
|
return list(set(OPTIONS) - set(actives))
|
|
|
|
|
|
def main(ip_addr):
|
|
unavails = get_unavail_options(ip_addr)
|
|
|
|
check_fram_tool(ip_addr)
|
|
|
|
model, ser = read_rigol_model_serial(ip_addr)
|
|
|
|
opts = []
|
|
key_hex = None
|
|
prev_key = True
|
|
|
|
for option in unavails:
|
|
opt_sign, key_hex, prev_key = sign_option({
|
|
'model': model,
|
|
'serial': ser,
|
|
'option': option,
|
|
'version': '1.0'
|
|
})
|
|
opts.append((option, opt_sign))
|
|
|
|
# Set this to False to force overwrite the key
|
|
# prev_key = False
|
|
cfram = None
|
|
if not prev_key:
|
|
cfram = read_cfram_data(ip_addr)
|
|
|
|
|
|
new_key = b'brainpoolP256r1;%s' % key_hex
|
|
# ELHER necessary due to different key on my scope.
|
|
# if len(new_key) < 388:
|
|
# new_key += (388 - len(new_key)) * b"\xaa"
|
|
new_key = encrypt_xxtea1(new_key)
|
|
|
|
# new_key =
|
|
# new_key = encrypt_xxtea2(b'brainpoolP256r1;%s' % key_hex)
|
|
|
|
if not prev_key:
|
|
new_cfram = replace_cfram_key(cfram, new_key)
|
|
apply_new_key(ip_addr, new_cfram, new_key)
|
|
|
|
model = model[:-2] + '00'
|
|
|
|
for opt in opts:
|
|
code = opt[0].encode()
|
|
activate_option(ip_addr, code.decode(), b'%s-%s@%s' % (model.encode(), code, opt[1]))
|
|
|
|
get_unavail_options(ip_addr)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
if len(sys.argv) < 2:
|
|
print('Usage: python rigol_kg.py 192.168.1.1')
|
|
main(sys.argv[1])
|