lldbreverse.py

importos
importos.path
importlldb
fromlldbsuite.test.lldbtestimport*
fromlldbsuite.test.gdbclientutilsimport*
fromlldbsuite.test.lldbgdbproxyimport*
importlldbgdbserverutils
importre


classThreadSnapshot:
def__init__(self, thread_id, registers):
self.thread_id=thread_id
self.registers=registers


classMemoryBlockSnapshot:
def__init__(self, address, data):
self.address=address
self.data=data


classStateSnapshot:
def__init__(self, thread_snapshots, memory):
self.thread_snapshots=thread_snapshots
self.memory=memory
self.thread_id=None


classRegisterInfo:
def__init__(self, lldb_index, name, bitsize, little_endian):
self.lldb_index=lldb_index
self.name=name
self.bitsize=bitsize
self.little_endian=little_endian


BELOW_STACK_POINTER=16384
ABOVE_STACK_POINTER=4096

BLOCK_SIZE=1024

SOFTWARE_BREAKPOINTS=0
HARDWARE_BREAKPOINTS=1
WRITE_WATCHPOINTS=2


classReverseTestBase(GDBProxyTestBase):
"""
 Base class for tests that need reverse execution.

 This class uses a gdbserver proxy to add very limited reverse-
 execution capability to lldb-server/debugserver for testing
 purposes only.

 To use this class, run the inferior forward until some stopping point.
 Then call `start_recording()` and execute forward again until reaching
 a software breakpoint; this class records the state before each execution executes.
 At that point, the server will accept "bc" and "bs" packets to step
 backwards through the state.
 When executing during recording, we only allow single-step and continue without
 delivering a signal, and only software breakpoint stops are allowed.

 We assume that while recording is enabled, the only effects of instructions
 are on general-purpose registers (read/written by the 'g' and 'G' packets)
 and on memory bytes between [SP - BELOW_STACK_POINTER, SP + ABOVE_STACK_POINTER).
 """

NO_DEBUG_INFO_TESTCASE=True

"""
 A list of StateSnapshots in time order.

 There is one snapshot per single-stepped instruction,
 representing the state before that instruction was
 executed. The last snapshot in the list is the
 snapshot before the last instruction was executed.
 This is an undo log; we snapshot a superset of the state that may have
 been changed by the instruction's execution.
 """
snapshots=None
recording_enabled=False

breakpoints=None

pc_register_info=None
sp_register_info=None
general_purpose_register_info=None

def__init__(self, *args, **kwargs):
GDBProxyTestBase.__init__(self, *args, **kwargs)
self.breakpoints= [set(), set(), set(), set(), set()]

defrespond(self, packet):
ifnotpacket:
raiseValueError("Invalid empty packet")
ifpacket==self.server.PACKET_INTERRUPT:
# Don't send a response. We'll just run to completion.
return []
ifself.is_command(packet, "qSupported", ":"):
# Disable multiprocess support in the server and in LLDB
# since Mac debugserver doesn't support it and we want lldb-server to
# be consistent with that
reply=self.pass_through(packet.replace(";multiprocess", ""))
returnreply.replace(";multiprocess", "") +";ReverseStep+;ReverseContinue+"
ifpacket=="c"orpacket=="s":
packet="vCont;"+packet
elif (
packet[0] =="c"orpacket[0] =="s"orpacket[0] =="C"orpacket[0] =="S"
 ):
raiseValueError(
"Old-style continuation packets with address or signal not supported yet"
 )
ifself.is_command(packet, "vCont", ";"):
ifself.recording_enabled:
returnself.continue_with_recording(packet)
snapshots= []
ifpacket=="bc":
returnself.reverse_continue()
ifpacket=="bs":
returnself.reverse_step()
ifpacket=="jThreadsInfo":
# Suppress this because it contains thread stop reasons which we might
# need to modify, and we don't want to have to implement that.
return""
ifpacket[0] =="x":
# Suppress *binary* reads as results starting with "O" can be mistaken for an output packet
# by the test server code
return""
ifpacket[0] =="z"orpacket[0] =="Z":
reply=self.pass_through(packet)
ifreply=="OK":
self.update_breakpoints(packet)
returnreply
returnGDBProxyTestBase.respond(self, packet)

defstart_recording(self):
self.recording_enabled=True
self.snapshots= []

defstop_recording(self):
"""
 Don't record when executing foward.

 Reverse execution is still supported until the next forward continue.
 """
self.recording_enabled=False

defis_command(self, packet, cmd, follow_token):
returnpacket==cmdorpacket[0 : len(cmd) +1] ==cmd+follow_token

defupdate_breakpoints(self, packet):
m=re.match("([zZ])([01234]),([0-9a-f]+),([0-9a-f]+)", packet)
ifmisNone:
raiseValueError("Invalid breakpoint packet: "+packet)
t=int(m.group(2))
addr=int(m.group(3), 16)
kind=int(m.group(4), 16)
ifm.group(1) =="Z":
self.breakpoints[t].add((addr, kind))
else:
self.breakpoints[t].discard((addr, kind))

defbreakpoint_triggered_at(self, pc):
ifany(addr==pcforaddr, kindinself.breakpoints[SOFTWARE_BREAKPOINTS]):
returnTrue
ifany(addr==pcforaddr, kindinself.breakpoints[HARDWARE_BREAKPOINTS]):
returnTrue
returnFalse

defwatchpoint_triggered(self, new_value_block, current_contents):
"""Returns the address or None."""
forwatch_addr, kindinself.breakpoints[WRITE_WATCHPOINTS]:
foroffsetinrange(0, kind):
addr=watch_addr+offset
if (
addr>=new_value_block.address
andaddr<new_value_block.address+len(new_value_block.data)
 ):
index=addr-new_value_block.address
if (
new_value_block.data[index*2 : (index+1) *2]
!=current_contents[index*2 : (index+1) *2]
 ):
returnwatch_addr
returnNone

defcontinue_with_recording(self, packet):
self.logger.debug("Continue with recording enabled")

step_packet="vCont;s"
ifpacket=="vCont":
requested_step=False
else:
m=re.match("vCont;(c|s)(.*)", packet)
ifmisNone:
raiseValueError("Unsupported vCont packet: "+packet)
requested_step=m.group(1) =="s"
step_packet+=m.group(2)

whileTrue:
snapshot=self.capture_snapshot()
reply=self.pass_through(step_packet)
 (stop_signal, stop_pairs) =self.parse_stop_reply(reply)
ifstop_signal!=5:
raiseValueError("Unexpected stop signal: "+reply)
is_swbreak=False
thread_id=None
forkey, valueinstop_pairs.items():
ifkey=="thread":
thread_id=self.parse_thread_id(value)
continue
ifre.match("[0-9a-f]+", key):
continue
ifkey=="swbreak"or (key=="reason"andvalue=="breakpoint"):
is_swbreak=True
continue
ifkey=="metype":
reason=self.stop_reason_from_mach_exception(stop_pairs)
ifreason=="breakpoint":
is_swbreak=True
elifreason!="singlestep":
raiseValueError(f"Unsupported stop reason in {reply}")
continue
ifkeyin [
"name",
"threads",
"thread-pcs",
"reason",
"mecount",
"medata",
"memory",
 ]:
continue
raiseValueError(f"Unknown stop key '{key}' in {reply}")
ifis_swbreak:
self.logger.debug("Recording stopped")
returnreply
ifthread_idisNone:
returnValueError("Expected thread ID: "+reply)
snapshot.thread_id=thread_id
self.snapshots.append(snapshot)
ifrequested_step:
self.logger.debug("Recording stopped for step")
returnreply

defstop_reason_from_mach_exception(self, stop_pairs):
# See StopInfoMachException::CreateStopReasonWithMachException.
ifint(stop_pairs["metype"]) !=6: # EXC_BREAKPOINT
raiseValueError(f"Unsupported exception type {value} in {reply}")
medata=stop_pairs["medata"]
arch=self.getArchitecture()
ifarchin ["amd64", "i386", "x86_64"]:
ifint(medata[0], 16) ==2:
return"breakpoint"
ifint(medata[0], 16) ==1andint(medata[1], 16) ==0:
return"singlestep"
elifarchin ["arm64", "arm64e"]:
ifint(medata[0], 16) ==1andint(medata[1], 16) !=0:
return"breakpoint"
elifint(medata[0], 16) ==1andint(medata[1], 16) ==0:
return"singlestep"
else:
raiseValueError(f"Unsupported architecture '{arch}'")
raiseValueError(f"Unsupported exception details in {reply}")

defparse_stop_reply(self, reply):
ifnotreply:
raiseValueError("Invalid empty packet")
ifreply[0] =="T"andlen(reply) >=3:
result= {}
fork, vinself.parse_pairs(reply[3:]):
ifkin ["medata", "memory"]:
ifkinresult:
result[k].append(v)
else:
result[k] = [v]
else:
result[k] =v
return (int(reply[1:3], 16), result)
raiseValueError("Unsupported stop reply: "+reply)

