Improve tool/

This commit is contained in:
NIIBE Yutaka
2015-06-03 16:27:15 +09:00
parent bd02cbfdb0
commit ce070f85ff
3 changed files with 55 additions and 399 deletions

View File

@@ -123,15 +123,16 @@ class gnuk_token(object):
end = ((mem[7]*256 + mem[6])*256 + mem[5])*256 + mem[4]
return (start, end)
def download(self, start, data):
def download(self, start, data, verbose=False):
addr = start
addr_end = (start + len(data)) & 0xffffff00
i = (addr - 0x20000000) / 0x100
j = 0
print "start %08x" % addr
print "end %08x" % addr_end
print("start %08x" % addr)
print("end %08x" % addr_end)
while addr < addr_end:
print "# %08x: %d : %d" % (addr, i, 256)
if verbose:
print("# %08x: %d : %d" % (addr, i, 256))
self.__devhandle.controlMsg(requestType = 0x40, request = 1,
value = i, index = 0,
buffer = data[j*256:j*256+256],
@@ -141,7 +142,8 @@ class gnuk_token(object):
addr = addr + 256
residue = len(data) % 256
if residue != 0:
print "# %08x: %d : %d" % (addr, i, residue)
if verbose:
print("# %08x: %d : %d" % (addr, i, residue))
self.__devhandle.controlMsg(requestType = 0x40, request = 1,
value = i, index = 0,
buffer = data[j*256:],
@@ -157,7 +159,7 @@ class gnuk_token(object):
def icc_get_result(self):
msg = self.__devhandle.bulkRead(self.__bulkin, 1024, self.__timeout)
if len(msg) < 10:
print msg
print(msg)
raise ValueError("icc_get_result")
msg_type = msg[0]
data_len = msg[1] + (msg[2]<<8) + (msg[3]<<16) + (msg[4]<<24)
@@ -491,15 +493,16 @@ class regnual(object):
end = ((mem[7]*256 + mem[6])*256 + mem[5])*256 + mem[4]
return (start, end)
def download(self, start, data):
def download(self, start, data, verbose=False):
addr = start
addr_end = (start + len(data)) & 0xffffff00
i = (addr - 0x08000000) / 0x100
j = 0
print "start %08x" % addr
print "end %08x" % addr_end
print("start %08x" % addr)
print("end %08x" % addr_end)
while addr < addr_end:
print "# %08x: %d: %d : %d" % (addr, i, j, 256)
if verbose:
print("# %08x: %d: %d : %d" % (addr, i, j, 256))
self.__devhandle.controlMsg(requestType = 0x40, request = 1,
value = 0, index = 0,
buffer = data[j*256:j*256+256],
@@ -510,7 +513,7 @@ class regnual(object):
timeout = 10000)
r_value = ((res[3]*256 + res[2])*256 + res[1])*256 + res[0]
if (crc32code ^ r_value) != 0xffffffff:
print "failure"
print("failure")
self.__devhandle.controlMsg(requestType = 0x40, request = 3,
value = i, index = 0,
buffer = None,
@@ -521,13 +524,14 @@ class regnual(object):
timeout = 10000)
r_value = ((res[3]*256 + res[2])*256 + res[1])*256 + res[0]
if r_value == 0:
print "failure"
print("failure")
i = i+1
j = j+1
addr = addr + 256
residue = len(data) % 256
if residue != 0:
print "# %08x: %d : %d" % (addr, i, residue)
if verbose:
print("# %08x: %d : %d" % (addr, i, residue))
self.__devhandle.controlMsg(requestType = 0x40, request = 1,
value = 0, index = 0,
buffer = data[j*256:],
@@ -538,7 +542,7 @@ class regnual(object):
timeout = 10000)
r_value = ((res[3]*256 + res[2])*256 + res[1])*256 + res[0]
if (crc32code ^ r_value) != 0xffffffff:
print "failure"
print("failure")
self.__devhandle.controlMsg(requestType = 0x40, request = 3,
value = i, index = 0,
buffer = None,
@@ -549,7 +553,7 @@ class regnual(object):
timeout = 10000)
r_value = ((res[3]*256 + res[2])*256 + res[1])*256 + res[0]
if r_value == 0:
print "failure"
print("failure")
def protect(self):
self.__devhandle.controlMsg(requestType = 0x40, request = 4,
@@ -561,7 +565,7 @@ class regnual(object):
timeout = 10000)
r_value = ((res[3]*256 + res[2])*256 + res[1])*256 + res[0]
if r_value == 0:
print "protection failure"
print("protection failure")
def finish(self):
self.__devhandle.controlMsg(requestType = 0x40, request = 5,
@@ -611,9 +615,9 @@ def get_gnuk_device():
for (dev, config, intf) in gnuk_devices():
try:
icc = gnuk_token(dev, config, intf)
print "Device: ", dev.filename
print "Configuration: ", config.value
print "Interface: ", intf.interfaceNumber
print("Device: ", dev.filename)
print("Configuration: ", config.value)
print("Interface: ", intf.interfaceNumber)
break
except:
pass

View File

@@ -3,7 +3,7 @@
"""
gnuk_upgrade.py - a tool to upgrade firmware of Gnuk Token
Copyright (C) 2012 Free Software Initiative of Japan
Copyright (C) 2012, 2015 Free Software Initiative of Japan
Author: NIIBE Yutaka <gniibe@fsij.org>
This file is a part of Gnuk, a GnuPG USB Token implementation.
@@ -31,354 +31,7 @@ import sys, time, os, binascii, string
import usb
# USB class, subclass, protocol
CCID_CLASS = 0x0B
CCID_SUBCLASS = 0x00
CCID_PROTOCOL_0 = 0x00
def icc_compose(msg_type, data_len, slot, seq, param, data):
return pack('<BiBBBH', msg_type, data_len, slot, seq, 0, param) + data
def iso7816_compose(ins, p1, p2, data, cls=0x00):
data_len = len(data)
if data_len == 0:
return pack('>BBBB', cls, ins, p1, p2)
else:
return pack('>BBBBB', cls, ins, p1, p2, data_len) + data
class regnual(object):
def __init__(self, dev):
conf = dev.configurations[0]
intf_alt = conf.interfaces[0]
intf = intf_alt[0]
if intf.interfaceClass != 0xff:
raise ValueError, "Wrong interface class"
self.__devhandle = dev.open()
try:
self.__devhandle.setConfiguration(conf)
except:
pass
self.__devhandle.claimInterface(intf)
self.__devhandle.setAltInterface(0)
def mem_info(self):
mem = self.__devhandle.controlMsg(requestType = 0xc0, request = 0,
value = 0, index = 0, buffer = 8,
timeout = 10000)
start = ((mem[3]*256 + mem[2])*256 + mem[1])*256 + mem[0]
end = ((mem[7]*256 + mem[6])*256 + mem[5])*256 + mem[4]
return (start, end)
def download(self, start, data):
addr = start
addr_end = (start + len(data)) & 0xffffff00
i = (addr - 0x08000000) / 0x100
j = 0
print "start %08x" % addr
print "end %08x" % addr_end
while addr < addr_end:
print "# %08x: %d: %d : %d" % (addr, i, j, 256)
self.__devhandle.controlMsg(requestType = 0x40, request = 1,
value = 0, index = 0,
buffer = data[j*256:j*256+256],
timeout = 10000)
crc32code = crc32(data[j*256:j*256+256])
res = self.__devhandle.controlMsg(requestType = 0xc0, request = 2,
value = 0, index = 0, buffer = 4,
timeout = 10000)
r_value = ((res[3]*256 + res[2])*256 + res[1])*256 + res[0]
if (crc32code ^ r_value) != 0xffffffff:
print "failure"
self.__devhandle.controlMsg(requestType = 0x40, request = 3,
value = i, index = 0,
buffer = None,
timeout = 10000)
time.sleep(0.010)
res = self.__devhandle.controlMsg(requestType = 0xc0, request = 2,
value = 0, index = 0, buffer = 4,
timeout = 10000)
r_value = ((res[3]*256 + res[2])*256 + res[1])*256 + res[0]
if r_value == 0:
print "failure"
i = i+1
j = j+1
addr = addr + 256
residue = len(data) % 256
if residue != 0:
print "# %08x: %d : %d" % (addr, i, residue)
self.__devhandle.controlMsg(requestType = 0x40, request = 1,
value = 0, index = 0,
buffer = data[j*256:],
timeout = 10000)
crc32code = crc32(data[j*256:].ljust(256,chr(255)))
res = self.__devhandle.controlMsg(requestType = 0xc0, request = 2,
value = 0, index = 0, buffer = 4,
timeout = 10000)
r_value = ((res[3]*256 + res[2])*256 + res[1])*256 + res[0]
if (crc32code ^ r_value) != 0xffffffff:
print "failure"
self.__devhandle.controlMsg(requestType = 0x40, request = 3,
value = i, index = 0,
buffer = None,
timeout = 10000)
time.sleep(0.010)
res = self.__devhandle.controlMsg(requestType = 0xc0, request = 2,
value = 0, index = 0, buffer = 4,
timeout = 10000)
r_value = ((res[3]*256 + res[2])*256 + res[1])*256 + res[0]
if r_value == 0:
print "failure"
def protect(self):
self.__devhandle.controlMsg(requestType = 0x40, request = 4,
value = 0, index = 0, buffer = None,
timeout = 10000)
time.sleep(0.100)
res = self.__devhandle.controlMsg(requestType = 0xc0, request = 2,
value = 0, index = 0, buffer = 4,
timeout = 10000)
r_value = ((res[3]*256 + res[2])*256 + res[1])*256 + res[0]
if r_value == 0:
print "protection failure"
def finish(self):
self.__devhandle.controlMsg(requestType = 0x40, request = 5,
value = 0, index = 0, buffer = None,
timeout = 10000)
def reset_device(self):
try:
self.__devhandle.reset()
except:
pass
# This class only supports Gnuk (for now)
class gnuk_token(object):
def __init__(self, device, configuration, interface):
"""
__init__(device, configuration, interface) -> None
Initialize the device.
device: usb.Device object.
configuration: configuration number.
interface: usb.Interface object representing the interface and altenate setting.
"""
if interface.interfaceClass != CCID_CLASS:
raise ValueError, "Wrong interface class"
if interface.interfaceSubClass != CCID_SUBCLASS:
raise ValueError, "Wrong interface sub class"
self.__devhandle = device.open()
try:
self.__devhandle.setConfiguration(configuration)
except:
pass
self.__devhandle.claimInterface(interface)
self.__devhandle.setAltInterface(0)
self.__intf = interface.interfaceNumber
self.__alt = interface.alternateSetting
self.__conf = configuration
self.__bulkout = 1
self.__bulkin = 0x81
self.__timeout = 10000
self.__seq = 0
def reset_device(self):
try:
self.__devhandle.reset()
except:
pass
def stop_gnuk(self):
self.__devhandle.releaseInterface()
self.__devhandle.setConfiguration(0)
return
def mem_info(self):
mem = self.__devhandle.controlMsg(requestType = 0xc0, request = 0,
value = 0, index = 0, buffer = 8,
timeout = 10)
start = ((mem[3]*256 + mem[2])*256 + mem[1])*256 + mem[0]
end = ((mem[7]*256 + mem[6])*256 + mem[5])*256 + mem[4]
return (start, end)
def download(self, start, data):
addr = start
addr_end = (start + len(data)) & 0xffffff00
i = (addr - 0x20000000) / 0x100
j = 0
print "start %08x" % addr
print "end %08x" % addr_end
while addr < addr_end:
print "# %08x: %d : %d" % (addr, i, 256)
self.__devhandle.controlMsg(requestType = 0x40, request = 1,
value = i, index = 0,
buffer = data[j*256:j*256+256],
timeout = 10)
i = i+1
j = j+1
addr = addr + 256
residue = len(data) % 256
if residue != 0:
print "# %08x: %d : %d" % (addr, i, residue)
self.__devhandle.controlMsg(requestType = 0x40, request = 1,
value = i, index = 0,
buffer = data[j*256:],
timeout = 10)
def execute(self, last_addr):
i = (last_addr - 0x20000000) / 0x100
o = (last_addr - 0x20000000) % 0x100
self.__devhandle.controlMsg(requestType = 0x40, request = 2,
value = i, index = o, buffer = None,
timeout = 10)
def icc_get_result(self):
msg = self.__devhandle.bulkRead(self.__bulkin, 1024, self.__timeout)
if len(msg) < 10:
raise ValueError, "icc_get_result"
msg_type = msg[0]
data_len = msg[1] + (msg[2]<<8) + (msg[3]<<16) + (msg[4]<<24)
slot = msg[5]
seq = msg[6]
status = msg[7]
error = msg[8]
chain = msg[9]
data = msg[10:]
# XXX: check msg_type, data_len, slot, seq, error
return (status, chain, data)
def icc_get_status(self):
msg = icc_compose(0x65, 0, 0, self.__seq, 0, "")
self.__devhandle.bulkWrite(self.__bulkout, msg, self.__timeout)
self.__seq += 1
status, chain, data = self.icc_get_result()
# XXX: check chain, data
return status
def icc_power_on(self):
msg = icc_compose(0x62, 0, 0, self.__seq, 0, "")
self.__devhandle.bulkWrite(self.__bulkout, msg, self.__timeout)
self.__seq += 1
status, chain, data = self.icc_get_result()
# XXX: check status, chain
return data # ATR
def icc_power_off(self):
msg = icc_compose(0x63, 0, 0, self.__seq, 0, "")
self.__devhandle.bulkWrite(self.__bulkout, msg, self.__timeout)
self.__seq += 1
status, chain, data = self.icc_get_result()
# XXX: check chain, data
return status
def icc_send_data_block(self, data):
msg = icc_compose(0x6f, len(data), 0, self.__seq, 0, data)
self.__devhandle.bulkWrite(self.__bulkout, msg, self.__timeout)
self.__seq += 1
return self.icc_get_result()
def icc_send_cmd(self, data):
status, chain, data_rcv = self.icc_send_data_block(data)
if chain == 0:
return data_rcv
elif chain == 1:
d = data_rcv
while True:
msg = icc_compose(0x6f, 0, 0, self.__seq, 0x10, "")
self.__devhandle.bulkWrite(self.__bulkout, msg, self.__timeout)
self.__seq += 1
status, chain, data_rcv = self.icc_get_result()
# XXX: check status
d += data_rcv
if chain == 2:
break
elif chain == 3:
continue
else:
raise ValueError, "icc_send_cmd chain"
return d
else:
raise ValueError, "icc_send_cmd"
def cmd_get_response(self, expected_len):
cmd_data = iso7816_compose(0xc0, 0x00, 0x00, '') + pack('>B', expected_len)
response = self.icc_send_cmd(cmd_data)
return response[:-2]
def cmd_verify(self, who, passwd):
cmd_data = iso7816_compose(0x20, 0x00, 0x80+who, passwd)
sw = self.icc_send_cmd(cmd_data)
if len(sw) != 2:
raise ValueError, sw
if not (sw[0] == 0x90 and sw[1] == 0x00):
raise ValueError, sw
def cmd_select_openpgp(self):
cmd_data = iso7816_compose(0xa4, 0x04, 0x0c, "\xD2\x76\x00\x01\x24\x01")
sw = self.icc_send_cmd(cmd_data)
if len(sw) != 2:
raise ValueError, sw
if not (sw[0] == 0x90 and sw[1] == 0x00):
raise ValueError, ("%02x%02x" % (sw[0], sw[1]))
def cmd_external_authenticate(self, signed):
cmd_data = iso7816_compose(0x82, 0x00, 0x00, signed[0:128], cls=0x10)
sw = self.icc_send_cmd(cmd_data)
if len(sw) != 2:
raise ValueError, sw
if not (sw[0] == 0x90 and sw[1] == 0x00):
raise ValueError, ("%02x%02x" % (sw[0], sw[1]))
cmd_data = iso7816_compose(0x82, 0x00, 0x00, signed[128:])
sw = self.icc_send_cmd(cmd_data)
if len(sw) != 2:
raise ValueError, sw
if not (sw[0] == 0x90 and sw[1] == 0x00):
raise ValueError, ("%02x%02x" % (sw[0], sw[1]))
def cmd_get_challenge(self):
cmd_data = iso7816_compose(0x84, 0x00, 0x00, '')
sw = self.icc_send_cmd(cmd_data)
if len(sw) != 2:
raise ValueError, sw
if sw[0] != 0x61:
raise ValueError, ("%02x%02x" % (sw[0], sw[1]))
return self.cmd_get_response(sw[1])
def compare(data_original, data_in_device):
i = 0
for d in data_original:
if ord(d) != data_in_device[i]:
raise ValueError, "verify failed at %08x" % i
i += 1
def ccid_devices():
busses = usb.busses()
for bus in busses:
devices = bus.devices
for dev in devices:
for config in dev.configurations:
for intf in config.interfaces:
for alt in intf:
if alt.interfaceClass == CCID_CLASS and \
alt.interfaceSubClass == CCID_SUBCLASS and \
alt.interfaceProtocol == CCID_PROTOCOL_0:
yield dev, config, alt
USB_VENDOR_FSIJ=0x234b
USB_PRODUCT_GNUK=0x0000
def gnuk_devices():
busses = usb.busses()
for bus in busses:
devices = bus.devices
for dev in devices:
if dev.idVendor != USB_VENDOR_FSIJ:
continue
if dev.idProduct != USB_PRODUCT_GNUK:
continue
yield dev
from gnuk_token import *
def to_string(t):
result = ""
@@ -420,24 +73,17 @@ def gpg_sign(keygrip, hash):
pos = signed.index("D ") + 2
signed = signed[pos:-4] # \nOK\n
if len(signed) != 256:
raise ValueError, binascii.hexlify(signed)
raise ValueError(binascii.hexlify(signed))
return signed
def UNSIGNED(n):
return n & 0xffffffff
def crc32(bytestr):
crc = binascii.crc32(bytestr)
return UNSIGNED(crc)
def main(keygrip, data_regnual, data_upgrade):
def main(keyno,keygrip, data_regnual, data_upgrade):
l = len(data_regnual)
if (l & 0x03) != 0:
data_regnual = data_regnual.ljust(l + 4 - (l & 0x03), chr(0))
crc32code = crc32(data_regnual)
print "CRC32: %04x\n" % crc32code
data_regnual += pack('<I', crc32code)
for (dev, config, intf) in ccid_devices():
for (dev, config, intf) in gnuk_devices():
try:
icc = gnuk_token(dev, config, intf)
print "Device: ", dev.filename
@@ -447,13 +93,13 @@ def main(keygrip, data_regnual, data_upgrade):
except:
icc = None
if icc.icc_get_status() == 2:
raise ValueError, "No ICC present"
raise ValueError("No ICC present")
elif icc.icc_get_status() == 1:
icc.icc_power_on()
icc.cmd_select_openpgp()
challenge = icc.cmd_get_challenge()
signed = gpg_sign(keygrip, binascii.hexlify(to_string(challenge)))
icc.cmd_external_authenticate(signed)
icc.cmd_external_authenticate(keyno, signed)
icc.stop_gnuk()
mem_info = icc.mem_info()
print "%08x:%08x" % mem_info
@@ -471,7 +117,7 @@ def main(keygrip, data_regnual, data_upgrade):
time.sleep(3)
# Then, send upgrade program...
reg = None
for dev in gnuk_devices():
for dev in gnuk_devices_by_vidpid():
try:
reg = regnual(dev)
print "Device: ", dev.filename
@@ -489,6 +135,7 @@ def main(keygrip, data_regnual, data_upgrade):
if __name__ == '__main__':
keyno = 0
keygrip = None
if sys.argv[1] == '-k':
sys.argv.pop(1)
@@ -504,4 +151,4 @@ if __name__ == '__main__':
data_upgrade = f.read()
f.close()
print "%s: %d" % (filename_upgrade, len(data_upgrade))
main(keygrip, data_regnual, data_upgrade[4096:])
main(keyno, keygrip, data_regnual, data_upgrade[4096:])

View File

@@ -4,7 +4,7 @@
upgrade_by_passwd.py - a tool to install another firmware for Gnuk Token
which is just shipped from factory
Copyright (C) 2012, 2013 Free Software Initiative of Japan
Copyright (C) 2012, 2013, 2015 Free Software Initiative of Japan
Author: NIIBE Yutaka <gniibe@fsij.org>
This file is a part of Gnuk, a GnuPG USB Token implementation.
@@ -37,7 +37,7 @@ def main(keyno, passwd, data_regnual, data_upgrade):
if (l & 0x03) != 0:
data_regnual = data_regnual.ljust(l + 4 - (l & 0x03), chr(0))
crc32code = crc32(data_regnual)
print "CRC32: %04x\n" % crc32code
print("CRC32: %04x\n" % crc32code)
data_regnual += pack('<I', crc32code)
rsa_key = rsa.read_key_from_file('rsa_example.key')
@@ -55,11 +55,11 @@ def main(keyno, passwd, data_regnual, data_upgrade):
gnuk.cmd_external_authenticate(keyno, signed_bytes)
gnuk.stop_gnuk()
mem_info = gnuk.mem_info()
print "%08x:%08x" % mem_info
print("%08x:%08x" % mem_info)
print "Downloading flash upgrade program..."
print("Downloading flash upgrade program...")
gnuk.download(mem_info[0], data_regnual)
print "Run flash upgrade program..."
print("Run flash upgrade program...")
gnuk.execute(mem_info[0] + len(data_regnual) - 4)
#
time.sleep(3)
@@ -67,51 +67,56 @@ def main(keyno, passwd, data_regnual, data_upgrade):
del gnuk
gnuk = None
#
print "Wait 3 seconds..."
print("Wait 3 seconds...")
time.sleep(3)
# Then, send upgrade program...
reg = None
for dev in gnuk_devices_by_vidpid():
try:
reg = regnual(dev)
print "Device: ", dev.filename
print("Device: ", dev.filename)
break
except:
pass
mem_info = reg.mem_info()
print "%08x:%08x" % mem_info
print "Downloading the program"
print("%08x:%08x" % mem_info)
print("Downloading the program")
reg.download(mem_info[0], data_upgrade)
reg.protect()
reg.finish()
reg.reset_device()
return 0
from getpass import getpass
if __name__ == '__main__':
if os.getcwd() != os.path.dirname(os.path.abspath(__file__)):
print "Please change working directory to: %s" % os.path.dirname(os.path.abspath(__file__))
print("Please change working directory to: %s" % os.path.dirname(os.path.abspath(__file__)))
exit(1)
passwd = DEFAULT_PW3
keyno = 0
passwd = None
while len(sys.argv) > 3:
option = sys.argv[1]
sys.argv.pop(1)
if option == '-p':
from getpass import getpass
passwd = getpass("Admin password: ")
elif option == '-k':
if option == '-f': # F for Factory setting
passwd = DEFAULT_PW3
elif option == '-k': # K for Key number
keyno = int(sys.argv[1])
sys.argv.pop(1)
else:
raise ValueError("unknown option", option)
if not passwd:
passwd = getpass("Admin password: ")
filename_regnual = sys.argv[1]
filename_upgrade = sys.argv[2]
f = open(filename_regnual)
data_regnual = f.read()
f.close()
print "%s: %d" % (filename_regnual, len(data_regnual))
print("%s: %d" % (filename_regnual, len(data_regnual)))
f = open(filename_upgrade)
data_upgrade = f.read()
f.close()
print "%s: %d" % (filename_upgrade, len(data_upgrade))
print("%s: %d" % (filename_upgrade, len(data_upgrade)))
# First 4096-byte in data_upgrade is SYS, so, skip it.
main(keyno, passwd, data_regnual, data_upgrade[4096:])