Port gdb.py and hexprog.py to Python 3.
Unfortunately, the serial data are not 7-bit clean (see write_mem()). So getpacket() and putpacket() use bytes objects rather than strings.
This commit is contained in:
Mark Rages
Piotr Esden-Tempski
parent
fcb2a609fc
commit
f43101fd9f
227
scripts/gdb.py
227
scripts/gdb.py
|
|
@ -24,18 +24,12 @@ import struct
|
|||
import time
|
||||
|
||||
def hexify(s):
|
||||
"""Convert a binary string into hex representation"""
|
||||
ret = ''
|
||||
for c in s:
|
||||
ret += "%02X" % ord(c)
|
||||
return ret
|
||||
"""Convert a bytes object into hex bytes representation"""
|
||||
return s.hex().encode()
|
||||
|
||||
def unhexify(s):
|
||||
"""Convert a hex string into binary representation"""
|
||||
ret = ''
|
||||
for i in range(0, len(s), 2):
|
||||
ret += chr(int(s[i:i+2], 16))
|
||||
return ret
|
||||
"""Convert a hex-encoded bytes into bytes object"""
|
||||
return bytes.fromhex(s.decode())
|
||||
|
||||
class FakeSocket:
|
||||
"""Emulate socket functions send and recv on a file object"""
|
||||
|
|
@ -55,145 +49,183 @@ class Target:
|
|||
else:
|
||||
self.sock = FakeSocket(sock)
|
||||
|
||||
self.PacketSize=0x100 # default
|
||||
|
||||
def getpacket(self):
|
||||
"""Return the first correctly received packet from GDB target"""
|
||||
while True:
|
||||
while self.sock.recv(1) != '$': pass
|
||||
while self.sock.recv(1) != b'$':
|
||||
pass
|
||||
|
||||
csum = 0
|
||||
packet = ''
|
||||
packet = [] # list-of-small-int
|
||||
|
||||
while True:
|
||||
c = self.sock.recv(1)
|
||||
if c == '#': break
|
||||
c, = self.sock.recv(1)
|
||||
if c == ord('#'):
|
||||
break
|
||||
|
||||
if c == '$':
|
||||
packet = ''
|
||||
if c == ord('$'):
|
||||
packet = []
|
||||
csum = 0
|
||||
continue
|
||||
|
||||
if c == '}':
|
||||
c = self.sock.recv(1)
|
||||
csum += ord(c) + ord('}')
|
||||
packet += chr(ord(c) ^ 0x20)
|
||||
if c == ord('}'):
|
||||
c, = self.sock.recv(1)
|
||||
csum += c + ord('}')
|
||||
packet.append(c ^ 0x20)
|
||||
continue
|
||||
|
||||
packet += c
|
||||
csum += ord(c)
|
||||
packet.append(c)
|
||||
csum += c
|
||||
|
||||
if (csum & 0xFF) == int(self.sock.recv(2),16): break
|
||||
if (csum & 0xFF) == int(self.sock.recv(2),16):
|
||||
break
|
||||
|
||||
self.sock.send('-')
|
||||
|
||||
self.sock.send('+')
|
||||
return packet
|
||||
self.sock.send(b'-')
|
||||
|
||||
self.sock.send(b'+')
|
||||
return bytes(packet)
|
||||
|
||||
def putpacket(self, packet):
|
||||
"""Send packet to GDB target and wait for acknowledge"""
|
||||
"""Send packet to GDB target and wait for acknowledge
|
||||
packet is bytes or string"""
|
||||
|
||||
if type(packet) == str:
|
||||
packet = packet.encode()
|
||||
|
||||
while True:
|
||||
self.sock.send('$')
|
||||
csum = 0
|
||||
out = []
|
||||
|
||||
for c in packet:
|
||||
if (c == '$') or (c == '#') or (c == '}'):
|
||||
self.sock.send('}')
|
||||
self.sock.send(chr(ord(c) ^ 0x20))
|
||||
csum += (ord(c) ^ 0x20) + ord('}')
|
||||
if (c in b'$#}'):
|
||||
out.append(ord('}'))
|
||||
out.append(c ^ 0x20)
|
||||
else:
|
||||
self.sock.send(c)
|
||||
csum += ord(c)
|
||||
self.sock.send('#')
|
||||
self.sock.send("%02X" % (csum & 0xFF))
|
||||
if self.sock.recv(1) == '+': break
|
||||
out.append(c)
|
||||
|
||||
csum = sum(out)
|
||||
outb = b'$'+bytes(out)+b'#%02X' % (csum & 0xff)
|
||||
|
||||
self.sock.send(outb)
|
||||
if self.sock.recv(1) == b'+':
|
||||
break
|
||||
|
||||
def monitor(self, cmd):
|
||||
"""Send gdb "monitor" command to target"""
|
||||
if type(cmd) == str:
|
||||
cmd = cmd.encode()
|
||||
|
||||
ret = []
|
||||
self.putpacket("qRcmd," + hexify(cmd))
|
||||
self.putpacket(b"qRcmd," + hexify(cmd))
|
||||
|
||||
while True:
|
||||
s = self.getpacket()
|
||||
if s == '': return None
|
||||
if s == 'OK': return ret
|
||||
if s.startswith('O'): ret.append(unhexify(s[1:]))
|
||||
|
||||
if s == b'':
|
||||
return None
|
||||
if s == b'OK':
|
||||
return ret
|
||||
if s.startswith(b'O'):
|
||||
ret.append(unhexify(s[1:]))
|
||||
else:
|
||||
raise Exception('Invalid GDB stub response')
|
||||
raise Exception('Invalid GDB stub response %r'%s.decode())
|
||||
|
||||
def attach(self, pid):
|
||||
"""Attach to target process (gdb "attach" command)"""
|
||||
self.putpacket("vAttach;%08X" % pid)
|
||||
self.putpacket(b"vAttach;%08X" % pid)
|
||||
reply = self.getpacket()
|
||||
if (len(reply) == 0) or (reply[0] == 'E'):
|
||||
if (reply == b'') or (reply[:1] == b'E'):
|
||||
raise Exception('Failed to attach to remote pid %d' % pid)
|
||||
|
||||
def detach(self):
|
||||
"""Detach from target process (gdb "detach" command)"""
|
||||
self.putpacket("D")
|
||||
if self.getpacket() != 'OK':
|
||||
self.putpacket(b"D")
|
||||
if self.getpacket() != b'OK':
|
||||
raise Exception("Failed to detach from remote process")
|
||||
|
||||
def reset(self):
|
||||
"""Reset the target system"""
|
||||
self.putpacket("r")
|
||||
self.putpacket(b"r")
|
||||
|
||||
def read_mem(self, addr, length):
|
||||
"""Read length bytes from target at address addr"""
|
||||
self.putpacket("m%08X,%08X" % (addr, length))
|
||||
reply = self.getpacket()
|
||||
if (len(reply) == 0) or (reply[0] == 'E'):
|
||||
raise Exception('Error reading memory at 0x%08X' % addr)
|
||||
try:
|
||||
data = unhexify(reply)
|
||||
except Excpetion:
|
||||
raise Exception('Invalid response to memory read packet: %r' % reply)
|
||||
return data
|
||||
ret = b''
|
||||
while length:
|
||||
# print "Read"
|
||||
packlen = min(length,self.PacketSize//2)
|
||||
self.putpacket(b"m%08X,%08X" % (addr, packlen))
|
||||
reply = self.getpacket()
|
||||
if (reply == b'') or (reply[:1] == b'E'):
|
||||
raise Exception('Error reading memory at 0x%08X' % addr)
|
||||
try:
|
||||
data = unhexify(reply)
|
||||
except Exception:
|
||||
raise Exception('Invalid response to memory read packet: %r' % reply)
|
||||
ret += data
|
||||
length -= packlen
|
||||
addr += packlen
|
||||
|
||||
return ret
|
||||
|
||||
def write_mem(self, addr, data):
|
||||
"""Write data to target at address addr"""
|
||||
self.putpacket("X%08X,%08X:%s" % (addr, len(data), data))
|
||||
if self.getpacket() != 'OK':
|
||||
raise Exception('Error writing to memory at 0x%08X' % addr)
|
||||
data = bytes(data)
|
||||
|
||||
while data:
|
||||
d = data[:self.PacketSize-44]
|
||||
data = data[len(d):]
|
||||
#print("Writing %d bytes at 0x%X" % (len(d), addr))
|
||||
pack = b"X%08X,%08X:%s" % (addr, len(d), d)
|
||||
self.putpacket(pack)
|
||||
|
||||
if self.getpacket() != b'OK':
|
||||
raise Exception('Error writing to memory at 0x%08X' % addr)
|
||||
addr += len(d)
|
||||
|
||||
def read_regs(self):
|
||||
"""Read target core registers"""
|
||||
self.putpacket("g")
|
||||
self.putpacket(b"g")
|
||||
reply = self.getpacket()
|
||||
if (len(reply) == 0) or (reply[0] == 'E'):
|
||||
if (reply == b'') or (reply[:1] == b'E'):
|
||||
raise Exception('Error reading memory at 0x%08X' % addr)
|
||||
try:
|
||||
data = unhexify(reply)
|
||||
except Excpetion:
|
||||
except Exception:
|
||||
raise Exception('Invalid response to memory read packet: %r' % reply)
|
||||
return struct.unpack("=20L", data)
|
||||
ret = array.array('I',data)
|
||||
return ret
|
||||
|
||||
def write_regs(self, *regs):
|
||||
"""Write target core registers"""
|
||||
data = struct.pack("=%dL" % len(regs), *regs)
|
||||
self.putpacket("G" + hexify(data))
|
||||
if self.getpacket() != 'OK':
|
||||
self.putpacket(b"G" + hexify(data))
|
||||
if self.getpacket() != b'OK':
|
||||
raise Exception('Error writing to target core registers')
|
||||
|
||||
def memmap_read(self):
|
||||
"""Read the XML memory map from target"""
|
||||
offset = 0
|
||||
ret = ''
|
||||
ret = b''
|
||||
while True:
|
||||
self.putpacket("qXfer:memory-map:read::%08X,%08X" % (offset, 512))
|
||||
self.putpacket(b"qXfer:memory-map:read::%08X,%08X" % (offset, 512))
|
||||
reply = self.getpacket()
|
||||
if (reply[0] == 'm') or (reply[0] == 'l'):
|
||||
if (reply[0] in b'ml'):
|
||||
offset += len(reply) - 1
|
||||
ret += reply[1:]
|
||||
else:
|
||||
raise Exception("Invalid GDB stub response")
|
||||
raise Exception('Invalid GDB stub response %r'%reply)
|
||||
|
||||
if reply[0] == 'l': return ret
|
||||
if reply[:1] == b'l':
|
||||
return ret
|
||||
|
||||
def resume(self):
|
||||
"""Resume target execution"""
|
||||
self.putpacket("c")
|
||||
self.putpacket(b'c')
|
||||
|
||||
def interrupt(self):
|
||||
"""Interrupt target execution"""
|
||||
self.sock.send("\x03")
|
||||
self.sock.send(b'\x03')
|
||||
|
||||
def run_stub(self, stub, address, *args):
|
||||
"""Execute a binary stub at address, passing args in core registers."""
|
||||
|
|
@ -205,11 +237,19 @@ class Target:
|
|||
regs[15] = address
|
||||
self.write_regs(*regs)
|
||||
self.resume()
|
||||
reply = self.getpacket()
|
||||
reply = None
|
||||
while not reply:
|
||||
reply = self.getpacket()
|
||||
if not reply.startswith("T05"):
|
||||
raise Exception("Invalid stop response: %r" % reply)
|
||||
if not reply.startswith(b"T05"):
|
||||
message = "Invalid stop response: %r" % reply
|
||||
try:
|
||||
message += {'T02':' (SIGINT)',
|
||||
'T05':' (SIGTRAP)',
|
||||
'T0B':' (SIGSEGV)',
|
||||
'T1D':' (SIGLOST)'}[reply]
|
||||
except KeyError:
|
||||
pass
|
||||
raise Exception(message)
|
||||
|
||||
class FlashMemory:
|
||||
def __init__(self, target, offset, length, blocksize):
|
||||
|
|
@ -217,26 +257,29 @@ class Target:
|
|||
self.offset = offset
|
||||
self.length = length
|
||||
self.blocksize = blocksize
|
||||
self.blocks = list(None for i in range(length / blocksize))
|
||||
self.blocks = list(None for i in range(length // blocksize))
|
||||
|
||||
def prog(self, offset, data):
|
||||
assert type(data)==bytes
|
||||
|
||||
assert ((offset >= self.offset) and
|
||||
(offset + len(data) <= self.offset + self.length))
|
||||
|
||||
while data:
|
||||
index = (offset - self.offset) / self.blocksize
|
||||
index = (offset - self.offset) // self.blocksize
|
||||
bloffset = (offset - self.offset) % self.blocksize
|
||||
bldata = data[:self.blocksize-bloffset]
|
||||
data = data[len(bldata):]; offset += len(bldata)
|
||||
if self.blocks[index] is None: # Initialize a clear block
|
||||
self.blocks[index] = "".join(chr(0xff) for i in range(self.blocksize))
|
||||
self.blocks[index] = bytes(0xff for i in range(self.blocksize))
|
||||
self.blocks[index] = (self.blocks[index][:bloffset] + bldata +
|
||||
self.blocks[index][bloffset+len(bldata):])
|
||||
|
||||
def commit(self, progress_cb=None):
|
||||
totalblocks = 0
|
||||
for b in self.blocks:
|
||||
if b is not None: totalblocks += 1
|
||||
if b is not None:
|
||||
totalblocks += 1
|
||||
|
||||
block = 0
|
||||
for i in range(len(self.blocks)):
|
||||
|
|
@ -247,27 +290,28 @@ class Target:
|
|||
# Erase the block
|
||||
data = self.blocks[i]
|
||||
addr = self.offset + self.blocksize * i
|
||||
if data is None: continue
|
||||
if data is None:
|
||||
continue
|
||||
#print "Erasing flash at 0x%X" % (self.offset + self.blocksize*i)
|
||||
self.target.putpacket("vFlashErase:%08X,%08X" %
|
||||
self.target.putpacket(b"vFlashErase:%08X,%08X" %
|
||||
(self.offset + self.blocksize*i, self.blocksize))
|
||||
if self.target.getpacket() != 'OK':
|
||||
if self.target.getpacket() != b'OK':
|
||||
raise Exception("Failed to erase flash")
|
||||
|
||||
while data:
|
||||
d = data[0:980]
|
||||
data = data[len(d):]
|
||||
#print "Writing %d bytes at 0x%X" % (len(d), addr)
|
||||
self.target.putpacket("vFlashWrite:%08X:%s" % (addr, d))
|
||||
self.target.putpacket(b"vFlashWrite:%08X:%s" % (addr, d))
|
||||
addr += len(d)
|
||||
if self.target.getpacket() != 'OK':
|
||||
if self.target.getpacket() != b'OK':
|
||||
raise Exception("Failed to write flash")
|
||||
|
||||
self.target.putpacket("vFlashDone")
|
||||
if self.target.getpacket() != 'OK':
|
||||
self.target.putpacket(b"vFlashDone")
|
||||
if self.target.getpacket() != b'OK':
|
||||
raise Exception("Failed to commit")
|
||||
|
||||
self.blocks = list(None for i in range(self.length / self.blocksize))
|
||||
self.blocks = list(None for i in range(self.length // self.blocksize))
|
||||
|
||||
|
||||
def flash_probe(self):
|
||||
|
|
@ -275,7 +319,8 @@ class Target:
|
|||
xmldom = parseString(self.memmap_read())
|
||||
|
||||
for memrange in xmldom.getElementsByTagName("memory"):
|
||||
if memrange.getAttribute("type") != "flash": continue
|
||||
if memrange.getAttribute("type") != "flash":
|
||||
continue
|
||||
offset = eval(memrange.getAttribute("start"))
|
||||
length = eval(memrange.getAttribute("length"))
|
||||
for property in memrange.getElementsByTagName("property"):
|
||||
|
|
@ -297,5 +342,3 @@ class Target:
|
|||
def flash_commit(self, progress_cb=None):
|
||||
for m in self.mem:
|
||||
m.commit(progress_cb)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -31,15 +31,16 @@ def flash_write_hex(target, hexfile, progress_cb=None):
|
|||
f = open(hexfile)
|
||||
addrhi = 0
|
||||
for line in f:
|
||||
if line[0] != ':': raise Exception("Error in hex file")
|
||||
if line[0] != ':':
|
||||
raise Exception("Error in hex file")
|
||||
reclen = int(line[1:3], 16)
|
||||
addrlo = int(line[3:7], 16)
|
||||
rectype = int(line[7:9], 16);
|
||||
if sum(ord(x) for x in gdb.unhexify(line[1:11+reclen*2])) & 0xff != 0:
|
||||
if sum(x for x in bytes.fromhex(line[1:11+reclen*2])) & 0xff != 0:
|
||||
raise Exception("Checksum error in hex file")
|
||||
if rectype == 0: # Data record
|
||||
addr = (addrhi << 16) + addrlo
|
||||
data = gdb.unhexify(line[9:9+reclen*2])
|
||||
data = bytes.fromhex(line[9:9+reclen*2])
|
||||
target.flash_write_prepare(addr, data)
|
||||
pass
|
||||
elif rectype == 4: # High address record
|
||||
|
|
@ -55,7 +56,7 @@ def flash_write_hex(target, hexfile, progress_cb=None):
|
|||
try:
|
||||
target.flash_commit(progress_cb)
|
||||
except:
|
||||
print "Flash write failed! Is device protected?\n"
|
||||
print("Flash write failed! Is device protected?\n")
|
||||
exit(-1)
|
||||
|
||||
|
||||
|
|
@ -67,8 +68,8 @@ if __name__ == "__main__":
|
|||
if platform == "linux2":
|
||||
print ("\x1b\x5b\x48\x1b\x5b\x32\x4a") # clear terminal screen
|
||||
print("Black Magic Probe -- Target Production Programming Tool -- version 1.0")
|
||||
print "Copyright (C) 2011 Black Sphere Technologies"
|
||||
print "License GPLv3+: GNU GPL version 3 or later <http://gnu.org/licenses/gpl.html>"
|
||||
print("Copyright (C) 2011 Black Sphere Technologies")
|
||||
print("License GPLv3+: GNU GPL version 3 or later <http://gnu.org/licenses/gpl.html>")
|
||||
print("")
|
||||
|
||||
dev = "COM1" if platform == "win32" else "/dev/ttyACM0"
|
||||
|
|
@ -80,13 +81,20 @@ if __name__ == "__main__":
|
|||
try:
|
||||
opts, args = getopt(argv[1:], "sd:b:t:rR")
|
||||
for opt in opts:
|
||||
if opt[0] == "-s": scan = "swdp_scan"
|
||||
elif opt[0] == "-b": baud = int(opt[1])
|
||||
elif opt[0] == "-d": dev = opt[1]
|
||||
elif opt[0] == "-t": targetno = int(opt[1])
|
||||
elif opt[0] == "-r": unprot = True
|
||||
elif opt[0] == "-R": prot = True
|
||||
else: raise Exception()
|
||||
if opt[0] == "-s":
|
||||
scan = "swdp_scan"
|
||||
elif opt[0] == "-b":
|
||||
baud = int(opt[1])
|
||||
elif opt[0] == "-d":
|
||||
dev = opt[1]
|
||||
elif opt[0] == "-t":
|
||||
targetno = int(opt[1])
|
||||
elif opt[0] == "-r":
|
||||
unprot = True
|
||||
elif opt[0] == "-R":
|
||||
prot = True
|
||||
else:
|
||||
raise Exception()
|
||||
|
||||
hexfile = args[0]
|
||||
except:
|
||||
|
|
@ -101,39 +109,44 @@ if __name__ == "__main__":
|
|||
exit(-1)
|
||||
|
||||
try:
|
||||
s = Serial(dev, baud, timeout=3)
|
||||
s.setDTR(1)
|
||||
while s.read(1024):
|
||||
pass
|
||||
s = Serial(dev) #, baud, timeout=0.1)
|
||||
#s.setDTR(1)
|
||||
#s.flushInput()
|
||||
|
||||
#while s.read(1024):
|
||||
# pass
|
||||
|
||||
target = gdb.Target(s)
|
||||
|
||||
except SerialException, e:
|
||||
except SerialException as e:
|
||||
print("FATAL: Failed to open serial device!\n%s\n" % e[0])
|
||||
exit(-1)
|
||||
|
||||
try:
|
||||
r = target.monitor("version")
|
||||
for s in r: print s,
|
||||
except SerialException, e:
|
||||
for s in r:
|
||||
print(s.decode(), end=' ')
|
||||
except SerialException as e:
|
||||
print("FATAL: Serial communication failure!\n%s\n" % e[0])
|
||||
exit(-1)
|
||||
except: pass
|
||||
#except: pass
|
||||
|
||||
print "Target device scan:"
|
||||
print("Target device scan:")
|
||||
targetlist = None
|
||||
r = target.monitor(scan)
|
||||
for s in r:
|
||||
print s,
|
||||
print
|
||||
print(s.decode(), end=' ')
|
||||
print()
|
||||
|
||||
r = target.monitor("targets")
|
||||
for s in r:
|
||||
if s.startswith("No. Att Driver"): targetlist = []
|
||||
if s.startswith(b"No. Att Driver"):
|
||||
targetlist = []
|
||||
try:
|
||||
if type(targetlist) is list:
|
||||
targetlist.append(int(s[:2]))
|
||||
except: pass
|
||||
except:
|
||||
pass
|
||||
|
||||
#if not targetlist:
|
||||
# print("FATAL: No usable targets found!\n")
|
||||
|
|
@ -161,7 +174,7 @@ if __name__ == "__main__":
|
|||
print("FLASH memory -- Offset: 0x%X BlockSize:0x%X\n" % (m.offset, m.blocksize))
|
||||
|
||||
def progress(percent):
|
||||
print ("Progress: %d%%\r" % percent),
|
||||
print("Progress: %d%%\r" % percent, end=' ')
|
||||
stdout.flush()
|
||||
|
||||
print("Programming target")
|
||||
|
|
@ -179,4 +192,3 @@ if __name__ == "__main__":
|
|||
target.detach()
|
||||
|
||||
print("\nAll operations complete!\n")
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue