2006 lines · python
1#!/usr/bin/env python2 3import binascii4import json5import optparse6import os7import pprint8import socket9import string10import subprocess11import signal12import sys13import threading14import warnings15import time16from typing import (17 Any,18 Optional,19 Dict,20 cast,21 List,22 Callable,23 IO,24 Union,25 BinaryIO,26 TextIO,27 TypedDict,28 Literal,29)30 31# set timeout based on whether ASAN was enabled or not. Increase32# timeout by a factor of 10 if ASAN is enabled.33DEFAULT_TIMEOUT = 10 * (10 if ("ASAN_OPTIONS" in os.environ) else 1)34 35# See lldbtest.Base.spawnSubprocess, which should help ensure any processes36# created by the DAP client are terminated correctly when the test ends.37SpawnHelperCallback = Callable[[str, List[str], List[str]], subprocess.Popen]38 39## DAP type references40 41 42class Event(TypedDict):43 type: Literal["event"]44 seq: int45 event: str46 body: Any47 48 49class Request(TypedDict, total=False):50 type: Literal["request"]51 seq: int52 command: str53 arguments: Any54 55 56class Response(TypedDict):57 type: Literal["response"]58 seq: int59 request_seq: int60 success: bool61 command: str62 message: Optional[str]63 body: Any64 65 66ProtocolMessage = Union[Event, Request, Response]67 68 69class Source(TypedDict, total=False):70 name: str71 path: str72 sourceReference: int73 74 @staticmethod75 def build(76 *,77 name: Optional[str] = None,78 path: Optional[str] = None,79 source_reference: Optional[int] = None,80 ) -> "Source":81 """Builds a source from the given name, path or source_reference."""82 if not name and not path and not source_reference:83 raise ValueError(84 "Source.build requires either name, path, or source_reference"85 )86 87 s = Source()88 if name:89 s["name"] = name90 if path:91 if not name:92 s["name"] = os.path.basename(path)93 s["path"] = path94 if source_reference is not None:95 s["sourceReference"] = source_reference96 return s97 98 99class Breakpoint(TypedDict, total=False):100 id: int101 verified: bool102 source: Source103 104 @staticmethod105 def is_verified(src: "Breakpoint") -> bool:106 return src.get("verified", False)107 108 109def dump_memory(base_addr, data, num_per_line, outfile):110 data_len = len(data)111 hex_string = binascii.hexlify(data)112 addr = base_addr113 ascii_str = ""114 i = 0115 while i < data_len:116 outfile.write("0x%8.8x: " % (addr + i))117 bytes_left = data_len - i118 if bytes_left >= num_per_line:119 curr_data_len = num_per_line120 else:121 curr_data_len = bytes_left122 hex_start_idx = i * 2123 hex_end_idx = hex_start_idx + curr_data_len * 2124 curr_hex_str = hex_string[hex_start_idx:hex_end_idx]125 # 'curr_hex_str' now contains the hex byte string for the126 # current line with no spaces between bytes127 t = iter(curr_hex_str)128 # Print hex bytes separated by space129 outfile.write(" ".join(a + b for a, b in zip(t, t)))130 # Print two spaces131 outfile.write(" ")132 # Calculate ASCII string for bytes into 'ascii_str'133 ascii_str = ""134 for j in range(i, i + curr_data_len):135 ch = data[j]136 if ch in string.printable and ch not in string.whitespace:137 ascii_str += "%c" % (ch)138 else:139 ascii_str += "."140 # Print ASCII representation and newline141 outfile.write(ascii_str)142 i = i + curr_data_len143 outfile.write("\n")144 145 146def read_packet(147 f: IO[bytes], trace_file: Optional[IO[str]] = None148) -> Optional[ProtocolMessage]:149 """Decode a JSON packet that starts with the content length and is150 followed by the JSON bytes from a file 'f'. Returns None on EOF.151 """152 line = f.readline().decode("utf-8")153 if len(line) == 0:154 return None # EOF.155 156 # Watch for line that starts with the prefix157 prefix = "Content-Length: "158 if line.startswith(prefix):159 # Decode length of JSON bytes160 length = int(line[len(prefix) :])161 # Skip empty line162 separator = f.readline().decode()163 if separator != "":164 Exception("malformed DAP content header, unexpected line: " + separator)165 # Read JSON bytes166 json_str = f.read(length).decode()167 if trace_file:168 trace_file.write("from adapter:\n%s\n" % (json_str))169 # Decode the JSON bytes into a python dictionary170 return json.loads(json_str)171 172 raise Exception("unexpected malformed message from lldb-dap: " + line)173 174 175def packet_type_is(packet, packet_type):176 return "type" in packet and packet["type"] == packet_type177 178 179def dump_dap_log(log_file: Optional[str]) -> None:180 print("========= DEBUG ADAPTER PROTOCOL LOGS =========", file=sys.stderr)181 if log_file is None:182 print("no log file available", file=sys.stderr)183 else:184 with open(log_file, "r") as file:185 print(file.read(), file=sys.stderr)186 print("========= END =========", file=sys.stderr)187 188 189class NotSupportedError(KeyError):190 """Raised if a feature is not supported due to its capabilities."""191 192 193class DebugCommunication(object):194 @property195 def is_stopped(self) -> bool:196 """Returns True if the debuggee is stopped, otherwise False."""197 return len(self.thread_stop_reasons) > 0 or self.exit_status is not None198 199 def __init__(200 self,201 recv: BinaryIO,202 send: BinaryIO,203 init_commands: Optional[List[str]] = None,204 log_file: Optional[str] = None,205 spawn_helper: Optional[SpawnHelperCallback] = None,206 ):207 # For debugging test failures, try setting `trace_file = sys.stderr`.208 self.trace_file: Optional[TextIO] = None209 self.log_file = log_file210 self.send = send211 self.recv = recv212 self.spawn_helper = spawn_helper213 214 # Packets that have been received and processed but have not yet been215 # requested by a test case.216 self._pending_packets: List[Optional[ProtocolMessage]] = []217 # Received packets that have not yet been processed.218 self._recv_packets: List[Optional[ProtocolMessage]] = []219 # Used as a mutex for _recv_packets and for notify when _recv_packets220 # changes.221 self._recv_condition = threading.Condition()222 self._recv_thread = threading.Thread(target=self._read_packet_thread)223 224 # session state225 self.init_commands = init_commands if init_commands else []226 self.exit_status: Optional[int] = None227 self.capabilities: Dict = {}228 self.initialized: bool = False229 self.configuration_done_sent: bool = False230 self.process_event_body: Optional[Dict] = None231 self.terminated: bool = False232 self.events: List[Event] = []233 self.progress_events: List[Event] = []234 self.invalidated_event: Optional[Event] = None235 self.memory_event: Optional[Event] = None236 self.reverse_requests: List[Request] = []237 self.module_events: List[Dict] = []238 self.sequence: int = 1239 self.output: Dict[str, str] = {}240 241 # debuggee state242 self.threads: Optional[dict] = None243 self.thread_stop_reasons: Dict[str, Any] = {}244 self.frame_scopes: Dict[str, Any] = {}245 # keyed by breakpoint id246 self.resolved_breakpoints: dict[str, Breakpoint] = {}247 248 # trigger enqueue thread249 self._recv_thread.start()250 251 @classmethod252 def encode_content(cls, s: str) -> bytes:253 return ("Content-Length: %u\r\n\r\n%s" % (len(s), s)).encode("utf-8")254 255 @classmethod256 def validate_response(cls, command, response):257 if command["command"] != response["command"]:258 raise ValueError(259 f"command mismatch in response {command['command']} != {response['command']}"260 )261 if command["seq"] != response["request_seq"]:262 raise ValueError(263 f"seq mismatch in response {command['seq']} != {response['request_seq']}"264 )265 266 def _read_packet_thread(self):267 try:268 while True:269 packet = read_packet(self.recv, trace_file=self.trace_file)270 # `packet` will be `None` on EOF. We want to pass it down to271 # handle_recv_packet anyway so the main thread can handle unexpected272 # termination of lldb-dap and stop waiting for new packets.273 if not self._handle_recv_packet(packet):274 break275 finally:276 dump_dap_log(self.log_file)277 278 def get_modules(279 self, start_module: Optional[int] = None, module_count: Optional[int] = None280 ) -> Dict:281 resp = self.request_modules(start_module, module_count)282 if not resp["success"]:283 raise ValueError(f"request_modules failed: {resp!r}")284 modules = {}285 module_list = resp["body"]["modules"]286 for module in module_list:287 modules[module["name"]] = module288 return modules289 290 def get_output(self, category: str, clear=True) -> str:291 output = ""292 if category in self.output:293 output = self.output.get(category, "")294 if clear:295 del self.output[category]296 return output297 298 def collect_output(299 self,300 category: str,301 pattern: Optional[str] = None,302 clear=True,303 ) -> str:304 """Collect output from 'output' events.305 Args:306 category: The category to collect.307 pattern:308 Optional, if set, return once this pattern is detected in the309 collected output.310 Returns:311 The collected output.312 """313 deadline = time.monotonic() + DEFAULT_TIMEOUT314 output = self.get_output(category, clear)315 while deadline >= time.monotonic() and (316 pattern is None or pattern not in output317 ):318 event = self.wait_for_event(["output"])319 if not event: # Timeout or EOF320 break321 output += self.get_output(category, clear=clear)322 return output323 324 def _handle_recv_packet(self, packet: Optional[ProtocolMessage]) -> bool:325 """Handles an incoming packet.326 327 Called by the read thread that is waiting for all incoming packets328 to store the incoming packet in "self._recv_packets" in a thread safe329 way. This function will then signal the "self._recv_condition" to330 indicate a new packet is available.331 332 Args:333 packet: A new packet to store.334 335 Returns:336 True if the caller should keep calling this function for more337 packets.338 """339 with self._recv_condition:340 self._recv_packets.append(packet)341 self._recv_condition.notify()342 # packet is None on EOF343 return packet is not None and not (344 packet["type"] == "response" and packet["command"] == "disconnect"345 )346 347 def _recv_packet(348 self,349 *,350 predicate: Optional[Callable[[ProtocolMessage], bool]] = None,351 timeout: Optional[float] = DEFAULT_TIMEOUT,352 ) -> Optional[ProtocolMessage]:353 """Processes received packets from the adapter.354 Updates the DebugCommunication stateful properties based on the received355 packets in the order they are received.356 NOTE: The only time the session state properties should be updated is357 during this call to ensure consistency during tests.358 Args:359 predicate:360 Optional, if specified, returns the first packet that matches361 the given predicate.362 timeout:363 Optional, if specified, processes packets until either the364 timeout occurs or the predicate matches a packet, whichever365 occurs first.366 Returns:367 The first matching packet for the given predicate, if specified,368 otherwise None.369 """370 assert (371 threading.current_thread != self._recv_thread372 ), "Must not be called from the _recv_thread"373 374 def process_until_match():375 self._process_recv_packets()376 for i, packet in enumerate(self._pending_packets):377 if packet is None:378 # We need to return a truthy value to break out of the379 # wait_for, use `EOFError` as an indicator of EOF.380 return EOFError()381 if predicate and predicate(packet):382 self._pending_packets.pop(i)383 return packet384 385 with self._recv_condition:386 packet = self._recv_condition.wait_for(process_until_match, timeout)387 return None if isinstance(packet, EOFError) else packet388 389 def _process_recv_packets(self) -> None:390 """Process received packets, updating the session state."""391 with self._recv_condition:392 for packet in self._recv_packets:393 if packet and ("seq" not in packet or packet["seq"] == 0):394 warnings.warn(395 f"received a malformed packet, expected 'seq != 0' for {packet!r}"396 )397 # Handle events that may modify any stateful properties of398 # the DAP session.399 if packet and packet["type"] == "event":400 self._handle_event(packet)401 elif packet and packet["type"] == "request":402 # Handle reverse requests and keep processing.403 self._handle_reverse_request(packet)404 # Move the packet to the pending queue.405 self._pending_packets.append(packet)406 self._recv_packets.clear()407 408 def _handle_event(self, packet: Event) -> None:409 """Handle any events that modify debug session state we track."""410 event = packet["event"]411 body: Optional[Dict] = packet.get("body", None)412 413 if event == "output" and body:414 # Store any output we receive so clients can retrieve it later.415 category = body["category"]416 output = body["output"]417 if category in self.output:418 self.output[category] += output419 else:420 self.output[category] = output421 elif event == "initialized":422 self.initialized = True423 elif event == "process":424 # When a new process is attached or launched, remember the425 # details that are available in the body of the event426 self.process_event_body = body427 elif event == "exited" and body:428 # Process exited, mark the status to indicate the process is not429 # alive.430 self.exit_status = body["exitCode"]431 elif event == "continued" and body:432 # When the process continues, clear the known threads and433 # thread_stop_reasons.434 all_threads_continued = body.get("allThreadsContinued", True)435 tid = body["threadId"]436 if tid in self.thread_stop_reasons:437 del self.thread_stop_reasons[tid]438 self._process_continued(all_threads_continued)439 elif event == "stopped" and body:440 # Each thread that stops with a reason will send a441 # 'stopped' event. We need to remember the thread stop442 # reasons since the 'threads' command doesn't return443 # that information.444 self._process_stopped()445 tid = body["threadId"]446 self.thread_stop_reasons[tid] = body447 elif event.startswith("progress"):448 # Progress events come in as 'progressStart', 'progressUpdate',449 # and 'progressEnd' events. Keep these around in case test450 # cases want to verify them.451 self.progress_events.append(packet)452 elif event == "breakpoint" and body:453 # Breakpoint events are sent when a breakpoint is resolved454 self._update_verified_breakpoints([body["breakpoint"]])455 elif event == "capabilities" and body:456 # Update the capabilities with new ones from the event.457 self.capabilities.update(body["capabilities"])458 elif event == "invalidated":459 self.invalidated_event = packet460 elif event == "memory":461 self.memory_event = packet462 463 def _handle_reverse_request(self, request: Request) -> None:464 if request in self.reverse_requests:465 return466 self.reverse_requests.append(request)467 arguments = request.get("arguments")468 if request["command"] == "runInTerminal" and arguments is not None:469 assert self.spawn_helper is not None, "Not configured to spawn subprocesses"470 [exe, *args] = arguments["args"]471 env = [f"{k}={v}" for k, v in arguments.get("env", {}).items()]472 proc = self.spawn_helper(exe, args, env)473 body = {"processId": proc.pid}474 self.send_packet(475 {476 "type": "response",477 "seq": 0,478 "request_seq": request["seq"],479 "success": True,480 "command": "runInTerminal",481 "body": body,482 }483 )484 elif request["command"] == "startDebugging":485 self.send_packet(486 {487 "type": "response",488 "seq": 0,489 "request_seq": request["seq"],490 "success": True,491 "message": None,492 "command": "startDebugging",493 "body": {},494 }495 )496 else:497 desc = 'unknown reverse request "%s"' % (request["command"])498 raise ValueError(desc)499 500 def _process_continued(self, all_threads_continued: bool):501 self.frame_scopes = {}502 if all_threads_continued:503 self.thread_stop_reasons = {}504 505 def _update_verified_breakpoints(self, breakpoints: list[Breakpoint]):506 for bp in breakpoints:507 # If no id is set, we cannot correlate the given breakpoint across508 # requests, ignore it.509 if "id" not in bp:510 continue511 512 self.resolved_breakpoints[str(bp["id"])] = bp513 514 def send_packet(self, packet: ProtocolMessage) -> int:515 """Takes a dictionary representation of a DAP request and send the request to the debug adapter.516 517 Returns the seq number of the request.518 """519 # Set the seq for requests.520 if packet["type"] == "request":521 packet["seq"] = self.sequence522 self.sequence += 1523 else:524 packet["seq"] = 0525 526 # Encode our command dictionary as a JSON string527 json_str = json.dumps(packet, separators=(",", ":"))528 529 if self.trace_file:530 self.trace_file.write("to adapter:\n%s\n" % (json_str))531 532 length = len(json_str)533 if length > 0:534 # Send the encoded JSON packet and flush the 'send' file535 self.send.write(self.encode_content(json_str))536 self.send.flush()537 538 return packet["seq"]539 540 def _send_recv(self, request: Request) -> Optional[Response]:541 """Send a command python dictionary as JSON and receive the JSON542 response. Validates that the response is the correct sequence and543 command in the reply. Any events that are received are added to the544 events list in this object"""545 seq = self.send_packet(request)546 response = self.receive_response(seq)547 if response is None:548 raise ValueError(f"no response for {request!r}")549 self.validate_response(request, response)550 return response551 552 def receive_response(self, seq: int) -> Optional[Response]:553 """Waits for a response with the associated request_sec."""554 555 def predicate(p: ProtocolMessage):556 return p["type"] == "response" and p["request_seq"] == seq557 558 return cast(Optional[Response], self._recv_packet(predicate=predicate))559 560 def wait_for_event(self, filter: List[str] = []) -> Optional[Event]:561 """Wait for the first event that matches the filter."""562 563 def predicate(p: ProtocolMessage):564 return p["type"] == "event" and p["event"] in filter565 566 return cast(567 Optional[Event],568 self._recv_packet(predicate=predicate),569 )570 571 def wait_for_stopped(self) -> Optional[List[Event]]:572 stopped_events = []573 stopped_event = self.wait_for_event(filter=["stopped", "exited"])574 while stopped_event:575 stopped_events.append(stopped_event)576 # If we exited, then we are done577 if stopped_event["event"] == "exited":578 break579 580 # Otherwise we stopped and there might be one or more 'stopped'581 # events for each thread that stopped with a reason, so keep582 # checking for more 'stopped' events and return all of them583 # Use a shorter timeout for additional stopped events584 def predicate(p: ProtocolMessage):585 return p["type"] == "event" and p["event"] in ["stopped", "exited"]586 587 stopped_event = cast(588 Optional[Event], self._recv_packet(predicate=predicate, timeout=0.25)589 )590 return stopped_events591 592 def wait_for_breakpoint_events(self):593 breakpoint_events: list[Event] = []594 while True:595 event = self.wait_for_event(["breakpoint"])596 if not event:597 break598 breakpoint_events.append(event)599 return breakpoint_events600 601 def wait_for_breakpoints_to_be_verified(self, breakpoint_ids: list[str]):602 """Wait for all breakpoints to be verified. Return all unverified breakpoints."""603 while any(id not in self.resolved_breakpoints for id in breakpoint_ids):604 breakpoint_event = self.wait_for_event(["breakpoint"])605 if breakpoint_event is None:606 break607 608 return [609 id610 for id in breakpoint_ids611 if (612 id not in self.resolved_breakpoints613 or not Breakpoint.is_verified(self.resolved_breakpoints[id])614 )615 ]616 617 def wait_for_exited(self):618 event_dict = self.wait_for_event(["exited"])619 if event_dict is None:620 raise ValueError("didn't get exited event")621 return event_dict622 623 def wait_for_terminated(self):624 event_dict = self.wait_for_event(["terminated"])625 if event_dict is None:626 raise ValueError("didn't get terminated event")627 return event_dict628 629 def get_capability(self, key: str):630 """Get a value for the given key if it there is a key/value pair in631 the capabilities reported by the adapter.632 """633 if key in self.capabilities:634 return self.capabilities[key]635 raise NotSupportedError(key)636 637 def get_threads(self):638 if self.threads is None:639 self.request_threads()640 return self.threads641 642 def get_thread_id(self, threadIndex=0):643 """Utility function to get the first thread ID in the thread list.644 If the thread list is empty, then fetch the threads.645 """646 if self.threads is None:647 self.request_threads()648 if self.threads and threadIndex < len(self.threads):649 return self.threads[threadIndex]["id"]650 return None651 652 def get_stackFrame(self, frameIndex=0, threadId=None):653 """Get a single "StackFrame" object from a "stackTrace" request and654 return the "StackFrame" as a python dictionary, or None on failure655 """656 if threadId is None:657 threadId = self.get_thread_id()658 if threadId is None:659 print("invalid threadId")660 return None661 response = self.request_stackTrace(threadId, startFrame=frameIndex, levels=1)662 if response:663 return response["body"]["stackFrames"][0]664 print("invalid response")665 return None666 667 def get_completions(self, text, frameId=None):668 if frameId is None:669 stackFrame = self.get_stackFrame()670 frameId = stackFrame["id"]671 response = self.request_completions(text, frameId)672 return response["body"]["targets"]673 674 def get_scope_variables(self, scope_name, frameIndex=0, threadId=None, is_hex=None):675 stackFrame = self.get_stackFrame(frameIndex=frameIndex, threadId=threadId)676 if stackFrame is None:677 return []678 frameId = stackFrame["id"]679 if frameId in self.frame_scopes:680 frame_scopes = self.frame_scopes[frameId]681 else:682 scopes_response = self.request_scopes(frameId)683 frame_scopes = scopes_response["body"]["scopes"]684 self.frame_scopes[frameId] = frame_scopes685 for scope in frame_scopes:686 if scope["name"] == scope_name:687 varRef = scope["variablesReference"]688 variables_response = self.request_variables(varRef, is_hex=is_hex)689 if variables_response:690 if "body" in variables_response:691 body = variables_response["body"]692 if "variables" in body:693 vars = body["variables"]694 return vars695 return []696 697 def get_global_variables(self, frameIndex=0, threadId=None):698 return self.get_scope_variables(699 "Globals", frameIndex=frameIndex, threadId=threadId700 )701 702 def get_local_variables(self, frameIndex=0, threadId=None, is_hex=None):703 return self.get_scope_variables(704 "Locals", frameIndex=frameIndex, threadId=threadId, is_hex=is_hex705 )706 707 def get_registers(self, frameIndex=0, threadId=None):708 return self.get_scope_variables(709 "Registers", frameIndex=frameIndex, threadId=threadId710 )711 712 def get_local_variable(self, name, frameIndex=0, threadId=None, is_hex=None):713 locals = self.get_local_variables(714 frameIndex=frameIndex, threadId=threadId, is_hex=is_hex715 )716 for local in locals:717 if "name" in local and local["name"] == name:718 return local719 return None720 721 def get_local_variable_value(self, name, frameIndex=0, threadId=None, is_hex=None):722 variable = self.get_local_variable(723 name, frameIndex=frameIndex, threadId=threadId, is_hex=is_hex724 )725 if variable and "value" in variable:726 return variable["value"]727 return None728 729 def get_local_variable_child(730 self, name, child_name, frameIndex=0, threadId=None, is_hex=None731 ):732 local = self.get_local_variable(name, frameIndex, threadId)733 if local["variablesReference"] == 0:734 return None735 children = self.request_variables(local["variablesReference"], is_hex=is_hex)[736 "body"737 ]["variables"]738 for child in children:739 if child["name"] == child_name:740 return child741 return None742 743 def replay_packets(self, replay_file_path):744 f = open(replay_file_path, "r")745 mode = "invalid"746 set_sequence = False747 command_dict = None748 while mode != "eof":749 if mode == "invalid":750 line = f.readline()751 if line.startswith("to adapter:"):752 mode = "send"753 elif line.startswith("from adapter:"):754 mode = "recv"755 elif mode == "send":756 command_dict = read_packet(f)757 # Skip the end of line that follows the JSON758 f.readline()759 if command_dict is None:760 raise ValueError("decode packet failed from replay file")761 print("Sending:")762 pprint.PrettyPrinter(indent=2).pprint(command_dict)763 # raw_input('Press ENTER to send:')764 self.send_packet(command_dict, set_sequence)765 mode = "invalid"766 elif mode == "recv":767 print("Replay response:")768 replay_response = read_packet(f)769 # Skip the end of line that follows the JSON770 f.readline()771 pprint.PrettyPrinter(indent=2).pprint(replay_response)772 actual_response = self.recv_packet()773 if actual_response:774 type = actual_response["type"]775 print("Actual response:")776 if type == "response":777 self.validate_response(command_dict, actual_response)778 pprint.PrettyPrinter(indent=2).pprint(actual_response)779 else:780 print("error: didn't get a valid response")781 mode = "invalid"782 783 def request_attach(784 self,785 *,786 program: Optional[str] = None,787 pid: Optional[int] = None,788 debuggerId: Optional[int] = None,789 targetId: Optional[int] = None,790 waitFor=False,791 initCommands: Optional[list[str]] = None,792 preRunCommands: Optional[list[str]] = None,793 attachCommands: Optional[list[str]] = None,794 postRunCommands: Optional[list[str]] = None,795 stopCommands: Optional[list[str]] = None,796 exitCommands: Optional[list[str]] = None,797 terminateCommands: Optional[list[str]] = None,798 coreFile: Optional[str] = None,799 stopOnEntry=False,800 sourceMap: Optional[Union[list[tuple[str, str]], dict[str, str]]] = None,801 gdbRemotePort: Optional[int] = None,802 gdbRemoteHostname: Optional[str] = None,803 ):804 args_dict = {}805 if pid is not None:806 args_dict["pid"] = pid807 if program is not None:808 args_dict["program"] = program809 if debuggerId is not None:810 args_dict["debuggerId"] = debuggerId811 if targetId is not None:812 args_dict["targetId"] = targetId813 if waitFor:814 args_dict["waitFor"] = waitFor815 args_dict["initCommands"] = self.init_commands816 if initCommands:817 args_dict["initCommands"].extend(initCommands)818 if preRunCommands:819 args_dict["preRunCommands"] = preRunCommands820 if stopCommands:821 args_dict["stopCommands"] = stopCommands822 if exitCommands:823 args_dict["exitCommands"] = exitCommands824 if terminateCommands:825 args_dict["terminateCommands"] = terminateCommands826 if attachCommands:827 args_dict["attachCommands"] = attachCommands828 if coreFile:829 args_dict["coreFile"] = coreFile830 if stopOnEntry:831 args_dict["stopOnEntry"] = stopOnEntry832 if postRunCommands:833 args_dict["postRunCommands"] = postRunCommands834 if sourceMap:835 args_dict["sourceMap"] = sourceMap836 if gdbRemotePort is not None:837 args_dict["gdb-remote-port"] = gdbRemotePort838 if gdbRemoteHostname is not None:839 args_dict["gdb-remote-hostname"] = gdbRemoteHostname840 command_dict = {"command": "attach", "type": "request", "arguments": args_dict}841 return self._send_recv(command_dict)842 843 def request_breakpointLocations(844 self, file_path, line, end_line=None, column=None, end_column=None845 ):846 (dir, base) = os.path.split(file_path)847 source_dict = {"name": base, "path": file_path}848 args_dict = {}849 args_dict["source"] = source_dict850 if line is not None:851 args_dict["line"] = line852 if end_line is not None:853 args_dict["endLine"] = end_line854 if column is not None:855 args_dict["column"] = column856 if end_column is not None:857 args_dict["endColumn"] = end_column858 command_dict = {859 "command": "breakpointLocations",860 "type": "request",861 "arguments": args_dict,862 }863 return self._send_recv(command_dict)864 865 def request_configurationDone(self):866 command_dict = {867 "command": "configurationDone",868 "type": "request",869 "arguments": {},870 }871 response = self._send_recv(command_dict)872 if response:873 self.configuration_done_sent = True874 stopped_on_entry = self.is_stopped875 self.request_threads()876 if not stopped_on_entry:877 # Drop the initial cached threads if we did not stop-on-entry.878 # In VSCode, immediately following 'configurationDone', a879 # 'threads' request is made to get the initial set of threads,880 # specifically the main threads id and name.881 # We issue the threads request to mimic this pattern but in our882 # tests we don't want to cache the result unless the process is883 # actually stopped.884 self.threads = None885 return response886 887 def _process_stopped(self):888 self.threads = None889 self.frame_scopes = {}890 891 def request_continue(self, threadId=None, singleThread=False):892 if self.exit_status is not None:893 raise ValueError("request_continue called after process exited")894 # If we have launched or attached, then the first continue is done by895 # sending the 'configurationDone' request896 if not self.configuration_done_sent:897 return self.request_configurationDone()898 args_dict = {}899 if threadId is None:900 threadId = self.get_thread_id()901 if threadId:902 args_dict["threadId"] = threadId903 if singleThread:904 args_dict["singleThread"] = True905 command_dict = {906 "command": "continue",907 "type": "request",908 "arguments": args_dict,909 }910 response = self._send_recv(command_dict)911 if response["success"]:912 self._process_continued(response["body"]["allThreadsContinued"])913 # Caller must still call wait_for_stopped.914 return response915 916 def request_restart(self, restartArguments=None):917 if self.exit_status is not None:918 raise ValueError("request_restart called after process exited")919 self.get_capability("supportsRestartRequest")920 command_dict = {921 "command": "restart",922 "type": "request",923 }924 if restartArguments:925 command_dict["arguments"] = restartArguments926 927 response = self._send_recv(command_dict)928 # Caller must still call wait_for_stopped.929 return response930 931 def request_disconnect(self, terminateDebuggee=None):932 args_dict = {}933 if terminateDebuggee is not None:934 if terminateDebuggee:935 args_dict["terminateDebuggee"] = True936 else:937 args_dict["terminateDebuggee"] = False938 command_dict = {939 "command": "disconnect",940 "type": "request",941 "arguments": args_dict,942 }943 return self._send_recv(command_dict)944 945 def request_disassemble(946 self,947 memoryReference,948 instructionOffset=-50,949 instructionCount=200,950 resolveSymbols=True,951 ):952 args_dict = {953 "memoryReference": memoryReference,954 "instructionOffset": instructionOffset,955 "instructionCount": instructionCount,956 "resolveSymbols": resolveSymbols,957 }958 command_dict = {959 "command": "disassemble",960 "type": "request",961 "arguments": args_dict,962 }963 return self._send_recv(command_dict)["body"]["instructions"]964 965 def request_readMemory(self, memoryReference, offset, count):966 args_dict = {967 "memoryReference": memoryReference,968 "offset": offset,969 "count": count,970 }971 command_dict = {972 "command": "readMemory",973 "type": "request",974 "arguments": args_dict,975 }976 return self._send_recv(command_dict)977 978 def request_writeMemory(self, memoryReference, data, offset=0, allowPartial=False):979 args_dict = {980 "memoryReference": memoryReference,981 "data": data,982 }983 984 if offset:985 args_dict["offset"] = offset986 if allowPartial:987 args_dict["allowPartial"] = allowPartial988 989 command_dict = {990 "command": "writeMemory",991 "type": "request",992 "arguments": args_dict,993 }994 return self._send_recv(command_dict)995 996 def request_evaluate(997 self,998 expression,999 frameIndex=0,1000 threadId=None,1001 context=None,1002 is_hex: Optional[bool] = None,1003 ):1004 stackFrame = self.get_stackFrame(frameIndex=frameIndex, threadId=threadId)1005 if stackFrame is None:1006 return []1007 args_dict = {1008 "expression": expression,1009 "frameId": stackFrame["id"],1010 }1011 if context:1012 args_dict["context"] = context1013 if is_hex is not None:1014 args_dict["format"] = {"hex": is_hex}1015 command_dict = {1016 "command": "evaluate",1017 "type": "request",1018 "arguments": args_dict,1019 }1020 return self._send_recv(command_dict)1021 1022 def request_exceptionInfo(self, threadId=None):1023 if threadId is None:1024 threadId = self.get_thread_id()1025 args_dict = {"threadId": threadId}1026 command_dict = {1027 "command": "exceptionInfo",1028 "type": "request",1029 "arguments": args_dict,1030 }1031 return self._send_recv(command_dict)1032 1033 def request_initialize(self, sourceInitFile=False):1034 command_dict = {1035 "command": "initialize",1036 "type": "request",1037 "arguments": {1038 "adapterID": "lldb-native",1039 "clientID": "vscode",1040 "columnsStartAt1": True,1041 "linesStartAt1": True,1042 "locale": "en-us",1043 "pathFormat": "path",1044 "supportsRunInTerminalRequest": True,1045 "supportsVariablePaging": True,1046 "supportsVariableType": True,1047 "supportsStartDebuggingRequest": True,1048 "supportsProgressReporting": True,1049 "supportsInvalidatedEvent": True,1050 "supportsMemoryEvent": True,1051 "$__lldb_sourceInitFile": sourceInitFile,1052 },1053 }1054 response = self._send_recv(command_dict)1055 if response:1056 if "body" in response:1057 self.capabilities.update(response.get("body", {}))1058 return response1059 1060 def request_launch(1061 self,1062 program: str,1063 *,1064 args: Optional[list[str]] = None,1065 cwd: Optional[str] = None,1066 env: Optional[dict[str, str]] = None,1067 stopOnEntry=False,1068 disableASLR=False,1069 disableSTDIO=False,1070 shellExpandArguments=False,1071 console: Optional[str] = None,1072 stdio: Optional[list[str]] = None,1073 enableAutoVariableSummaries=False,1074 displayExtendedBacktrace=False,1075 enableSyntheticChildDebugging=False,1076 initCommands: Optional[list[str]] = None,1077 preRunCommands: Optional[list[str]] = None,1078 launchCommands: Optional[list[str]] = None,1079 postRunCommands: Optional[list[str]] = None,1080 stopCommands: Optional[list[str]] = None,1081 exitCommands: Optional[list[str]] = None,1082 terminateCommands: Optional[list[str]] = None,1083 sourceMap: Optional[Union[list[tuple[str, str]], dict[str, str]]] = None,1084 sourcePath: Optional[str] = None,1085 debuggerRoot: Optional[str] = None,1086 commandEscapePrefix: Optional[str] = None,1087 customFrameFormat: Optional[str] = None,1088 customThreadFormat: Optional[str] = None,1089 ):1090 args_dict = {"program": program}1091 if args:1092 args_dict["args"] = args1093 if cwd:1094 args_dict["cwd"] = cwd1095 if env:1096 args_dict["env"] = env1097 if stopOnEntry:1098 args_dict["stopOnEntry"] = stopOnEntry1099 if disableSTDIO:1100 args_dict["disableSTDIO"] = disableSTDIO1101 if shellExpandArguments:1102 args_dict["shellExpandArguments"] = shellExpandArguments1103 args_dict["initCommands"] = self.init_commands1104 if initCommands:1105 args_dict["initCommands"].extend(initCommands)1106 if preRunCommands:1107 args_dict["preRunCommands"] = preRunCommands1108 if stopCommands:1109 args_dict["stopCommands"] = stopCommands1110 if exitCommands:1111 args_dict["exitCommands"] = exitCommands1112 if terminateCommands:1113 args_dict["terminateCommands"] = terminateCommands1114 if sourcePath:1115 args_dict["sourcePath"] = sourcePath1116 if debuggerRoot:1117 args_dict["debuggerRoot"] = debuggerRoot1118 if launchCommands:1119 args_dict["launchCommands"] = launchCommands1120 if sourceMap:1121 args_dict["sourceMap"] = sourceMap1122 if console:1123 args_dict["console"] = console1124 if stdio:1125 args_dict["stdio"] = stdio1126 if postRunCommands:1127 args_dict["postRunCommands"] = postRunCommands1128 if customFrameFormat:1129 args_dict["customFrameFormat"] = customFrameFormat1130 if customThreadFormat:1131 args_dict["customThreadFormat"] = customThreadFormat1132 1133 args_dict["disableASLR"] = disableASLR1134 args_dict["enableAutoVariableSummaries"] = enableAutoVariableSummaries1135 args_dict["enableSyntheticChildDebugging"] = enableSyntheticChildDebugging1136 args_dict["displayExtendedBacktrace"] = displayExtendedBacktrace1137 if commandEscapePrefix is not None:1138 args_dict["commandEscapePrefix"] = commandEscapePrefix1139 command_dict = {"command": "launch", "type": "request", "arguments": args_dict}1140 return self._send_recv(command_dict)1141 1142 def request_next(self, threadId, granularity="statement"):1143 if self.exit_status is not None:1144 raise ValueError("request_continue called after process exited")1145 args_dict = {"threadId": threadId, "granularity": granularity}1146 command_dict = {"command": "next", "type": "request", "arguments": args_dict}1147 return self._send_recv(command_dict)1148 1149 def request_stepIn(self, threadId, targetId, granularity="statement"):1150 if self.exit_status is not None:1151 raise ValueError("request_stepIn called after process exited")1152 if threadId is None:1153 threadId = self.get_thread_id()1154 args_dict = {1155 "threadId": threadId,1156 "targetId": targetId,1157 "granularity": granularity,1158 }1159 command_dict = {"command": "stepIn", "type": "request", "arguments": args_dict}1160 return self._send_recv(command_dict)1161 1162 def request_stepInTargets(self, frameId):1163 if self.exit_status is not None:1164 raise ValueError("request_stepInTargets called after process exited")1165 self.get_capability("supportsStepInTargetsRequest")1166 args_dict = {"frameId": frameId}1167 command_dict = {1168 "command": "stepInTargets",1169 "type": "request",1170 "arguments": args_dict,1171 }1172 return self._send_recv(command_dict)1173 1174 def request_stepOut(self, threadId):1175 if self.exit_status is not None:1176 raise ValueError("request_stepOut called after process exited")1177 args_dict = {"threadId": threadId}1178 command_dict = {"command": "stepOut", "type": "request", "arguments": args_dict}1179 return self._send_recv(command_dict)1180 1181 def request_pause(self, threadId=None):1182 if self.exit_status is not None:1183 raise ValueError("request_pause called after process exited")1184 if threadId is None:1185 threadId = self.get_thread_id()1186 args_dict = {"threadId": threadId}1187 command_dict = {"command": "pause", "type": "request", "arguments": args_dict}1188 return self._send_recv(command_dict)1189 1190 def request_scopes(self, frameId):1191 args_dict = {"frameId": frameId}1192 command_dict = {"command": "scopes", "type": "request", "arguments": args_dict}1193 return self._send_recv(command_dict)1194 1195 def request_setBreakpoints(self, source: Source, line_array, data=None):1196 """data is array of parameters for breakpoints in line_array.1197 Each parameter object is 1:1 mapping with entries in line_entry.1198 It contains optional location/hitCondition/logMessage parameters.1199 """1200 args_dict = {1201 "source": source,1202 "sourceModified": False,1203 }1204 if line_array is not None:1205 args_dict["lines"] = line_array1206 breakpoints = []1207 for i, line in enumerate(line_array):1208 breakpoint_data = None1209 if data is not None and i < len(data):1210 breakpoint_data = data[i]1211 bp = {"line": line}1212 if breakpoint_data is not None:1213 if breakpoint_data.get("condition"):1214 bp["condition"] = breakpoint_data["condition"]1215 if breakpoint_data.get("hitCondition"):1216 bp["hitCondition"] = breakpoint_data["hitCondition"]1217 if breakpoint_data.get("logMessage"):1218 bp["logMessage"] = breakpoint_data["logMessage"]1219 if breakpoint_data.get("column"):1220 bp["column"] = breakpoint_data["column"]1221 breakpoints.append(bp)1222 args_dict["breakpoints"] = breakpoints1223 1224 command_dict = {1225 "command": "setBreakpoints",1226 "type": "request",1227 "arguments": args_dict,1228 }1229 response = self._send_recv(command_dict)1230 if response["success"]:1231 self._update_verified_breakpoints(response["body"]["breakpoints"])1232 return response1233 1234 def request_setExceptionBreakpoints(1235 self, *, filters: list[str] = [], filter_options: list[dict] = []1236 ):1237 args_dict = {"filters": filters}1238 if filter_options:1239 args_dict["filterOptions"] = filter_options1240 command_dict = {1241 "command": "setExceptionBreakpoints",1242 "type": "request",1243 "arguments": args_dict,1244 }1245 return self._send_recv(command_dict)1246 1247 def request_setFunctionBreakpoints(self, names, condition=None, hitCondition=None):1248 breakpoints = []1249 for name in names:1250 bp = {"name": name}1251 if condition is not None:1252 bp["condition"] = condition1253 if hitCondition is not None:1254 bp["hitCondition"] = hitCondition1255 breakpoints.append(bp)1256 args_dict = {"breakpoints": breakpoints}1257 command_dict = {1258 "command": "setFunctionBreakpoints",1259 "type": "request",1260 "arguments": args_dict,1261 }1262 response = self._send_recv(command_dict)1263 if response["success"]:1264 self._update_verified_breakpoints(response["body"]["breakpoints"])1265 return response1266 1267 def request_dataBreakpointInfo(1268 self, variablesReference, name, size=None, frameIndex=0, threadId=None1269 ):1270 stackFrame = self.get_stackFrame(frameIndex=frameIndex, threadId=threadId)1271 if stackFrame is None:1272 return []1273 args_dict = {"name": name}1274 if size is None:1275 args_dict["variablesReference"] = variablesReference1276 args_dict["frameId"] = stackFrame["id"]1277 else:1278 args_dict["asAddress"] = True1279 args_dict["bytes"] = size1280 command_dict = {1281 "command": "dataBreakpointInfo",1282 "type": "request",1283 "arguments": args_dict,1284 }1285 return self._send_recv(command_dict)1286 1287 def request_setDataBreakpoint(self, dataBreakpoints):1288 """dataBreakpoints is a list of dictionary with following fields:1289 {1290 dataId: (address in hex)/(size in bytes)1291 accessType: read/write/readWrite1292 [condition]: string1293 [hitCondition]: string1294 }1295 """1296 args_dict = {"breakpoints": dataBreakpoints}1297 command_dict = {1298 "command": "setDataBreakpoints",1299 "type": "request",1300 "arguments": args_dict,1301 }1302 return self._send_recv(command_dict)1303 1304 def request_compileUnits(self, moduleId):1305 args_dict = {"moduleId": moduleId}1306 command_dict = {1307 "command": "compileUnits",1308 "type": "request",1309 "arguments": args_dict,1310 }1311 response = self._send_recv(command_dict)1312 return response1313 1314 def request_completions(self, text, frameId=None):1315 args_dict = {"text": text, "column": len(text) + 1}1316 if frameId:1317 args_dict["frameId"] = frameId1318 command_dict = {1319 "command": "completions",1320 "type": "request",1321 "arguments": args_dict,1322 }1323 return self._send_recv(command_dict)1324 1325 def request_modules(1326 self,1327 start_module: Optional[int] = None,1328 module_count: Optional[int] = None,1329 ):1330 args_dict = {}1331 1332 if start_module is not None:1333 args_dict["startModule"] = start_module1334 if module_count is not None:1335 args_dict["moduleCount"] = module_count1336 1337 return self._send_recv(1338 {"command": "modules", "type": "request", "arguments": args_dict}1339 )1340 1341 def request_moduleSymbols(1342 self,1343 moduleId: str = "",1344 moduleName: str = "",1345 startIndex: int = 0,1346 count: int = 0,1347 ):1348 command_dict = {1349 "command": "__lldb_moduleSymbols",1350 "type": "request",1351 "arguments": {1352 "moduleId": moduleId,1353 "moduleName": moduleName,1354 "startIndex": startIndex,1355 "count": count,1356 },1357 }1358 return self._send_recv(command_dict)1359 1360 def request_stackTrace(1361 self, threadId=None, startFrame=None, levels=None, format=None, dump=False1362 ):1363 if threadId is None:1364 threadId = self.get_thread_id()1365 args_dict = {"threadId": threadId}1366 if startFrame is not None:1367 args_dict["startFrame"] = startFrame1368 if levels is not None:1369 args_dict["levels"] = levels1370 if format is not None:1371 args_dict["format"] = format1372 command_dict = {1373 "command": "stackTrace",1374 "type": "request",1375 "arguments": args_dict,1376 }1377 response = self._send_recv(command_dict)1378 if dump:1379 for idx, frame in enumerate(response["body"]["stackFrames"]):1380 name = frame["name"]1381 if "line" in frame and "source" in frame:1382 source = frame["source"]1383 if "sourceReference" not in source:1384 if "name" in source:1385 source_name = source["name"]1386 line = frame["line"]1387 print("[%3u] %s @ %s:%u" % (idx, name, source_name, line))1388 continue1389 print("[%3u] %s" % (idx, name))1390 return response1391 1392 def request_source(1393 self, *, source: Optional[Source] = None, sourceReference: Optional[int] = None1394 ):1395 """Request a source from a 'Source' reference."""1396 if source is None and sourceReference is None:1397 raise ValueError("request_source requires either source or sourceReference")1398 elif source is not None:1399 sourceReference = source["sourceReference"]1400 elif sourceReference is not None:1401 source = {"sourceReference": sourceReference}1402 else:1403 raise ValueError(1404 "request_source requires either source or sourceReference not both"1405 )1406 command_dict = {1407 "command": "source",1408 "type": "request",1409 "arguments": {1410 "source": source,1411 # legacy version of the request1412 "sourceReference": sourceReference,1413 },1414 }1415 return self._send_recv(command_dict)1416 1417 def request_threads(self):1418 """Request a list of all threads and combine any information from any1419 "stopped" events since those contain more information about why a1420 thread actually stopped. Returns an array of thread dictionaries1421 with information about all threads"""1422 command_dict = {"command": "threads", "type": "request", "arguments": {}}1423 response = self._send_recv(command_dict)1424 if not response["success"]:1425 self.threads = None1426 return response1427 body = response["body"]1428 # Fill in "self.threads" correctly so that clients that call1429 # self.get_threads() or self.get_thread_id(...) can get information1430 # on threads when the process is stopped.1431 if "threads" in body:1432 self.threads = body["threads"]1433 for thread in self.threads:1434 # Copy the thread dictionary so we can add key/value pairs to1435 # it without affecting the original info from the "threads"1436 # command.1437 tid = thread["id"]1438 if tid in self.thread_stop_reasons:1439 thread_stop_info = self.thread_stop_reasons[tid]1440 copy_keys = ["reason", "description", "text"]1441 for key in copy_keys:1442 if key in thread_stop_info:1443 thread[key] = thread_stop_info[key]1444 else:1445 self.threads = None1446 return response1447 1448 def request_variables(1449 self, variablesReference, start=None, count=None, is_hex=None1450 ):1451 args_dict = {"variablesReference": variablesReference}1452 if start is not None:1453 args_dict["start"] = start1454 if count is not None:1455 args_dict["count"] = count1456 if is_hex is not None:1457 args_dict["format"] = {"hex": is_hex}1458 command_dict = {1459 "command": "variables",1460 "type": "request",1461 "arguments": args_dict,1462 }1463 return self._send_recv(command_dict)1464 1465 def request_setVariable(self, containingVarRef, name, value, id=None):1466 args_dict = {1467 "variablesReference": containingVarRef,1468 "name": name,1469 "value": str(value),1470 }1471 if id is not None:1472 args_dict["id"] = id1473 command_dict = {1474 "command": "setVariable",1475 "type": "request",1476 "arguments": args_dict,1477 }1478 return self._send_recv(command_dict)1479 1480 def request_locations(self, locationReference):1481 args_dict = {1482 "locationReference": locationReference,1483 }1484 command_dict = {1485 "command": "locations",1486 "type": "request",1487 "arguments": args_dict,1488 }1489 return self._send_recv(command_dict)1490 1491 def request_testGetTargetBreakpoints(self):1492 """A request packet used in the LLDB test suite to get all currently1493 set breakpoint infos for all breakpoints currently set in the1494 target.1495 """1496 command_dict = {1497 "command": "_testGetTargetBreakpoints",1498 "type": "request",1499 "arguments": {},1500 }1501 return self._send_recv(command_dict)1502 1503 def terminate(self):1504 self.send.close()1505 if self._recv_thread.is_alive():1506 self._recv_thread.join()1507 1508 def request_setInstructionBreakpoints(self, memory_reference=[]):1509 breakpoints = []1510 for i in memory_reference:1511 args_dict = {1512 "instructionReference": i,1513 }1514 breakpoints.append(args_dict)1515 args_dict = {"breakpoints": breakpoints}1516 command_dict = {1517 "command": "setInstructionBreakpoints",1518 "type": "request",1519 "arguments": args_dict,1520 }1521 return self._send_recv(command_dict)1522 1523 1524class DebugAdapterServer(DebugCommunication):1525 def __init__(1526 self,1527 *,1528 executable: Optional[str] = None,1529 connection: Optional[str] = None,1530 init_commands: Optional[list[str]] = None,1531 log_file: Optional[str] = None,1532 env: Optional[Dict[str, str]] = None,1533 additional_args: Optional[List[str]] = None,1534 spawn_helper: Optional[SpawnHelperCallback] = None,1535 ):1536 self.process = None1537 self.connection = None1538 if executable is not None:1539 process, connection = DebugAdapterServer.launch(1540 executable=executable,1541 connection=connection,1542 env=env,1543 log_file=log_file,1544 additional_args=additional_args,1545 )1546 self.process = process1547 self.connection = connection1548 1549 if connection is not None:1550 scheme, address = connection.split("://")1551 if scheme == "unix-connect": # unix-connect:///path1552 s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)1553 s.connect(address)1554 elif scheme == "connection": # connection://[host]:port1555 host, port = address.rsplit(":", 1)1556 # create_connection with try both ipv4 and ipv6.1557 s = socket.create_connection((host.strip("[]"), int(port)))1558 else:1559 raise ValueError("invalid connection: {}".format(connection))1560 super().__init__(1561 s.makefile("rb"),1562 s.makefile("wb"),1563 init_commands,1564 log_file,1565 spawn_helper,1566 )1567 self.connection = connection1568 else:1569 super().__init__(1570 self.process.stdout,1571 self.process.stdin,1572 init_commands,1573 log_file,1574 spawn_helper,1575 )1576 1577 @classmethod1578 def launch(1579 cls,1580 *,1581 executable: str,1582 env: Optional[Dict[str, str]] = None,1583 log_file: Optional[str] = None,1584 connection: Optional[str] = None,1585 connection_timeout: Optional[int] = None,1586 additional_args: Optional[List[str]] = None,1587 ) -> tuple[subprocess.Popen, Optional[str]]:1588 adapter_env = os.environ.copy()1589 if env:1590 adapter_env.update(env)1591 1592 if log_file:1593 adapter_env["LLDBDAP_LOG"] = log_file1594 args = [executable]1595 1596 # Add additional arguments first (like --no-lldbinit)1597 if additional_args:1598 args.extend(additional_args)1599 1600 if connection is not None:1601 args.append("--connection")1602 args.append(connection)1603 1604 if connection_timeout is not None:1605 args.append("--connection-timeout")1606 args.append(str(connection_timeout))1607 1608 process = subprocess.Popen(1609 args,1610 stdin=subprocess.PIPE,1611 stdout=subprocess.PIPE,1612 stderr=sys.stderr,1613 env=adapter_env,1614 )1615 1616 if connection is None:1617 return (process, None)1618 1619 # lldb-dap will print the listening address once the listener is1620 # made to stdout. The listener is formatted like1621 # `connection://host:port` or `unix-connection:///path`.1622 expected_prefix = "Listening for: "1623 out = process.stdout.readline().decode()1624 if not out.startswith(expected_prefix):1625 process.kill()1626 raise ValueError(1627 "lldb-dap failed to print listening address, expected '{}', got '{}'".format(1628 expected_prefix, out1629 )1630 )1631 1632 # If the listener expanded into multiple addresses, use the first.1633 connection = out.removeprefix(expected_prefix).rstrip("\r\n").split(",", 1)[0]1634 1635 return (process, connection)1636 1637 def get_pid(self) -> int:1638 if self.process:1639 return self.process.pid1640 return -11641 1642 def terminate(self):1643 try:1644 if self.process is not None:1645 process = self.process1646 self.process = None1647 try:1648 # When we close stdin it should signal the lldb-dap that no1649 # new messages will arrive and it should shutdown on its1650 # own.1651 process.stdin.close()1652 process.wait(timeout=DEFAULT_TIMEOUT)1653 except subprocess.TimeoutExpired:1654 process.kill()1655 process.wait()1656 if process.returncode != 0:1657 raise DebugAdapterProcessError(process.returncode)1658 finally:1659 super(DebugAdapterServer, self).terminate()1660 1661 1662class DebugAdapterError(Exception):1663 pass1664 1665 1666class DebugAdapterProcessError(DebugAdapterError):1667 """Raised when the lldb-dap process exits with a non-zero exit status."""1668 1669 def __init__(self, returncode):1670 self.returncode = returncode1671 1672 def __str__(self):1673 if self.returncode and self.returncode < 0:1674 try:1675 return f"lldb-dap died with {signal.Signals(-self.returncode).name}."1676 except ValueError:1677 return f"lldb-dap died with unknown signal {-self.returncode}."1678 else:1679 return f"lldb-dap returned non-zero exit status {self.returncode}."1680 1681 1682def attach_options_specified(options):1683 if options.pid is not None:1684 return True1685 if options.waitFor:1686 return True1687 if options.attach:1688 return True1689 if options.attachCmds:1690 return True1691 return False1692 1693 1694def run_vscode(dbg, args, options):1695 dbg.request_initialize(options.sourceInitFile)1696 1697 if options.sourceBreakpoints:1698 source_to_lines = {}1699 for file_line in options.sourceBreakpoints:1700 (path, line) = file_line.split(":")1701 if len(path) == 0 or len(line) == 0:1702 print('error: invalid source with line "%s"' % (file_line))1703 1704 else:1705 if path in source_to_lines:1706 source_to_lines[path].append(int(line))1707 else:1708 source_to_lines[path] = [int(line)]1709 for source in source_to_lines:1710 dbg.request_setBreakpoints(Source(source), source_to_lines[source])1711 if options.funcBreakpoints:1712 dbg.request_setFunctionBreakpoints(options.funcBreakpoints)1713 1714 dbg.request_configurationDone()1715 1716 if attach_options_specified(options):1717 response = dbg.request_attach(1718 program=options.program,1719 pid=options.pid,1720 waitFor=options.waitFor,1721 attachCommands=options.attachCmds,1722 initCommands=options.initCmds,1723 preRunCommands=options.preRunCmds,1724 stopCommands=options.stopCmds,1725 exitCommands=options.exitCmds,1726 terminateCommands=options.terminateCmds,1727 )1728 else:1729 response = dbg.request_launch(1730 options.program,1731 args=args,1732 env=options.envs,1733 cwd=options.workingDir,1734 debuggerRoot=options.debuggerRoot,1735 sourcePath=options.sourcePath,1736 initCommands=options.initCmds,1737 preRunCommands=options.preRunCmds,1738 stopCommands=options.stopCmds,1739 exitCommands=options.exitCmds,1740 terminateCommands=options.terminateCmds,1741 )1742 1743 if response["success"]:1744 dbg.wait_for_stopped()1745 else:1746 if "message" in response:1747 print(response["message"])1748 dbg.request_disconnect(terminateDebuggee=True)1749 1750 1751def main():1752 parser = optparse.OptionParser(1753 description=(1754 "A testing framework for the Visual Studio Code Debug Adapter protocol"1755 )1756 )1757 1758 parser.add_option(1759 "--vscode",1760 type="string",1761 dest="vscode_path",1762 help=(1763 "The path to the command line program that implements the "1764 "Visual Studio Code Debug Adapter protocol."1765 ),1766 default=None,1767 )1768 1769 parser.add_option(1770 "--program",1771 type="string",1772 dest="program",1773 help="The path to the program to debug.",1774 default=None,1775 )1776 1777 parser.add_option(1778 "--workingDir",1779 type="string",1780 dest="workingDir",1781 default=None,1782 help="Set the working directory for the process we launch.",1783 )1784 1785 parser.add_option(1786 "--sourcePath",1787 type="string",1788 dest="sourcePath",1789 default=None,1790 help=(1791 "Set the relative source root for any debug info that has "1792 "relative paths in it."1793 ),1794 )1795 1796 parser.add_option(1797 "--debuggerRoot",1798 type="string",1799 dest="debuggerRoot",1800 default=None,1801 help=(1802 "Set the working directory for lldb-dap for any object files "1803 "with relative paths in the Mach-o debug map."1804 ),1805 )1806 1807 parser.add_option(1808 "-r",1809 "--replay",1810 type="string",1811 dest="replay",1812 help=(1813 "Specify a file containing a packet log to replay with the "1814 "current Visual Studio Code Debug Adapter executable."1815 ),1816 default=None,1817 )1818 1819 parser.add_option(1820 "-g",1821 "--debug",1822 action="store_true",1823 dest="debug",1824 default=False,1825 help="Pause waiting for a debugger to attach to the debug adapter",1826 )1827 1828 parser.add_option(1829 "--sourceInitFile",1830 action="store_true",1831 dest="sourceInitFile",1832 default=False,1833 help="Whether lldb-dap should source .lldbinit file or not",1834 )1835 1836 parser.add_option(1837 "--connection",1838 dest="connection",1839 help="Attach a socket connection of using STDIN for VSCode",1840 default=None,1841 )1842 1843 parser.add_option(1844 "--pid",1845 type="int",1846 dest="pid",1847 help="The process ID to attach to",1848 default=None,1849 )1850 1851 parser.add_option(1852 "--attach",1853 action="store_true",1854 dest="attach",1855 default=False,1856 help=(1857 "Specify this option to attach to a process by name. The "1858 "process name is the basename of the executable specified with "1859 "the --program option."1860 ),1861 )1862 1863 parser.add_option(1864 "-f",1865 "--function-bp",1866 type="string",1867 action="append",1868 dest="funcBreakpoints",1869 help=(1870 "Specify the name of a function to break at. "1871 "Can be specified more than once."1872 ),1873 default=[],1874 )1875 1876 parser.add_option(1877 "-s",1878 "--source-bp",1879 type="string",1880 action="append",1881 dest="sourceBreakpoints",1882 default=[],1883 help=(1884 "Specify source breakpoints to set in the format of "1885 "<source>:<line>. "1886 "Can be specified more than once."1887 ),1888 )1889 1890 parser.add_option(1891 "--attachCommand",1892 type="string",1893 action="append",1894 dest="attachCmds",1895 default=[],1896 help=(1897 "Specify a LLDB command that will attach to a process. "1898 "Can be specified more than once."1899 ),1900 )1901 1902 parser.add_option(1903 "--initCommand",1904 type="string",1905 action="append",1906 dest="initCmds",1907 default=[],1908 help=(1909 "Specify a LLDB command that will be executed before the target "1910 "is created. Can be specified more than once."1911 ),1912 )1913 1914 parser.add_option(1915 "--preRunCommand",1916 type="string",1917 action="append",1918 dest="preRunCmds",1919 default=[],1920 help=(1921 "Specify a LLDB command that will be executed after the target "1922 "has been created. Can be specified more than once."1923 ),1924 )1925 1926 parser.add_option(1927 "--stopCommand",1928 type="string",1929 action="append",1930 dest="stopCmds",1931 default=[],1932 help=(1933 "Specify a LLDB command that will be executed each time the"1934 "process stops. Can be specified more than once."1935 ),1936 )1937 1938 parser.add_option(1939 "--exitCommand",1940 type="string",1941 action="append",1942 dest="exitCmds",1943 default=[],1944 help=(1945 "Specify a LLDB command that will be executed when the process "1946 "exits. Can be specified more than once."1947 ),1948 )1949 1950 parser.add_option(1951 "--terminateCommand",1952 type="string",1953 action="append",1954 dest="terminateCmds",1955 default=[],1956 help=(1957 "Specify a LLDB command that will be executed when the debugging "1958 "session is terminated. Can be specified more than once."1959 ),1960 )1961 1962 parser.add_option(1963 "--env",1964 type="string",1965 action="append",1966 dest="envs",1967 default=[],1968 help=("Specify environment variables to pass to the launched " "process."),1969 )1970 1971 parser.add_option(1972 "--waitFor",1973 action="store_true",1974 dest="waitFor",1975 default=False,1976 help=(1977 "Wait for the next process to be launched whose name matches "1978 "the basename of the program specified with the --program "1979 "option"1980 ),1981 )1982 1983 (options, args) = parser.parse_args(sys.argv[1:])1984 1985 if options.vscode_path is None and options.connection is None:1986 print(1987 "error: must either specify a path to a Visual Studio Code "1988 "Debug Adapter vscode executable path using the --vscode "1989 "option, or using the --connection option"1990 )1991 return1992 dbg = DebugAdapterServer(1993 executable=options.vscode_path, connection=options.connection1994 )1995 if options.debug:1996 raw_input('Waiting for debugger to attach pid "%i"' % (dbg.get_pid()))1997 if options.replay:1998 dbg.replay_packets(options.replay)1999 else:2000 run_vscode(dbg, args, options)2001 dbg.terminate()2002 2003 2004if __name__ == "__main__":2005 main()2006