defparse_pairs(self, text):
forpairintext.split(";"):
ifnotpair:
continue
m=re.match("([^:]+):(.*)", pair)
ifmisNone:
raiseValueError("Invalid pair text: "+text)
yield (m.group(1), m.group(2))

defcapture_snapshot(self):
"""Snapshot all threads and their stack memories."""
self.ensure_register_info()
current_thread=self.get_current_thread()
thread_snapshots= []
memory= []
forthread_idinself.get_thread_list():
registers= {}
forindexinsorted(self.general_purpose_register_info.keys()):
reply=self.pass_through(f"p{index:x};thread:{thread_id:x};")
ifreply==""orreply[0] =="E":
# Mac debugserver tells us about registers that it won't let
# us actually read. Ignore those registers.
self.logger.debug(f"Failed to read register {index:x}")
continue
registers[index] =reply
thread_snapshot=ThreadSnapshot(thread_id, registers)
thread_sp=self.get_register(
self.sp_register_info, thread_snapshot.registers
 )

# The memory above or below the stack pointer may be mapped, but not
# both readable and writeable. For example on Arm 32-bit Linux, there
# is a "[vectors]" mapping above the stack, which can be read but not
# written to.
#
# Therefore, we should limit any reads to the stack region, which we
# know is readable and writeable.
region_info=self.get_memory_region_info(thread_sp)
lower=max(thread_sp-BELOW_STACK_POINTER, region_info["start"])
upper=min(
thread_sp+ABOVE_STACK_POINTER,
region_info["start"] +region_info["size"],
 )

memory+=self.read_memory(lower, upper)
thread_snapshots.append(thread_snapshot)
self.set_current_thread(current_thread)
returnStateSnapshot(thread_snapshots, memory)

defrestore_snapshot(self, snapshot):
"""
 Restore the snapshot during reverse execution.

 If this triggers a breakpoint or watchpoint, return the stop reply,
 otherwise None.
 """
current_thread=self.get_current_thread()
stop_reasons= []
forthread_snapshotinsnapshot.thread_snapshots:
thread_id=thread_snapshot.thread_id
forlldb_indexinsorted(thread_snapshot.registers.keys()):
data=thread_snapshot.registers[lldb_index]
reply=self.pass_through(
f"P{lldb_index:x}={data};thread:{thread_id:x};"
 )
ifreply!="OK":
try:
reg_name=self.general_purpose_register_info[lldb_index].name
exceptKeyError:
reg_name=f"with index {lldb_index}"
raiseValueError(f"Can't restore thread register {reg_name}")
ifthread_id==snapshot.thread_id:
new_pc=self.get_register(
self.pc_register_info, thread_snapshot.registers
 )
ifself.breakpoint_triggered_at(new_pc):
stop_reasons.append([("reason", "breakpoint")])
self.set_current_thread(current_thread)
forblockinsnapshot.memory:
current_memory=self.pass_through(
f"m{block.address:x},{(len(block.data)//2):x}"
 )
ifnotcurrent_memoryorcurrent_memory[0] =="E":
raiseValueError("Can't read back memory")
reply=self.pass_through(
f"M{block.address:x},{len(block.data)//2:x}:"+block.data
 )
ifreply!="OK":
raiseValueError(
f"Can't restore memory block ranging from 0x{block.address:x} to 0x{block.address+len(block.data):x}."
 )
watch_addr=self.watchpoint_triggered(block, current_memory)
ifwatch_addrisnotNone:
stop_reasons.append(
 [("reason", "watchpoint"), ("watch", f"{watch_addr:x}")]
 )
ifstop_reasons:
pairs=";".join(f"{key}:{value}"forkey, valueinstop_reasons[0])
returnf"T05thread:{snapshot.thread_id:x};{pairs};"
returnNone

defreverse_step(self):
ifnotself.snapshots:
self.logger.debug("Reverse-step at history boundary")
returnself.history_boundary_reply(self.get_current_thread())
self.logger.debug("Reverse-step started")
snapshot=self.snapshots.pop()
stop_reply=self.restore_snapshot(snapshot)
self.set_current_thread(snapshot.thread_id)
self.logger.debug("Reverse-step stopped")
ifstop_replyisNone:
returnself.singlestep_stop_reply(snapshot.thread_id)
returnstop_reply

defreverse_continue(self):
self.logger.debug("Reverse-continue started")
thread_id=None
whileself.snapshots:
snapshot=self.snapshots.pop()
stop_reply=self.restore_snapshot(snapshot)
thread_id=snapshot.thread_id
ifstop_replyisnotNone:
self.set_current_thread(thread_id)
self.logger.debug("Reverse-continue stopped")
returnstop_reply
ifthread_idisNone:
thread_id=self.get_current_thread()
else:
self.set_current_thread(snapshot.thread_id)
self.logger.debug("Reverse-continue stopped at history boundary")
returnself.history_boundary_reply(thread_id)

