| #!/usr/bin/env python | |
| import binascii | |
| import json | |
| import optparse | |
| import os | |
| import pprint | |
| import socket | |
| import string | |
| import subprocess | |
| import sys | |
| import threading | |
| import time | |
| def dump_memory(base_addr, data, num_per_line, outfile): | |
| data_len = len(data) | |
| hex_string = binascii.hexlify(data) | |
| addr = base_addr | |
| ascii_str = "" | |
| i = 0 | |
| while i < data_len: | |
| outfile.write("0x%8.8x: " % (addr + i)) | |
| bytes_left = data_len - i | |
| if bytes_left >= num_per_line: | |
| curr_data_len = num_per_line | |
| else: | |
| curr_data_len = bytes_left | |
| hex_start_idx = i * 2 | |
| hex_end_idx = hex_start_idx + curr_data_len * 2 | |
| curr_hex_str = hex_string[hex_start_idx:hex_end_idx] | |
| # 'curr_hex_str' now contains the hex byte string for the | |
| # current line with no spaces between bytes | |
| t = iter(curr_hex_str) | |
| # Print hex bytes separated by space | |
| outfile.write(" ".join(a + b for a, b in zip(t, t))) | |
| # Print two spaces | |
| outfile.write(" ") | |
| # Calculate ASCII string for bytes into 'ascii_str' | |
| ascii_str = "" | |
| for j in range(i, i + curr_data_len): | |
| ch = data[j] | |
| if ch in string.printable and ch not in string.whitespace: | |
| ascii_str += "%c" % (ch) | |
| else: | |
| ascii_str += "." | |
| # Print ASCII representation and newline | |
| outfile.write(ascii_str) | |
| i = i + curr_data_len | |
| outfile.write("\n") | |
| def read_packet(f, verbose=False, trace_file=None): | |
| """Decode a JSON packet that starts with the content length and is | |
| followed by the JSON bytes from a file 'f'. Returns None on EOF. | |
| """ | |
| line = f.readline().decode("utf-8") | |
| if len(line) == 0: | |
| return None # EOF. | |
| # Watch for line that starts with the prefix | |
| prefix = "Content-Length: " | |
| if line.startswith(prefix): | |
| # Decode length of JSON bytes | |
| if verbose: | |
| print('content: "%s"' % (line)) | |
| length = int(line[len(prefix) :]) | |
| if verbose: | |
| print('length: "%u"' % (length)) | |
| # Skip empty line | |
| line = f.readline() | |
| if verbose: | |
| print('empty: "%s"' % (line)) | |
| # Read JSON bytes | |
| json_str = f.read(length) | |
| if verbose: | |
| print('json: "%s"' % (json_str)) | |
| if trace_file: | |
| trace_file.write("from adaptor:\n%s\n" % (json_str)) | |
| # Decode the JSON bytes into a python dictionary | |
| return json.loads(json_str) | |
| raise Exception("unexpected malformed message from lldb-dap: " + line) | |
| def packet_type_is(packet, packet_type): | |
| return "type" in packet and packet["type"] == packet_type | |
| def dump_dap_log(log_file): | |
| print("========= DEBUG ADAPTER PROTOCOL LOGS =========") | |
| if log_file is None: | |
| print("no log file available") | |
| else: | |
| with open(log_file, "r") as file: | |
| print(file.read()) | |
| print("========= END =========") | |
| def read_packet_thread(vs_comm, log_file): | |
| done = False | |
| try: | |
| while not done: | |
| packet = read_packet(vs_comm.recv, trace_file=vs_comm.trace_file) | |
| # `packet` will be `None` on EOF. We want to pass it down to | |
| # handle_recv_packet anyway so the main thread can handle unexpected | |
| # termination of lldb-dap and stop waiting for new packets. | |
| done = not vs_comm.handle_recv_packet(packet) | |
| finally: | |
| dump_dap_log(log_file) | |
| class DebugCommunication(object): | |
| def __init__(self, recv, send, init_commands, log_file=None): | |
| self.trace_file = None | |
| self.send = send | |
| self.recv = recv | |
| self.recv_packets = [] | |
| self.recv_condition = threading.Condition() | |
| self.recv_thread = threading.Thread( | |
| target=read_packet_thread, args=(self, log_file) | |
| ) | |
| self.process_event_body = None | |
| self.exit_status = None | |
| self.initialize_body = None | |
| self.thread_stop_reasons = {} | |
| self.breakpoint_events = [] | |
| self.progress_events = [] | |
| self.reverse_requests = [] | |
| self.sequence = 1 | |
| self.threads = None | |
| self.recv_thread.start() | |
| self.output_condition = threading.Condition() | |
| self.output = {} | |
| self.configuration_done_sent = False | |
| self.frame_scopes = {} | |
| self.init_commands = init_commands | |
| self.disassembled_instructions = {} | |
| @classmethod | |
| def encode_content(cls, s): | |
| return ("Content-Length: %u\r\n\r\n%s" % (len(s), s)).encode("utf-8") | |
| @classmethod | |
| def validate_response(cls, command, response): | |
| if command["command"] != response["command"]: | |
| raise ValueError("command mismatch in response") | |
| if command["seq"] != response["request_seq"]: | |
| raise ValueError("seq mismatch in response") | |
| def get_modules(self): | |
| module_list = self.request_modules()["body"]["modules"] | |
| modules = {} | |
| for module in module_list: | |
| modules[module["name"]] = module | |
| return modules | |
| def get_output(self, category, timeout=0.0, clear=True): | |
| self.output_condition.acquire() | |
| output = None | |
| if category in self.output: | |
| output = self.output[category] | |
| if clear: | |
| del self.output[category] | |
| elif timeout != 0.0: | |
| self.output_condition.wait(timeout) | |
| if category in self.output: | |
| output = self.output[category] | |
| if clear: | |
| del self.output[category] | |
| self.output_condition.release() | |
| return output | |
| def collect_output(self, category, timeout_secs, pattern, clear=True): | |
| end_time = time.time() + timeout_secs | |
| collected_output = "" | |
| while end_time > time.time(): | |
| output = self.get_output(category, timeout=0.25, clear=clear) | |
| if output: | |
| collected_output += output | |
| if pattern is not None and pattern in output: | |
| break | |
| return collected_output if collected_output else None | |
| def enqueue_recv_packet(self, packet): | |
| self.recv_condition.acquire() | |
| self.recv_packets.append(packet) | |
| self.recv_condition.notify() | |
| self.recv_condition.release() | |
| def handle_recv_packet(self, packet): | |
| """Called by the read thread that is waiting for all incoming packets | |
| to store the incoming packet in "self.recv_packets" in a thread safe | |
| way. This function will then signal the "self.recv_condition" to | |
| indicate a new packet is available. Returns True if the caller | |
| should keep calling this function for more packets. | |
| """ | |
| # If EOF, notify the read thread by enqueuing a None. | |
| if not packet: | |
| self.enqueue_recv_packet(None) | |
| return False | |
| # Check the packet to see if is an event packet | |
| keepGoing = True | |
| packet_type = packet["type"] | |
| if packet_type == "event": | |
| event = packet["event"] | |
| body = None | |
| if "body" in packet: | |
| body = packet["body"] | |
| # Handle the event packet and cache information from these packets | |
| # as they come in | |
| if event == "output": | |
| # Store any output we receive so clients can retrieve it later. | |
| category = body["category"] | |
| output = body["output"] | |
| self.output_condition.acquire() | |
| if category in self.output: | |
| self.output[category] += output | |
| else: | |
| self.output[category] = output | |
| self.output_condition.notify() | |
| self.output_condition.release() | |
| # no need to add 'output' event packets to our packets list | |
| return keepGoing | |
| elif event == "process": | |
| # When a new process is attached or launched, remember the | |
| # details that are available in the body of the event | |
| self.process_event_body = body | |
| elif event == "stopped": | |
| # Each thread that stops with a reason will send a | |
| # 'stopped' event. We need to remember the thread stop | |
| # reasons since the 'threads' command doesn't return | |
| # that information. | |
| self._process_stopped() | |
| tid = body["threadId"] | |
| self.thread_stop_reasons[tid] = body | |
| elif event == "breakpoint": | |
| # Breakpoint events come in when a breakpoint has locations | |
| # added or removed. Keep track of them so we can look for them | |
| # in tests. | |
| self.breakpoint_events.append(packet) | |
| # no need to add 'breakpoint' event packets to our packets list | |
| return keepGoing | |
| elif event.startswith("progress"): | |
| # Progress events come in as 'progressStart', 'progressUpdate', | |
| # and 'progressEnd' events. Keep these around in case test | |
| # cases want to verify them. | |
| self.progress_events.append(packet) | |
| # No need to add 'progress' event packets to our packets list. | |
| return keepGoing | |
| elif packet_type == "response": | |
| if packet["command"] == "disconnect": | |
| keepGoing = False | |
| self.enqueue_recv_packet(packet) | |
| return keepGoing | |
| def send_packet(self, command_dict, set_sequence=True): | |
| """Take the "command_dict" python dictionary and encode it as a JSON | |
| string and send the contents as a packet to the VSCode debug | |
| adaptor""" | |
| # Set the sequence ID for this command automatically | |
| if set_sequence: | |
| command_dict["seq"] = self.sequence | |
| self.sequence += 1 | |
| # Encode our command dictionary as a JSON string | |
| json_str = json.dumps(command_dict, separators=(",", ":")) | |
| if self.trace_file: | |
| self.trace_file.write("to adaptor:\n%s\n" % (json_str)) | |
| length = len(json_str) | |
| if length > 0: | |
| # Send the encoded JSON packet and flush the 'send' file | |
| self.send.write(self.encode_content(json_str)) | |
| self.send.flush() | |
| def recv_packet(self, filter_type=None, filter_event=None, timeout=None): | |
| """Get a JSON packet from the VSCode debug adaptor. This function | |
| assumes a thread that reads packets is running and will deliver | |
| any received packets by calling handle_recv_packet(...). This | |
| function will wait for the packet to arrive and return it when | |
| it does.""" | |
| while True: | |
| try: | |
| self.recv_condition.acquire() | |
| packet = None | |
| while True: | |
| for i, curr_packet in enumerate(self.recv_packets): | |
| if not curr_packet: | |
| raise EOFError | |
| packet_type = curr_packet["type"] | |
| if filter_type is None or packet_type in filter_type: | |
| if filter_event is None or ( | |
| packet_type == "event" | |
| and curr_packet["event"] in filter_event | |
| ): | |
| packet = self.recv_packets.pop(i) | |
| break | |
| if packet: | |
| break | |
| # Sleep until packet is received | |
| len_before = len(self.recv_packets) | |
| self.recv_condition.wait(timeout) | |
| len_after = len(self.recv_packets) | |
| if len_before == len_after: | |
| return None # Timed out | |
| return packet | |
| except EOFError: | |
| return None | |
| finally: | |
| self.recv_condition.release() | |
| return None | |
| def send_recv(self, command): | |
| """Send a command python dictionary as JSON and receive the JSON | |
| response. Validates that the response is the correct sequence and | |
| command in the reply. Any events that are received are added to the | |
| events list in this object""" | |
| self.send_packet(command) | |
| done = False | |
| while not done: | |
| response_or_request = self.recv_packet(filter_type=["response", "request"]) | |
| if response_or_request is None: | |
| desc = 'no response for "%s"' % (command["command"]) | |
| raise ValueError(desc) | |
| if response_or_request["type"] == "response": | |
| self.validate_response(command, response_or_request) | |
| return response_or_request | |
| else: | |
| self.reverse_requests.append(response_or_request) | |
| if response_or_request["command"] == "runInTerminal": | |
| subprocess.Popen( | |
| response_or_request["arguments"]["args"], | |
| env=response_or_request["arguments"]["env"], | |
| ) | |
| self.send_packet( | |
| { | |
| "type": "response", | |
| "seq": -1, | |
| "request_seq": response_or_request["seq"], | |
| "success": True, | |
| "command": "runInTerminal", | |
| "body": {}, | |
| }, | |
| set_sequence=False, | |
| ) | |
| elif response_or_request["command"] == "startDebugging": | |
| self.send_packet( | |
| { | |
| "type": "response", | |
| "seq": -1, | |
| "request_seq": response_or_request["seq"], | |
| "success": True, | |
| "command": "startDebugging", | |
| "body": {}, | |
| }, | |
| set_sequence=False, | |
| ) | |
| else: | |
| desc = 'unknown reverse request "%s"' % ( | |
| response_or_request["command"] | |
| ) | |
| raise ValueError(desc) | |
| return None | |
| def wait_for_event(self, filter=None, timeout=None): | |
| while True: | |
| return self.recv_packet( | |
| filter_type="event", filter_event=filter, timeout=timeout | |
| ) | |
| return None | |
| def wait_for_stopped(self, timeout=None): | |
| stopped_events = [] | |
| stopped_event = self.wait_for_event( | |
| filter=["stopped", "exited"], timeout=timeout | |
| ) | |
| exited = False | |
| while stopped_event: | |
| stopped_events.append(stopped_event) | |
| # If we exited, then we are done | |
| if stopped_event["event"] == "exited": | |
| self.exit_status = stopped_event["body"]["exitCode"] | |
| exited = True | |
| break | |
| # Otherwise we stopped and there might be one or more 'stopped' | |
| # events for each thread that stopped with a reason, so keep | |
| # checking for more 'stopped' events and return all of them | |
| stopped_event = self.wait_for_event(filter="stopped", timeout=0.25) | |
| if exited: | |
| self.threads = [] | |
| return stopped_events | |
| def wait_for_exited(self): | |
| event_dict = self.wait_for_event("exited") | |
| if event_dict is None: | |
| raise ValueError("didn't get exited event") | |
| return event_dict | |
| def wait_for_terminated(self): | |
| event_dict = self.wait_for_event("terminated") | |
| if event_dict is None: | |
| raise ValueError("didn't get terminated event") | |
| return event_dict | |
| def get_initialize_value(self, key): | |
| """Get a value for the given key if it there is a key/value pair in | |
| the "initialize" request response body. | |
| """ | |
| if self.initialize_body and key in self.initialize_body: | |
| return self.initialize_body[key] | |
| return None | |
| def get_threads(self): | |
| if self.threads is None: | |
| self.request_threads() | |
| return self.threads | |
| def get_thread_id(self, threadIndex=0): | |
| """Utility function to get the first thread ID in the thread list. | |
| If the thread list is empty, then fetch the threads. | |
| """ | |
| if self.threads is None: | |
| self.request_threads() | |
| if self.threads and threadIndex < len(self.threads): | |
| return self.threads[threadIndex]["id"] | |
| return None | |
| def get_stackFrame(self, frameIndex=0, threadId=None): | |
| """Get a single "StackFrame" object from a "stackTrace" request and | |
| return the "StackFrame" as a python dictionary, or None on failure | |
| """ | |
| if threadId is None: | |
| threadId = self.get_thread_id() | |
| if threadId is None: | |
| print("invalid threadId") | |
| return None | |
| response = self.request_stackTrace(threadId, startFrame=frameIndex, levels=1) | |
| if response: | |
| return response["body"]["stackFrames"][0] | |
| print("invalid response") | |
| return None | |
| def get_completions(self, text, frameId=None): | |
| if frameId is None: | |
| stackFrame = self.get_stackFrame() | |
| frameId = stackFrame["id"] | |
| response = self.request_completions(text, frameId) | |
| return response["body"]["targets"] | |
| def get_scope_variables(self, scope_name, frameIndex=0, threadId=None, is_hex=None): | |
| stackFrame = self.get_stackFrame(frameIndex=frameIndex, threadId=threadId) | |
| if stackFrame is None: | |
| return [] | |
| frameId = stackFrame["id"] | |
| if frameId in self.frame_scopes: | |
| frame_scopes = self.frame_scopes[frameId] | |
| else: | |
| scopes_response = self.request_scopes(frameId) | |
| frame_scopes = scopes_response["body"]["scopes"] | |
| self.frame_scopes[frameId] = frame_scopes | |
| for scope in frame_scopes: | |
| if scope["name"] == scope_name: | |
| varRef = scope["variablesReference"] | |
| variables_response = self.request_variables(varRef, is_hex=is_hex) | |
| if variables_response: | |
| if "body" in variables_response: | |
| body = variables_response["body"] | |
| if "variables" in body: | |
| vars = body["variables"] | |
| return vars | |
| return [] | |
| def get_global_variables(self, frameIndex=0, threadId=None): | |
| return self.get_scope_variables( | |
| "Globals", frameIndex=frameIndex, threadId=threadId | |
| ) | |
| def get_local_variables(self, frameIndex=0, threadId=None, is_hex=None): | |
| return self.get_scope_variables( | |
| "Locals", frameIndex=frameIndex, threadId=threadId, is_hex=is_hex | |
| ) | |
| def get_registers(self, frameIndex=0, threadId=None): | |
| return self.get_scope_variables( | |
| "Registers", frameIndex=frameIndex, threadId=threadId | |
| ) | |
| def get_local_variable(self, name, frameIndex=0, threadId=None, is_hex=None): | |
| locals = self.get_local_variables( | |
| frameIndex=frameIndex, threadId=threadId, is_hex=is_hex | |
| ) | |
| for local in locals: | |
| if "name" in local and local["name"] == name: | |
| return local | |
| return None | |
| def get_local_variable_value(self, name, frameIndex=0, threadId=None, is_hex=None): | |
| variable = self.get_local_variable( | |
| name, frameIndex=frameIndex, threadId=threadId, is_hex=is_hex | |
| ) | |
| if variable and "value" in variable: | |
| return variable["value"] | |
| return None | |
| def get_local_variable_child( | |
| self, name, child_name, frameIndex=0, threadId=None, is_hex=None | |
| ): | |
| local = self.get_local_variable(name, frameIndex, threadId) | |
| if local["variablesReference"] == 0: | |
| return None | |
| children = self.request_variables(local["variablesReference"], is_hex=is_hex)[ | |
| "body" | |
| ]["variables"] | |
| for child in children: | |
| if child["name"] == child_name: | |
| return child | |
| return None | |
| def replay_packets(self, replay_file_path): | |
| f = open(replay_file_path, "r") | |
| mode = "invalid" | |
| set_sequence = False | |
| command_dict = None | |
| while mode != "eof": | |
| if mode == "invalid": | |
| line = f.readline() | |
| if line.startswith("to adapter:"): | |
| mode = "send" | |
| elif line.startswith("from adapter:"): | |
| mode = "recv" | |
| elif mode == "send": | |
| command_dict = read_packet(f) | |
| # Skip the end of line that follows the JSON | |
| f.readline() | |
| if command_dict is None: | |
| raise ValueError("decode packet failed from replay file") | |
| print("Sending:") | |
| pprint.PrettyPrinter(indent=2).pprint(command_dict) | |
| # raw_input('Press ENTER to send:') | |
| self.send_packet(command_dict, set_sequence) | |
| mode = "invalid" | |
| elif mode == "recv": | |
| print("Replay response:") | |
| replay_response = read_packet(f) | |
| # Skip the end of line that follows the JSON | |
| f.readline() | |
| pprint.PrettyPrinter(indent=2).pprint(replay_response) | |
| actual_response = self.recv_packet() | |
| if actual_response: | |
| type = actual_response["type"] | |
| print("Actual response:") | |
| if type == "response": | |
| self.validate_response(command_dict, actual_response) | |
| pprint.PrettyPrinter(indent=2).pprint(actual_response) | |
| else: | |
| print("error: didn't get a valid response") | |
| mode = "invalid" | |
| def request_attach( | |
| self, | |
| program=None, | |
| pid=None, | |
| waitFor=None, | |
| trace=None, | |
| initCommands=None, | |
| preRunCommands=None, | |
| stopCommands=None, | |
| exitCommands=None, | |
| attachCommands=None, | |
| terminateCommands=None, | |
| coreFile=None, | |
| postRunCommands=None, | |
| sourceMap=None, | |
| gdbRemotePort=None, | |
| gdbRemoteHostname=None, | |
| ): | |
| args_dict = {} | |
| if pid is not None: | |
| args_dict["pid"] = pid | |
| if program is not None: | |
| args_dict["program"] = program | |
| if waitFor is not None: | |
| args_dict["waitFor"] = waitFor | |
| if trace: | |
| args_dict["trace"] = trace | |
| args_dict["initCommands"] = self.init_commands | |
| if initCommands: | |
| args_dict["initCommands"].extend(initCommands) | |
| if preRunCommands: | |
| args_dict["preRunCommands"] = preRunCommands | |
| if stopCommands: | |
| args_dict["stopCommands"] = stopCommands | |
| if exitCommands: | |
| args_dict["exitCommands"] = exitCommands | |
| if terminateCommands: | |
| args_dict["terminateCommands"] = terminateCommands | |
| if attachCommands: | |
| args_dict["attachCommands"] = attachCommands | |
| if coreFile: | |
| args_dict["coreFile"] = coreFile | |
| if postRunCommands: | |
| args_dict["postRunCommands"] = postRunCommands | |
| if sourceMap: | |
| args_dict["sourceMap"] = sourceMap | |
| if gdbRemotePort is not None: | |
| args_dict["gdb-remote-port"] = gdbRemotePort | |
| if gdbRemoteHostname is not None: | |
| args_dict["gdb-remote-hostname"] = gdbRemoteHostname | |
| command_dict = {"command": "attach", "type": "request", "arguments": args_dict} | |
| return self.send_recv(command_dict) | |
| def request_configurationDone(self): | |
| command_dict = { | |
| "command": "configurationDone", | |
| "type": "request", | |
| "arguments": {}, | |
| } | |
| response = self.send_recv(command_dict) | |
| if response: | |
| self.configuration_done_sent = True | |
| return response | |
| def _process_stopped(self): | |
| self.threads = None | |
| self.frame_scopes = {} | |
| def request_continue(self, threadId=None): | |
| if self.exit_status is not None: | |
| raise ValueError("request_continue called after process exited") | |
| # If we have launched or attached, then the first continue is done by | |
| # sending the 'configurationDone' request | |
| if not self.configuration_done_sent: | |
| return self.request_configurationDone() | |
| args_dict = {} | |
| if threadId is None: | |
| threadId = self.get_thread_id() | |
| args_dict["threadId"] = threadId | |
| command_dict = { | |
| "command": "continue", | |
| "type": "request", | |
| "arguments": args_dict, | |
| } | |
| response = self.send_recv(command_dict) | |
| # Caller must still call wait_for_stopped. | |
| return response | |
| def request_restart(self, restartArguments=None): | |
| command_dict = { | |
| "command": "restart", | |
| "type": "request", | |
| } | |
| if restartArguments: | |
| command_dict["arguments"] = restartArguments | |
| response = self.send_recv(command_dict) | |
| # Caller must still call wait_for_stopped. | |
| return response | |
| def request_disconnect(self, terminateDebuggee=None): | |
| args_dict = {} | |
| if terminateDebuggee is not None: | |
| if terminateDebuggee: | |
| args_dict["terminateDebuggee"] = True | |
| else: | |
| args_dict["terminateDebuggee"] = False | |
| command_dict = { | |
| "command": "disconnect", | |
| "type": "request", | |
| "arguments": args_dict, | |
| } | |
| return self.send_recv(command_dict) | |
| def request_disassemble( | |
| self, memoryReference, offset=-50, instructionCount=200, resolveSymbols=True | |
| ): | |
| args_dict = { | |
| "memoryReference": memoryReference, | |
| "offset": offset, | |
| "instructionCount": instructionCount, | |
| "resolveSymbols": resolveSymbols, | |
| } | |
| command_dict = { | |
| "command": "disassemble", | |
| "type": "request", | |
| "arguments": args_dict, | |
| } | |
| instructions = self.send_recv(command_dict)["body"]["instructions"] | |
| for inst in instructions: | |
| self.disassembled_instructions[inst["address"]] = inst | |
| def request_readMemory(self, memoryReference, offset, count): | |
| args_dict = { | |
| "memoryReference": memoryReference, | |
| "offset": offset, | |
| "count": count, | |
| } | |
| command_dict = { | |
| "command": "readMemory", | |
| "type": "request", | |
| "arguments": args_dict, | |
| } | |
| return self.send_recv(command_dict) | |
| def request_evaluate(self, expression, frameIndex=0, threadId=None, context=None): | |
| stackFrame = self.get_stackFrame(frameIndex=frameIndex, threadId=threadId) | |
| if stackFrame is None: | |
| return [] | |
| args_dict = { | |
| "expression": expression, | |
| "context": context, | |
| "frameId": stackFrame["id"], | |
| } | |
| command_dict = { | |
| "command": "evaluate", | |
| "type": "request", | |
| "arguments": args_dict, | |
| } | |
| return self.send_recv(command_dict) | |
| def request_exceptionInfo(self, threadId=None): | |
| if threadId is None: | |
| threadId = self.get_thread_id() | |
| args_dict = {"threadId": threadId} | |
| command_dict = { | |
| "command": "exceptionInfo", | |
| "type": "request", | |
| "arguments": args_dict, | |
| } | |
| return self.send_recv(command_dict) | |
| def request_initialize(self, sourceInitFile): | |
| command_dict = { | |
| "command": "initialize", | |
| "type": "request", | |
| "arguments": { | |
| "adapterID": "lldb-native", | |
| "clientID": "vscode", | |
| "columnsStartAt1": True, | |
| "linesStartAt1": True, | |
| "locale": "en-us", | |
| "pathFormat": "path", | |
| "supportsRunInTerminalRequest": True, | |
| "supportsVariablePaging": True, | |
| "supportsVariableType": True, | |
| "supportsStartDebuggingRequest": True, | |
| "sourceInitFile": sourceInitFile, | |
| }, | |
| } | |
| response = self.send_recv(command_dict) | |
| if response: | |
| if "body" in response: | |
| self.initialize_body = response["body"] | |
| return response | |
| def request_launch( | |
| self, | |
| program, | |
| args=None, | |
| cwd=None, | |
| env=None, | |
| stopOnEntry=False, | |
| disableASLR=True, | |
| disableSTDIO=False, | |
| shellExpandArguments=False, | |
| trace=False, | |
| initCommands=None, | |
| preRunCommands=None, | |
| stopCommands=None, | |
| exitCommands=None, | |
| terminateCommands=None, | |
| sourcePath=None, | |
| debuggerRoot=None, | |
| launchCommands=None, | |
| sourceMap=None, | |
| runInTerminal=False, | |
| postRunCommands=None, | |
| enableAutoVariableSummaries=False, | |
| displayExtendedBacktrace=False, | |
| enableSyntheticChildDebugging=False, | |
| commandEscapePrefix=None, | |
| customFrameFormat=None, | |
| customThreadFormat=None, | |
| ): | |
| args_dict = {"program": program} | |
| if args: | |
| args_dict["args"] = args | |
| if cwd: | |
| args_dict["cwd"] = cwd | |
| if env: | |
| args_dict["env"] = env | |
| if stopOnEntry: | |
| args_dict["stopOnEntry"] = stopOnEntry | |
| if disableSTDIO: | |
| args_dict["disableSTDIO"] = disableSTDIO | |
| if shellExpandArguments: | |
| args_dict["shellExpandArguments"] = shellExpandArguments | |
| if trace: | |
| args_dict["trace"] = trace | |
| args_dict["initCommands"] = self.init_commands | |
| if initCommands: | |
| args_dict["initCommands"].extend(initCommands) | |
| if preRunCommands: | |
| args_dict["preRunCommands"] = preRunCommands | |
| if stopCommands: | |
| args_dict["stopCommands"] = stopCommands | |
| if exitCommands: | |
| args_dict["exitCommands"] = exitCommands | |
| if terminateCommands: | |
| args_dict["terminateCommands"] = terminateCommands | |
| if sourcePath: | |
| args_dict["sourcePath"] = sourcePath | |
| if debuggerRoot: | |
| args_dict["debuggerRoot"] = debuggerRoot | |
| if launchCommands: | |
| args_dict["launchCommands"] = launchCommands | |
| if sourceMap: | |
| args_dict["sourceMap"] = sourceMap | |
| if runInTerminal: | |
| args_dict["runInTerminal"] = runInTerminal | |
| if postRunCommands: | |
| args_dict["postRunCommands"] = postRunCommands | |
| if customFrameFormat: | |
| args_dict["customFrameFormat"] = customFrameFormat | |
| if customThreadFormat: | |
| args_dict["customThreadFormat"] = customThreadFormat | |
| args_dict["disableASLR"] = disableASLR | |
| args_dict["enableAutoVariableSummaries"] = enableAutoVariableSummaries | |
| args_dict["enableSyntheticChildDebugging"] = enableSyntheticChildDebugging | |
| args_dict["displayExtendedBacktrace"] = displayExtendedBacktrace | |
| args_dict["commandEscapePrefix"] = commandEscapePrefix | |
| command_dict = {"command": "launch", "type": "request", "arguments": args_dict} | |
| response = self.send_recv(command_dict) | |
| if response["success"]: | |
| # Wait for a 'process' and 'initialized' event in any order | |
| self.wait_for_event(filter=["process", "initialized"]) | |
| self.wait_for_event(filter=["process", "initialized"]) | |
| return response | |
| def request_next(self, threadId, granularity="statement"): | |
| if self.exit_status is not None: | |
| raise ValueError("request_continue called after process exited") | |
| args_dict = {"threadId": threadId, "granularity": granularity} | |
| command_dict = {"command": "next", "type": "request", "arguments": args_dict} | |
| return self.send_recv(command_dict) | |
| def request_stepIn(self, threadId, targetId, granularity="statement"): | |
| if self.exit_status is not None: | |
| raise ValueError("request_stepIn called after process exited") | |
| args_dict = { | |
| "threadId": threadId, | |
| "targetId": targetId, | |
| "granularity": granularity, | |
| } | |
| command_dict = {"command": "stepIn", "type": "request", "arguments": args_dict} | |
| return self.send_recv(command_dict) | |
| def request_stepInTargets(self, frameId): | |
| if self.exit_status is not None: | |
| raise ValueError("request_stepInTargets called after process exited") | |
| args_dict = {"frameId": frameId} | |
| command_dict = { | |
| "command": "stepInTargets", | |
| "type": "request", | |
| "arguments": args_dict, | |
| } | |
| return self.send_recv(command_dict) | |
| def request_stepOut(self, threadId): | |
| if self.exit_status is not None: | |
| raise ValueError("request_stepOut called after process exited") | |
| args_dict = {"threadId": threadId} | |
| command_dict = {"command": "stepOut", "type": "request", "arguments": args_dict} | |
| return self.send_recv(command_dict) | |
| def request_pause(self, threadId=None): | |
| if self.exit_status is not None: | |
| raise ValueError("request_pause called after process exited") | |
| if threadId is None: | |
| threadId = self.get_thread_id() | |
| args_dict = {"threadId": threadId} | |
| command_dict = {"command": "pause", "type": "request", "arguments": args_dict} | |
| return self.send_recv(command_dict) | |
| def request_scopes(self, frameId): | |
| args_dict = {"frameId": frameId} | |
| command_dict = {"command": "scopes", "type": "request", "arguments": args_dict} | |
| return self.send_recv(command_dict) | |
| def request_setBreakpoints(self, file_path, line_array, data=None): | |
| """data is array of parameters for breakpoints in line_array. | |
| Each parameter object is 1:1 mapping with entries in line_entry. | |
| It contains optional location/hitCondition/logMessage parameters. | |
| """ | |
| (dir, base) = os.path.split(file_path) | |
| source_dict = {"name": base, "path": file_path} | |
| args_dict = { | |
| "source": source_dict, | |
| "sourceModified": False, | |
| } | |
| if line_array is not None: | |
| args_dict["lines"] = "%s" % line_array | |
| breakpoints = [] | |
| for i, line in enumerate(line_array): | |
| breakpoint_data = None | |
| if data is not None and i < len(data): | |
| breakpoint_data = data[i] | |
| bp = {"line": line} | |
| if breakpoint_data is not None: | |
| if "condition" in breakpoint_data and breakpoint_data["condition"]: | |
| bp["condition"] = breakpoint_data["condition"] | |
| if ( | |
| "hitCondition" in breakpoint_data | |
| and breakpoint_data["hitCondition"] | |
| ): | |
| bp["hitCondition"] = breakpoint_data["hitCondition"] | |
| if ( | |
| "logMessage" in breakpoint_data | |
| and breakpoint_data["logMessage"] | |
| ): | |
| bp["logMessage"] = breakpoint_data["logMessage"] | |
| breakpoints.append(bp) | |
| args_dict["breakpoints"] = breakpoints | |
| command_dict = { | |
| "command": "setBreakpoints", | |
| "type": "request", | |
| "arguments": args_dict, | |
| } | |
| return self.send_recv(command_dict) | |
| def request_setExceptionBreakpoints(self, filters): | |
| args_dict = {"filters": filters} | |
| command_dict = { | |
| "command": "setExceptionBreakpoints", | |
| "type": "request", | |
| "arguments": args_dict, | |
| } | |
| return self.send_recv(command_dict) | |
| def request_setFunctionBreakpoints(self, names, condition=None, hitCondition=None): | |
| breakpoints = [] | |
| for name in names: | |
| bp = {"name": name} | |
| if condition is not None: | |
| bp["condition"] = condition | |
| if hitCondition is not None: | |
| bp["hitCondition"] = hitCondition | |
| breakpoints.append(bp) | |
| args_dict = {"breakpoints": breakpoints} | |
| command_dict = { | |
| "command": "setFunctionBreakpoints", | |
| "type": "request", | |
| "arguments": args_dict, | |
| } | |
| return self.send_recv(command_dict) | |
| def request_dataBreakpointInfo( | |
| self, variablesReference, name, frameIndex=0, threadId=None | |
| ): | |
| stackFrame = self.get_stackFrame(frameIndex=frameIndex, threadId=threadId) | |
| if stackFrame is None: | |
| return [] | |
| args_dict = { | |
| "variablesReference": variablesReference, | |
| "name": name, | |
| "frameId": stackFrame["id"], | |
| } | |
| command_dict = { | |
| "command": "dataBreakpointInfo", | |
| "type": "request", | |
| "arguments": args_dict, | |
| } | |
| return self.send_recv(command_dict) | |
| def request_setDataBreakpoint(self, dataBreakpoints): | |
| """dataBreakpoints is a list of dictionary with following fields: | |
| { | |
| dataId: (address in hex)/(size in bytes) | |
| accessType: read/write/readWrite | |
| [condition]: string | |
| [hitCondition]: string | |
| } | |
| """ | |
| args_dict = {"breakpoints": dataBreakpoints} | |
| command_dict = { | |
| "command": "setDataBreakpoints", | |
| "type": "request", | |
| "arguments": args_dict, | |
| } | |
| return self.send_recv(command_dict) | |
| def request_compileUnits(self, moduleId): | |
| args_dict = {"moduleId": moduleId} | |
| command_dict = { | |
| "command": "compileUnits", | |
| "type": "request", | |
| "arguments": args_dict, | |
| } | |
| response = self.send_recv(command_dict) | |
| return response | |
| def request_completions(self, text, frameId=None): | |
| args_dict = {"text": text, "column": len(text) + 1} | |
| if frameId: | |
| args_dict["frameId"] = frameId | |
| command_dict = { | |
| "command": "completions", | |
| "type": "request", | |
| "arguments": args_dict, | |
| } | |
| return self.send_recv(command_dict) | |
| def request_modules(self): | |
| return self.send_recv({"command": "modules", "type": "request"}) | |
| def request_stackTrace( | |
| self, threadId=None, startFrame=None, levels=None, dump=False | |
| ): | |
| if threadId is None: | |
| threadId = self.get_thread_id() | |
| args_dict = {"threadId": threadId} | |
| if startFrame is not None: | |
| args_dict["startFrame"] = startFrame | |
| if levels is not None: | |
| args_dict["levels"] = levels | |
| command_dict = { | |
| "command": "stackTrace", | |
| "type": "request", | |
| "arguments": args_dict, | |
| } | |
| response = self.send_recv(command_dict) | |
| if dump: | |
| for idx, frame in enumerate(response["body"]["stackFrames"]): | |
| name = frame["name"] | |
| if "line" in frame and "source" in frame: | |
| source = frame["source"] | |
| if "sourceReference" not in source: | |
| if "name" in source: | |
| source_name = source["name"] | |
| line = frame["line"] | |
| print("[%3u] %s @ %s:%u" % (idx, name, source_name, line)) | |
| continue | |
| print("[%3u] %s" % (idx, name)) | |
| return response | |
| def request_threads(self): | |
| """Request a list of all threads and combine any information from any | |
| "stopped" events since those contain more information about why a | |
| thread actually stopped. Returns an array of thread dictionaries | |
| with information about all threads""" | |
| command_dict = {"command": "threads", "type": "request", "arguments": {}} | |
| response = self.send_recv(command_dict) | |
| body = response["body"] | |
| # Fill in "self.threads" correctly so that clients that call | |
| # self.get_threads() or self.get_thread_id(...) can get information | |
| # on threads when the process is stopped. | |
| if "threads" in body: | |
| self.threads = body["threads"] | |
| for thread in self.threads: | |
| # Copy the thread dictionary so we can add key/value pairs to | |
| # it without affecting the original info from the "threads" | |
| # command. | |
| tid = thread["id"] | |
| if tid in self.thread_stop_reasons: | |
| thread_stop_info = self.thread_stop_reasons[tid] | |
| copy_keys = ["reason", "description", "text"] | |
| for key in copy_keys: | |
| if key in thread_stop_info: | |
| thread[key] = thread_stop_info[key] | |
| else: | |
| self.threads = None | |
| return response | |
| def request_variables( | |
| self, variablesReference, start=None, count=None, is_hex=None | |
| ): | |
| args_dict = {"variablesReference": variablesReference} | |
| if start is not None: | |
| args_dict["start"] = start | |
| if count is not None: | |
| args_dict["count"] = count | |
| if is_hex is not None: | |
| args_dict["format"] = {"hex": is_hex} | |
| command_dict = { | |
| "command": "variables", | |
| "type": "request", | |
| "arguments": args_dict, | |
| } | |
| return self.send_recv(command_dict) | |
| def request_setVariable(self, containingVarRef, name, value, id=None): | |
| args_dict = { | |
| "variablesReference": containingVarRef, | |
| "name": name, | |
| "value": str(value), | |
| } | |
| if id is not None: | |
| args_dict["id"] = id | |
| command_dict = { | |
| "command": "setVariable", | |
| "type": "request", | |
| "arguments": args_dict, | |
| } | |
| return self.send_recv(command_dict) | |
| def request_locations(self, locationReference): | |
| args_dict = { | |
| "locationReference": locationReference, | |
| } | |
| command_dict = { | |
| "command": "locations", | |
| "type": "request", | |
| "arguments": args_dict, | |
| } | |
| return self.send_recv(command_dict) | |
| def request_testGetTargetBreakpoints(self): | |
| """A request packet used in the LLDB test suite to get all currently | |
| set breakpoint infos for all breakpoints currently set in the | |
| target. | |
| """ | |
| command_dict = { | |
| "command": "_testGetTargetBreakpoints", | |
| "type": "request", | |
| "arguments": {}, | |
| } | |
| return self.send_recv(command_dict) | |
| def terminate(self): | |
| self.send.close() | |
| # self.recv.close() | |
| def request_setInstructionBreakpoints(self, memory_reference=[]): | |
| breakpoints = [] | |
| for i in memory_reference: | |
| args_dict = { | |
| "instructionReference": i, | |
| } | |
| breakpoints.append(args_dict) | |
| args_dict = {"breakpoints": breakpoints} | |
| command_dict = { | |
| "command": "setInstructionBreakpoints", | |
| "type": "request", | |
| "arguments": args_dict, | |
| } | |
| return self.send_recv(command_dict) | |
| class DebugAdaptorServer(DebugCommunication): | |
| def __init__( | |
| self, | |
| executable=None, | |
| port=None, | |
| init_commands=[], | |
| log_file=None, | |
| env=None, | |
| ): | |
| self.process = None | |
| if executable is not None: | |
| adaptor_env = os.environ.copy() | |
| if env is not None: | |
| adaptor_env.update(env) | |
| if log_file: | |
| adaptor_env["LLDBDAP_LOG"] = log_file | |
| self.process = subprocess.Popen( | |
| [executable], | |
| stdin=subprocess.PIPE, | |
| stdout=subprocess.PIPE, | |
| stderr=subprocess.PIPE, | |
| env=adaptor_env, | |
| ) | |
| DebugCommunication.__init__( | |
| self, self.process.stdout, self.process.stdin, init_commands, log_file | |
| ) | |
| elif port is not None: | |
| s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) | |
| s.connect(("127.0.0.1", port)) | |
| DebugCommunication.__init__( | |
| self, s.makefile("r"), s.makefile("w"), init_commands | |
| ) | |
| def get_pid(self): | |
| if self.process: | |
| return self.process.pid | |
| return -1 | |
| def terminate(self): | |
| super(DebugAdaptorServer, self).terminate() | |
| if self.process is not None: | |
| self.process.terminate() | |
| self.process.wait() | |
| self.process = None | |
| def attach_options_specified(options): | |
| if options.pid is not None: | |
| return True | |
| if options.waitFor: | |
| return True | |
| if options.attach: | |
| return True | |
| if options.attachCmds: | |
| return True | |
| return False | |
| def run_vscode(dbg, args, options): | |
| dbg.request_initialize(options.sourceInitFile) | |
| if attach_options_specified(options): | |
| response = dbg.request_attach( | |
| program=options.program, | |
| pid=options.pid, | |
| waitFor=options.waitFor, | |
| attachCommands=options.attachCmds, | |
| initCommands=options.initCmds, | |
| preRunCommands=options.preRunCmds, | |
| stopCommands=options.stopCmds, | |
| exitCommands=options.exitCmds, | |
| terminateCommands=options.terminateCmds, | |
| ) | |
| else: | |
| response = dbg.request_launch( | |
| options.program, | |
| args=args, | |
| env=options.envs, | |
| cwd=options.workingDir, | |
| debuggerRoot=options.debuggerRoot, | |
| sourcePath=options.sourcePath, | |
| initCommands=options.initCmds, | |
| preRunCommands=options.preRunCmds, | |
| stopCommands=options.stopCmds, | |
| exitCommands=options.exitCmds, | |
| terminateCommands=options.terminateCmds, | |
| ) | |
| if response["success"]: | |
| if options.sourceBreakpoints: | |
| source_to_lines = {} | |
| for file_line in options.sourceBreakpoints: | |
| (path, line) = file_line.split(":") | |
| if len(path) == 0 or len(line) == 0: | |
| print('error: invalid source with line "%s"' % (file_line)) | |
| else: | |
| if path in source_to_lines: | |
| source_to_lines[path].append(int(line)) | |
| else: | |
| source_to_lines[path] = [int(line)] | |
| for source in source_to_lines: | |
| dbg.request_setBreakpoints(source, source_to_lines[source]) | |
| if options.funcBreakpoints: | |
| dbg.request_setFunctionBreakpoints(options.funcBreakpoints) | |
| dbg.request_configurationDone() | |
| dbg.wait_for_stopped() | |
| else: | |
| if "message" in response: | |
| print(response["message"]) | |
| dbg.request_disconnect(terminateDebuggee=True) | |
| def main(): | |
| parser = optparse.OptionParser( | |
| description=( | |
| "A testing framework for the Visual Studio Code Debug Adaptor protocol" | |
| ) | |
| ) | |
| parser.add_option( | |
| "--vscode", | |
| type="string", | |
| dest="vscode_path", | |
| help=( | |
| "The path to the command line program that implements the " | |
| "Visual Studio Code Debug Adaptor protocol." | |
| ), | |
| default=None, | |
| ) | |
| parser.add_option( | |
| "--program", | |
| type="string", | |
| dest="program", | |
| help="The path to the program to debug.", | |
| default=None, | |
| ) | |
| parser.add_option( | |
| "--workingDir", | |
| type="string", | |
| dest="workingDir", | |
| default=None, | |
| help="Set the working directory for the process we launch.", | |
| ) | |
| parser.add_option( | |
| "--sourcePath", | |
| type="string", | |
| dest="sourcePath", | |
| default=None, | |
| help=( | |
| "Set the relative source root for any debug info that has " | |
| "relative paths in it." | |
| ), | |
| ) | |
| parser.add_option( | |
| "--debuggerRoot", | |
| type="string", | |
| dest="debuggerRoot", | |
| default=None, | |
| help=( | |
| "Set the working directory for lldb-dap for any object files " | |
| "with relative paths in the Mach-o debug map." | |
| ), | |
| ) | |
| parser.add_option( | |
| "-r", | |
| "--replay", | |
| type="string", | |
| dest="replay", | |
| help=( | |
| "Specify a file containing a packet log to replay with the " | |
| "current Visual Studio Code Debug Adaptor executable." | |
| ), | |
| default=None, | |
| ) | |
| parser.add_option( | |
| "-g", | |
| "--debug", | |
| action="store_true", | |
| dest="debug", | |
| default=False, | |
| help="Pause waiting for a debugger to attach to the debug adaptor", | |
| ) | |
| parser.add_option( | |
| "--sourceInitFile", | |
| action="store_true", | |
| dest="sourceInitFile", | |
| default=False, | |
| help="Whether lldb-dap should source .lldbinit file or not", | |
| ) | |
| parser.add_option( | |
| "--port", | |
| type="int", | |
| dest="port", | |
| help="Attach a socket to a port instead of using STDIN for VSCode", | |
| default=None, | |
| ) | |
| parser.add_option( | |
| "--pid", | |
| type="int", | |
| dest="pid", | |
| help="The process ID to attach to", | |
| default=None, | |
| ) | |
| parser.add_option( | |
| "--attach", | |
| action="store_true", | |
| dest="attach", | |
| default=False, | |
| help=( | |
| "Specify this option to attach to a process by name. The " | |
| "process name is the basename of the executable specified with " | |
| "the --program option." | |
| ), | |
| ) | |
| parser.add_option( | |
| "-f", | |
| "--function-bp", | |
| type="string", | |
| action="append", | |
| dest="funcBreakpoints", | |
| help=( | |
| "Specify the name of a function to break at. " | |
| "Can be specified more than once." | |
| ), | |
| default=[], | |
| ) | |
| parser.add_option( | |
| "-s", | |
| "--source-bp", | |
| type="string", | |
| action="append", | |
| dest="sourceBreakpoints", | |
| default=[], | |
| help=( | |
| "Specify source breakpoints to set in the format of " | |
| "<source>:<line>. " | |
| "Can be specified more than once." | |
| ), | |
| ) | |
| parser.add_option( | |
| "--attachCommand", | |
| type="string", | |
| action="append", | |
| dest="attachCmds", | |
| default=[], | |
| help=( | |
| "Specify a LLDB command that will attach to a process. " | |
| "Can be specified more than once." | |
| ), | |
| ) | |
| parser.add_option( | |
| "--initCommand", | |
| type="string", | |
| action="append", | |
| dest="initCmds", | |
| default=[], | |
| help=( | |
| "Specify a LLDB command that will be executed before the target " | |
| "is created. Can be specified more than once." | |
| ), | |
| ) | |
| parser.add_option( | |
| "--preRunCommand", | |
| type="string", | |
| action="append", | |
| dest="preRunCmds", | |
| default=[], | |
| help=( | |
| "Specify a LLDB command that will be executed after the target " | |
| "has been created. Can be specified more than once." | |
| ), | |
| ) | |
| parser.add_option( | |
| "--stopCommand", | |
| type="string", | |
| action="append", | |
| dest="stopCmds", | |
| default=[], | |
| help=( | |
| "Specify a LLDB command that will be executed each time the" | |
| "process stops. Can be specified more than once." | |
| ), | |
| ) | |
| parser.add_option( | |
| "--exitCommand", | |
| type="string", | |
| action="append", | |
| dest="exitCmds", | |
| default=[], | |
| help=( | |
| "Specify a LLDB command that will be executed when the process " | |
| "exits. Can be specified more than once." | |
| ), | |
| ) | |
| parser.add_option( | |
| "--terminateCommand", | |
| type="string", | |
| action="append", | |
| dest="terminateCmds", | |
| default=[], | |
| help=( | |
| "Specify a LLDB command that will be executed when the debugging " | |
| "session is terminated. Can be specified more than once." | |
| ), | |
| ) | |
| parser.add_option( | |
| "--env", | |
| type="string", | |
| action="append", | |
| dest="envs", | |
| default=[], | |
| help=("Specify environment variables to pass to the launched " "process."), | |
| ) | |
| parser.add_option( | |
| "--waitFor", | |
| action="store_true", | |
| dest="waitFor", | |
| default=False, | |
| help=( | |
| "Wait for the next process to be launched whose name matches " | |
| "the basename of the program specified with the --program " | |
| "option" | |
| ), | |
| ) | |
| (options, args) = parser.parse_args(sys.argv[1:]) | |
| if options.vscode_path is None and options.port is None: | |
| print( | |
| "error: must either specify a path to a Visual Studio Code " | |
| "Debug Adaptor vscode executable path using the --vscode " | |
| "option, or a port to attach to for an existing lldb-dap " | |
| "using the --port option" | |
| ) | |
| return | |
| dbg = DebugAdaptorServer(executable=options.vscode_path, port=options.port) | |
| if options.debug: | |
| raw_input('Waiting for debugger to attach pid "%i"' % (dbg.get_pid())) | |
| if options.replay: | |
| dbg.replay_packets(options.replay) | |
| else: | |
| run_vscode(dbg, args, options) | |
| dbg.terminate() | |
| if __name__ == "__main__": | |
| main() |