defget_current_thread(self):
reply=self.pass_through("qC")
returnself.parse_thread_id(reply[2:])

defparse_thread_id(self, thread_id):
m=re.match("([0-9a-f]+)", thread_id)
ifmisNone:
raiseValueError("Invalid thread ID: "+thread_id)
returnint(m.group(1), 16)

defhistory_boundary_reply(self, thread_id):
returnf"T00thread:{thread_id:x};replaylog:begin;"

defsinglestep_stop_reply(self, thread_id):
returnf"T05thread:{thread_id:x};"

defset_current_thread(self, thread_id):
"""
 Set current thread in inner gdbserver.
 """
ifthread_id>=0:
self.pass_through(f"Hg{thread_id:x}")
self.pass_through(f"Hc{thread_id:x}")
else:
self.pass_through(f"Hc-1")
self.pass_through(f"Hg-1")

defget_register(self, register_info, registers):
ifregister_info.bitsize%8!=0:
raiseValueError("Register size must be a multiple of 8 bits")
ifregister_info.lldb_indexnotinregisters:
raiseValueError("Register value not captured")
data=registers[register_info.lldb_index]
num_bytes=register_info.bitsize//8
bytes= []
foriinrange(0, num_bytes):
bytes.append(int(data[i*2 : (i+1) *2], 16))
ifregister_info.little_endian:
bytes.reverse()
result=0
forbyteinbytes:
result= (result<<8) +byte
returnresult

defget_memory_region_info(self, addr):
reply=self.pass_through(f"qMemoryRegionInfo:{addr:x}")
ifnotreplyorreply[0] =="E":
raiseRuntimeError("Failed to get memory region info.")

# Valid reply looks like:
# start:fffcf000;size:21000;permissions:rw;flags:;name:5b737461636b5d;
values= [vforvinreply.strip().split(";") ifv]
region_info= {}
forvalueinvalues:
key, value=value.split(
":",
 )
region_info[key] =value

ifnot ("start"inregion_infoand"size"inregion_info):
raiseRuntimeError("Did not get extent of memory region.")

region_info["start"] =int(region_info["start"], 16)
region_info["size"] =int(region_info["size"], 16)

returnregion_info

defread_memory(self, start_addr, end_addr):
"""
 Read a region of memory from the target.

 Some of the addresses may extend into memory we cannot read, skip those.

 Return a list of blocks containing the valid area(s) in the
 requested range.
 """
regions= []
start_addr=start_addr- (start_addr%BLOCK_SIZE)
ifend_addr%BLOCK_SIZE>0:
end_addr=end_addr- (end_addr%BLOCK_SIZE) +BLOCK_SIZE
foraddrinrange(start_addr, end_addr, BLOCK_SIZE):
reply=self.pass_through(f"m{addr:x},{(BLOCK_SIZE-1):x}")
ifreplyandreply[0] !="E":
block=MemoryBlockSnapshot(addr, reply)
regions.append(block)
returnregions

defensure_register_info(self):
ifself.general_purpose_register_infoisnotNone:
return
reply=self.pass_through("qHostInfo")
little_endian=any(
kv== ("endian", "little") forkvinself.parse_pairs(reply)
 )
self.general_purpose_register_info= {}
lldb_index=0
whileTrue:
reply=self.pass_through(f"qRegisterInfo{lldb_index:x}")
ifnotreplyorreply[0] =="E":
break
info= {k: vfork, vinself.parse_pairs(reply)}
reg_info=RegisterInfo(
lldb_index, info["name"], int(info["bitsize"]), little_endian
 )
if (
info["set"] =="General Purpose Registers"
andnot"container-regs"ininfo
 ):
self.general_purpose_register_info[lldb_index] =reg_info
if"generic"ininfo:
ifinfo["generic"] =="pc":
self.pc_register_info=reg_info
elifinfo["generic"] =="sp":
self.sp_register_info=reg_info
lldb_index+=1
ifself.pc_register_infoisNoneorself.sp_register_infoisNone:
raiseValueError("Can't find generic pc or sp register")

defget_thread_list(self):
threads= []
reply=self.pass_through("qfThreadInfo")
whileTrue:
ifnotreply:
raiseValueError("Missing reply packet")
ifreply[0] =="m":
foridinreply[1:].split(","):
threads.append(self.parse_thread_id(id))
elifreply[0] =="l":
returnthreads
reply=self.pass_through("qsThreadInfo")