1""" 2Test case for testing the gdbremote protocol. 3 4Tests run against debugserver and lldb-server (llgs). 5lldb-server tests run where the lldb-server exe is 6available. 7 8This class will be broken into smaller test case classes by 9gdb remote packet functional areas. For now it contains 10the initial set of tests implemented. 11""" 12 13import unittest2 14import gdbremote_testcase 15import lldbgdbserverutils 16from lldbsuite.support import seven 17from lldbsuite.test.decorators import * 18from lldbsuite.test.lldbtest import * 19from lldbsuite.test.lldbdwarf import * 20from lldbsuite.test import lldbutil 21 22 23class LldbGdbServerTestCase(gdbremote_testcase.GdbRemoteTestCaseBase, DwarfOpcodeParser): 24 25 mydir = TestBase.compute_mydir(__file__) 26 27 def test_thread_suffix_supported(self): 28 server = self.connect_to_debug_monitor() 29 self.assertIsNotNone(server) 30 31 self.add_no_ack_remote_stream() 32 self.test_sequence.add_log_lines( 33 ["lldb-server < 26> read packet: $QThreadSuffixSupported#e4", 34 "lldb-server < 6> send packet: $OK#9a"], 35 True) 36 37 self.expect_gdbremote_sequence() 38 39 40 def test_list_threads_in_stop_reply_supported(self): 41 server = self.connect_to_debug_monitor() 42 self.assertIsNotNone(server) 43 44 self.add_no_ack_remote_stream() 45 self.test_sequence.add_log_lines( 46 ["lldb-server < 27> read packet: $QListThreadsInStopReply#21", 47 "lldb-server < 6> send packet: $OK#9a"], 48 True) 49 self.expect_gdbremote_sequence() 50 51 def test_c_packet_works(self): 52 self.build() 53 procs = self.prep_debug_monitor_and_inferior() 54 self.test_sequence.add_log_lines( 55 ["read packet: $c#63", 56 "send packet: $W00#00"], 57 True) 58 59 self.expect_gdbremote_sequence() 60 61 @skipIfWindows # No pty support to test any inferior output 62 def test_inferior_print_exit(self): 63 self.build() 64 procs = self.prep_debug_monitor_and_inferior( 65 inferior_args=["hello, world"]) 66 self.test_sequence.add_log_lines( 67 ["read packet: $vCont;c#a8", 68 {"type": "output_match", "regex": self.maybe_strict_output_regex(r"hello, world\r\n")}, 69 "send packet: $W00#00"], 70 True) 71 72 context = self.expect_gdbremote_sequence() 73 self.assertIsNotNone(context) 74 75 def test_first_launch_stop_reply_thread_matches_first_qC(self): 76 self.build() 77 procs = self.prep_debug_monitor_and_inferior() 78 self.test_sequence.add_log_lines(["read packet: $qC#00", 79 {"direction": "send", 80 "regex": r"^\$QC([0-9a-fA-F]+)#", 81 "capture": {1: "thread_id"}}, 82 "read packet: $?#00", 83 {"direction": "send", 84 "regex": r"^\$T[0-9a-fA-F]{2}thread:([0-9a-fA-F]+)", 85 "expect_captures": {1: "thread_id"}}], 86 True) 87 self.expect_gdbremote_sequence() 88 89 def test_attach_commandline_continue_app_exits(self): 90 self.build() 91 self.set_inferior_startup_attach() 92 procs = self.prep_debug_monitor_and_inferior() 93 self.test_sequence.add_log_lines( 94 ["read packet: $vCont;c#a8", 95 "send packet: $W00#00"], 96 True) 97 self.expect_gdbremote_sequence() 98 99 # Wait a moment for completed and now-detached inferior process to 100 # clear. 101 time.sleep(1) 102 103 if not lldb.remote_platform: 104 # Process should be dead now. Reap results. 105 poll_result = procs["inferior"].poll() 106 self.assertIsNotNone(poll_result) 107 108 # Where possible, verify at the system level that the process is not 109 # running. 110 self.assertFalse( 111 lldbgdbserverutils.process_is_running( 112 procs["inferior"].pid, False)) 113 114 def test_qRegisterInfo_returns_one_valid_result(self): 115 self.build() 116 self.prep_debug_monitor_and_inferior() 117 self.test_sequence.add_log_lines( 118 ["read packet: $qRegisterInfo0#00", 119 {"direction": "send", "regex": r"^\$(.+);#[0-9A-Fa-f]{2}", "capture": {1: "reginfo_0"}}], 120 True) 121 122 # Run the stream 123 context = self.expect_gdbremote_sequence() 124 self.assertIsNotNone(context) 125 126 reg_info_packet = context.get("reginfo_0") 127 self.assertIsNotNone(reg_info_packet) 128 self.assert_valid_reg_info( 129 lldbgdbserverutils.parse_reg_info_response(reg_info_packet)) 130 131 def test_qRegisterInfo_returns_all_valid_results(self): 132 self.build() 133 self.prep_debug_monitor_and_inferior() 134 self.add_register_info_collection_packets() 135 136 # Run the stream. 137 context = self.expect_gdbremote_sequence() 138 self.assertIsNotNone(context) 139 140 # Validate that each register info returned validates. 141 for reg_info in self.parse_register_info_packets(context): 142 self.assert_valid_reg_info(reg_info) 143 144 def test_qRegisterInfo_contains_required_generics_debugserver(self): 145 self.build() 146 self.prep_debug_monitor_and_inferior() 147 self.add_register_info_collection_packets() 148 149 # Run the packet stream. 150 context = self.expect_gdbremote_sequence() 151 self.assertIsNotNone(context) 152 153 # Gather register info entries. 154 reg_infos = self.parse_register_info_packets(context) 155 156 # Collect all generic registers found. 157 generic_regs = { 158 reg_info['generic']: 1 for reg_info in reg_infos if 'generic' in reg_info} 159 160 # Ensure we have a program counter register. 161 self.assertIn('pc', generic_regs) 162 163 # Ensure we have a frame pointer register. PPC64le's FP is the same as SP 164 if self.getArchitecture() != 'powerpc64le': 165 self.assertIn('fp', generic_regs) 166 167 # Ensure we have a stack pointer register. 168 self.assertIn('sp', generic_regs) 169 170 # Ensure we have a flags register. 171 self.assertIn('flags', generic_regs) 172 173 def test_qRegisterInfo_contains_at_least_one_register_set(self): 174 self.build() 175 self.prep_debug_monitor_and_inferior() 176 self.add_register_info_collection_packets() 177 178 # Run the packet stream. 179 context = self.expect_gdbremote_sequence() 180 self.assertIsNotNone(context) 181 182 # Gather register info entries. 183 reg_infos = self.parse_register_info_packets(context) 184 185 # Collect all register sets found. 186 register_sets = { 187 reg_info['set']: 1 for reg_info in reg_infos if 'set' in reg_info} 188 self.assertTrue(len(register_sets) >= 1) 189 190 def targetHasAVX(self): 191 triple = self.dbg.GetSelectedPlatform().GetTriple() 192 193 # TODO other platforms, please implement this function 194 if not re.match(".*-.*-linux", triple): 195 return True 196 197 # Need to do something different for non-Linux/Android targets 198 if lldb.remote_platform: 199 self.runCmd('platform get-file "/proc/cpuinfo" "cpuinfo"') 200 cpuinfo_path = "cpuinfo" 201 self.addTearDownHook(lambda: os.unlink("cpuinfo")) 202 else: 203 cpuinfo_path = "/proc/cpuinfo" 204 205 f = open(cpuinfo_path, 'r') 206 cpuinfo = f.read() 207 f.close() 208 return " avx " in cpuinfo 209 210 @expectedFailureAll(oslist=["windows"]) # no avx for now. 211 @add_test_categories(["llgs"]) 212 def test_qRegisterInfo_contains_avx_registers(self): 213 self.build() 214 self.prep_debug_monitor_and_inferior() 215 self.add_register_info_collection_packets() 216 217 # Run the packet stream. 218 context = self.expect_gdbremote_sequence() 219 self.assertIsNotNone(context) 220 221 # Gather register info entries. 222 reg_infos = self.parse_register_info_packets(context) 223 224 # Collect all generics found. 225 register_sets = { 226 reg_info['set']: 1 for reg_info in reg_infos if 'set' in reg_info} 227 self.assertEqual( 228 self.targetHasAVX(), 229 "Advanced Vector Extensions" in register_sets) 230 231 def qThreadInfo_contains_thread(self): 232 procs = self.prep_debug_monitor_and_inferior() 233 self.add_threadinfo_collection_packets() 234 235 # Run the packet stream. 236 context = self.expect_gdbremote_sequence() 237 self.assertIsNotNone(context) 238 239 # Gather threadinfo entries. 240 threads = self.parse_threadinfo_packets(context) 241 self.assertIsNotNone(threads) 242 243 # We should have exactly one thread. 244 self.assertEqual(len(threads), 1) 245 246 def test_qThreadInfo_contains_thread_launch(self): 247 self.build() 248 self.set_inferior_startup_launch() 249 self.qThreadInfo_contains_thread() 250 251 @expectedFailureAll(oslist=["windows"]) # expect one more thread stopped 252 def test_qThreadInfo_contains_thread_attach(self): 253 self.build() 254 self.set_inferior_startup_attach() 255 self.qThreadInfo_contains_thread() 256 257 def qThreadInfo_matches_qC(self): 258 procs = self.prep_debug_monitor_and_inferior() 259 260 self.add_threadinfo_collection_packets() 261 self.test_sequence.add_log_lines( 262 ["read packet: $qC#00", 263 {"direction": "send", "regex": r"^\$QC([0-9a-fA-F]+)#", "capture": {1: "thread_id"}} 264 ], True) 265 266 # Run the packet stream. 267 context = self.expect_gdbremote_sequence() 268 self.assertIsNotNone(context) 269 270 # Gather threadinfo entries. 271 threads = self.parse_threadinfo_packets(context) 272 self.assertIsNotNone(threads) 273 274 # We should have exactly one thread from threadinfo. 275 self.assertEqual(len(threads), 1) 276 277 # We should have a valid thread_id from $QC. 278 QC_thread_id_hex = context.get("thread_id") 279 self.assertIsNotNone(QC_thread_id_hex) 280 QC_thread_id = int(QC_thread_id_hex, 16) 281 282 # Those two should be the same. 283 self.assertEqual(threads[0], QC_thread_id) 284 285 def test_qThreadInfo_matches_qC_launch(self): 286 self.build() 287 self.set_inferior_startup_launch() 288 self.qThreadInfo_matches_qC() 289 290 @expectedFailureAll(oslist=["windows"]) # expect one more thread stopped 291 def test_qThreadInfo_matches_qC_attach(self): 292 self.build() 293 self.set_inferior_startup_attach() 294 self.qThreadInfo_matches_qC() 295 296 def test_p_returns_correct_data_size_for_each_qRegisterInfo_launch(self): 297 self.build() 298 self.set_inferior_startup_launch() 299 procs = self.prep_debug_monitor_and_inferior() 300 self.add_register_info_collection_packets() 301 302 # Run the packet stream. 303 context = self.expect_gdbremote_sequence() 304 self.assertIsNotNone(context) 305 306 # Gather register info entries. 307 reg_infos = self.parse_register_info_packets(context) 308 self.assertIsNotNone(reg_infos) 309 self.assertTrue(len(reg_infos) > 0) 310 311 byte_order = self.get_target_byte_order() 312 313 # Read value for each register. 314 reg_index = 0 315 for reg_info in reg_infos: 316 # Skip registers that don't have a register set. For x86, these are 317 # the DRx registers, which have no LLDB-kind register number and thus 318 # cannot be read via normal 319 # NativeRegisterContext::ReadRegister(reg_info,...) calls. 320 if not "set" in reg_info: 321 continue 322 323 # Clear existing packet expectations. 324 self.reset_test_sequence() 325 326 # Run the register query 327 self.test_sequence.add_log_lines( 328 ["read packet: $p{0:x}#00".format(reg_index), 329 {"direction": "send", "regex": r"^\$([0-9a-fA-F]+)#", "capture": {1: "p_response"}}], 330 True) 331 context = self.expect_gdbremote_sequence() 332 self.assertIsNotNone(context) 333 334 # Verify the response length. 335 p_response = context.get("p_response") 336 self.assertIsNotNone(p_response) 337 338 # Skip erraneous (unsupported) registers. 339 # TODO: remove this once we make unsupported registers disappear. 340 if p_response.startswith("E") and len(p_response) == 3: 341 continue 342 343 if "dynamic_size_dwarf_expr_bytes" in reg_info: 344 self.updateRegInfoBitsize(reg_info, byte_order) 345 self.assertEqual(len(p_response), 2 * int(reg_info["bitsize"]) / 8, 346 reg_info) 347 348 # Increment loop 349 reg_index += 1 350 351 def Hg_switches_to_3_threads(self): 352 # Startup the inferior with three threads (main + 2 new ones). 353 procs = self.prep_debug_monitor_and_inferior( 354 inferior_args=["thread:new", "thread:new"]) 355 356 # Let the inferior process have a few moments to start up the thread 357 # when launched. (The launch scenario has no time to run, so threads 358 # won't be there yet.) 359 self.run_process_then_stop(run_seconds=1) 360 361 # Wait at most x seconds for 3 threads to be present. 362 threads = self.wait_for_thread_count(3) 363 self.assertEqual(len(threads), 3) 364 365 # verify we can $H to each thead, and $qC matches the thread we set. 366 for thread in threads: 367 # Change to each thread, verify current thread id. 368 self.reset_test_sequence() 369 self.test_sequence.add_log_lines( 370 ["read packet: $Hg{0:x}#00".format(thread), # Set current thread. 371 "send packet: $OK#00", 372 "read packet: $qC#00", 373 {"direction": "send", "regex": r"^\$QC([0-9a-fA-F]+)#", "capture": {1: "thread_id"}}], 374 True) 375 376 context = self.expect_gdbremote_sequence() 377 self.assertIsNotNone(context) 378 379 # Verify the thread id. 380 self.assertIsNotNone(context.get("thread_id")) 381 self.assertEqual(int(context.get("thread_id"), 16), thread) 382 383 @expectedFailureAll(oslist=["windows"]) # expect 4 threads 384 def test_Hg_switches_to_3_threads_launch(self): 385 self.build() 386 self.set_inferior_startup_launch() 387 self.Hg_switches_to_3_threads() 388 389 @expectedFailureAll(oslist=["windows"]) # expecting one more thread 390 def test_Hg_switches_to_3_threads_attach(self): 391 self.build() 392 self.set_inferior_startup_attach() 393 self.Hg_switches_to_3_threads() 394 395 def Hc_then_Csignal_signals_correct_thread(self, segfault_signo): 396 # NOTE only run this one in inferior-launched mode: we can't grab inferior stdout when running attached, 397 # and the test requires getting stdout from the exe. 398 399 NUM_THREADS = 3 400 401 # Startup the inferior with three threads (main + NUM_THREADS-1 worker threads). 402 # inferior_args=["thread:print-ids"] 403 inferior_args = ["thread:segfault"] 404 for i in range(NUM_THREADS - 1): 405 # if i > 0: 406 # Give time between thread creation/segfaulting for the handler to work. 407 # inferior_args.append("sleep:1") 408 inferior_args.append("thread:new") 409 inferior_args.append("sleep:10") 410 411 # Launch/attach. (In our case, this should only ever be launched since 412 # we need inferior stdout/stderr). 413 procs = self.prep_debug_monitor_and_inferior( 414 inferior_args=inferior_args) 415 self.test_sequence.add_log_lines(["read packet: $c#63"], True) 416 context = self.expect_gdbremote_sequence() 417 418 # Let the inferior process have a few moments to start up the thread when launched. 419 # context = self.run_process_then_stop(run_seconds=1) 420 421 # Wait at most x seconds for all threads to be present. 422 # threads = self.wait_for_thread_count(NUM_THREADS) 423 # self.assertEquals(len(threads), NUM_THREADS) 424 425 signaled_tids = {} 426 print_thread_ids = {} 427 428 # Switch to each thread, deliver a signal, and verify signal delivery 429 for i in range(NUM_THREADS - 1): 430 # Run until SIGSEGV comes in. 431 self.reset_test_sequence() 432 self.test_sequence.add_log_lines([{"direction": "send", 433 "regex": r"^\$T([0-9a-fA-F]{2})thread:([0-9a-fA-F]+);", 434 "capture": {1: "signo", 435 2: "thread_id"}}], 436 True) 437 438 context = self.expect_gdbremote_sequence() 439 self.assertIsNotNone(context) 440 signo = context.get("signo") 441 self.assertEqual(int(signo, 16), segfault_signo) 442 443 # Ensure we haven't seen this tid yet. 444 thread_id = int(context.get("thread_id"), 16) 445 self.assertNotIn(thread_id, signaled_tids) 446 signaled_tids[thread_id] = 1 447 448 # Send SIGUSR1 to the thread that signaled the SIGSEGV. 449 self.reset_test_sequence() 450 self.test_sequence.add_log_lines( 451 [ 452 # Set the continue thread. 453 # Set current thread. 454 "read packet: $Hc{0:x}#00".format(thread_id), 455 "send packet: $OK#00", 456 457 # Continue sending the signal number to the continue thread. 458 # The commented out packet is a way to do this same operation without using 459 # a $Hc (but this test is testing $Hc, so we'll stick with the former). 460 "read packet: $C{0:x}#00".format(lldbutil.get_signal_number('SIGUSR1')), 461 # "read packet: $vCont;C{0:x}:{1:x};c#00".format(lldbutil.get_signal_number('SIGUSR1'), thread_id), 462 463 # FIXME: Linux does not report the thread stop on the delivered signal (SIGUSR1 here). MacOSX debugserver does. 464 # But MacOSX debugserver isn't guaranteeing the thread the signal handler runs on, so currently its an XFAIL. 465 # Need to rectify behavior here. The linux behavior is more intuitive to me since we're essentially swapping out 466 # an about-to-be-delivered signal (for which we already sent a stop packet) to a different signal. 467 # {"direction":"send", "regex":r"^\$T([0-9a-fA-F]{2})thread:([0-9a-fA-F]+);", "capture":{1:"stop_signo", 2:"stop_thread_id"} }, 468 # "read packet: $c#63", 469 {"type": "output_match", "regex": r"^received SIGUSR1 on thread id: ([0-9a-fA-F]+)\r\nthread ([0-9a-fA-F]+): past SIGSEGV\r\n", "capture": {1: "print_thread_id", 2: "post_handle_thread_id"}}, 470 ], 471 True) 472 473 # Run the sequence. 474 context = self.expect_gdbremote_sequence() 475 self.assertIsNotNone(context) 476 477 # Ensure the stop signal is the signal we delivered. 478 # stop_signo = context.get("stop_signo") 479 # self.assertIsNotNone(stop_signo) 480 # self.assertEquals(int(stop_signo,16), lldbutil.get_signal_number('SIGUSR1')) 481 482 # Ensure the stop thread is the thread to which we delivered the signal. 483 # stop_thread_id = context.get("stop_thread_id") 484 # self.assertIsNotNone(stop_thread_id) 485 # self.assertEquals(int(stop_thread_id,16), thread_id) 486 487 # Ensure we haven't seen this thread id yet. The inferior's 488 # self-obtained thread ids are not guaranteed to match the stub 489 # tids (at least on MacOSX). 490 print_thread_id = context.get("print_thread_id") 491 self.assertIsNotNone(print_thread_id) 492 print_thread_id = int(print_thread_id, 16) 493 self.assertNotIn(print_thread_id, print_thread_ids) 494 495 # Now remember this print (i.e. inferior-reflected) thread id and 496 # ensure we don't hit it again. 497 print_thread_ids[print_thread_id] = 1 498 499 # Ensure post signal-handle thread id matches the thread that 500 # initially raised the SIGSEGV. 501 post_handle_thread_id = context.get("post_handle_thread_id") 502 self.assertIsNotNone(post_handle_thread_id) 503 post_handle_thread_id = int(post_handle_thread_id, 16) 504 self.assertEqual(post_handle_thread_id, print_thread_id) 505 506 @expectedFailureDarwin 507 @skipIfWindows # no SIGSEGV support 508 @expectedFailureAll(oslist=["freebsd"], bugnumber="llvm.org/pr48419") 509 @expectedFailureNetBSD 510 def test_Hc_then_Csignal_signals_correct_thread_launch(self): 511 self.build() 512 self.set_inferior_startup_launch() 513 514 if self.platformIsDarwin(): 515 # Darwin debugserver translates some signals like SIGSEGV into some gdb 516 # expectations about fixed signal numbers. 517 self.Hc_then_Csignal_signals_correct_thread(self.TARGET_EXC_BAD_ACCESS) 518 else: 519 self.Hc_then_Csignal_signals_correct_thread( 520 lldbutil.get_signal_number('SIGSEGV')) 521 522 @skipIfWindows # No pty support to test any inferior output 523 def test_m_packet_reads_memory(self): 524 self.build() 525 self.set_inferior_startup_launch() 526 # This is the memory we will write into the inferior and then ensure we 527 # can read back with $m. 528 MEMORY_CONTENTS = "Test contents 0123456789 ABCDEFGHIJKLMNOPQRSTUVWXYZ abcdefghijklmnopqrstuvwxyz" 529 530 # Start up the inferior. 531 procs = self.prep_debug_monitor_and_inferior( 532 inferior_args=[ 533 "set-message:%s" % 534 MEMORY_CONTENTS, 535 "get-data-address-hex:g_message", 536 "sleep:5"]) 537 538 # Run the process 539 self.test_sequence.add_log_lines( 540 [ 541 # Start running after initial stop. 542 "read packet: $c#63", 543 # Match output line that prints the memory address of the message buffer within the inferior. 544 # Note we require launch-only testing so we can get inferior otuput. 545 {"type": "output_match", "regex": self.maybe_strict_output_regex(r"data address: 0x([0-9a-fA-F]+)\r\n"), 546 "capture": {1: "message_address"}}, 547 # Now stop the inferior. 548 "read packet: {}".format(chr(3)), 549 # And wait for the stop notification. 550 {"direction": "send", "regex": r"^\$T([0-9a-fA-F]{2})thread:([0-9a-fA-F]+);", "capture": {1: "stop_signo", 2: "stop_thread_id"}}], 551 True) 552 553 # Run the packet stream. 554 context = self.expect_gdbremote_sequence() 555 self.assertIsNotNone(context) 556 557 # Grab the message address. 558 self.assertIsNotNone(context.get("message_address")) 559 message_address = int(context.get("message_address"), 16) 560 561 # Grab contents from the inferior. 562 self.reset_test_sequence() 563 self.test_sequence.add_log_lines( 564 ["read packet: $m{0:x},{1:x}#00".format(message_address, len(MEMORY_CONTENTS)), 565 {"direction": "send", "regex": r"^\$(.+)#[0-9a-fA-F]{2}$", "capture": {1: "read_contents"}}], 566 True) 567 568 # Run the packet stream. 569 context = self.expect_gdbremote_sequence() 570 self.assertIsNotNone(context) 571 572 # Ensure what we read from inferior memory is what we wrote. 573 self.assertIsNotNone(context.get("read_contents")) 574 read_contents = seven.unhexlify(context.get("read_contents")) 575 self.assertEqual(read_contents, MEMORY_CONTENTS) 576 577 def test_qMemoryRegionInfo_is_supported(self): 578 self.build() 579 self.set_inferior_startup_launch() 580 # Start up the inferior. 581 procs = self.prep_debug_monitor_and_inferior() 582 583 # Ask if it supports $qMemoryRegionInfo. 584 self.test_sequence.add_log_lines( 585 ["read packet: $qMemoryRegionInfo#00", 586 "send packet: $OK#00" 587 ], True) 588 self.expect_gdbremote_sequence() 589 590 @skipIfWindows # No pty support to test any inferior output 591 def test_qMemoryRegionInfo_reports_code_address_as_executable(self): 592 self.build() 593 self.set_inferior_startup_launch() 594 595 # Start up the inferior. 596 procs = self.prep_debug_monitor_and_inferior( 597 inferior_args=["get-code-address-hex:hello", "sleep:5"]) 598 599 # Run the process 600 self.test_sequence.add_log_lines( 601 [ 602 # Start running after initial stop. 603 "read packet: $c#63", 604 # Match output line that prints the memory address of the message buffer within the inferior. 605 # Note we require launch-only testing so we can get inferior otuput. 606 {"type": "output_match", "regex": self.maybe_strict_output_regex(r"code address: 0x([0-9a-fA-F]+)\r\n"), 607 "capture": {1: "code_address"}}, 608 # Now stop the inferior. 609 "read packet: {}".format(chr(3)), 610 # And wait for the stop notification. 611 {"direction": "send", "regex": r"^\$T([0-9a-fA-F]{2})thread:([0-9a-fA-F]+);", "capture": {1: "stop_signo", 2: "stop_thread_id"}}], 612 True) 613 614 # Run the packet stream. 615 context = self.expect_gdbremote_sequence() 616 self.assertIsNotNone(context) 617 618 # Grab the code address. 619 self.assertIsNotNone(context.get("code_address")) 620 code_address = int(context.get("code_address"), 16) 621 622 # Grab memory region info from the inferior. 623 self.reset_test_sequence() 624 self.add_query_memory_region_packets(code_address) 625 626 # Run the packet stream. 627 context = self.expect_gdbremote_sequence() 628 self.assertIsNotNone(context) 629 mem_region_dict = self.parse_memory_region_packet(context) 630 631 # Ensure there are no errors reported. 632 self.assertNotIn("error", mem_region_dict) 633 634 # Ensure code address is readable and executable. 635 self.assertIn("permissions", mem_region_dict) 636 self.assertIn("r", mem_region_dict["permissions"]) 637 self.assertIn("x", mem_region_dict["permissions"]) 638 639 # Ensure the start address and size encompass the address we queried. 640 self.assert_address_within_memory_region(code_address, mem_region_dict) 641 642 @skipIfWindows # No pty support to test any inferior output 643 def test_qMemoryRegionInfo_reports_stack_address_as_rw(self): 644 self.build() 645 self.set_inferior_startup_launch() 646 647 # Start up the inferior. 648 procs = self.prep_debug_monitor_and_inferior( 649 inferior_args=["get-stack-address-hex:", "sleep:5"]) 650 651 # Run the process 652 self.test_sequence.add_log_lines( 653 [ 654 # Start running after initial stop. 655 "read packet: $c#63", 656 # Match output line that prints the memory address of the message buffer within the inferior. 657 # Note we require launch-only testing so we can get inferior otuput. 658 {"type": "output_match", "regex": self.maybe_strict_output_regex(r"stack address: 0x([0-9a-fA-F]+)\r\n"), 659 "capture": {1: "stack_address"}}, 660 # Now stop the inferior. 661 "read packet: {}".format(chr(3)), 662 # And wait for the stop notification. 663 {"direction": "send", "regex": r"^\$T([0-9a-fA-F]{2})thread:([0-9a-fA-F]+);", "capture": {1: "stop_signo", 2: "stop_thread_id"}}], 664 True) 665 666 # Run the packet stream. 667 context = self.expect_gdbremote_sequence() 668 self.assertIsNotNone(context) 669 670 # Grab the address. 671 self.assertIsNotNone(context.get("stack_address")) 672 stack_address = int(context.get("stack_address"), 16) 673 674 # Grab memory region info from the inferior. 675 self.reset_test_sequence() 676 self.add_query_memory_region_packets(stack_address) 677 678 # Run the packet stream. 679 context = self.expect_gdbremote_sequence() 680 self.assertIsNotNone(context) 681 mem_region_dict = self.parse_memory_region_packet(context) 682 683 # Ensure there are no errors reported. 684 self.assertNotIn("error", mem_region_dict) 685 686 # Ensure address is readable and executable. 687 self.assertIn("permissions", mem_region_dict) 688 self.assertIn("r", mem_region_dict["permissions"]) 689 self.assertIn("w", mem_region_dict["permissions"]) 690 691 # Ensure the start address and size encompass the address we queried. 692 self.assert_address_within_memory_region( 693 stack_address, mem_region_dict) 694 695 @skipIfWindows # No pty support to test any inferior output 696 def test_qMemoryRegionInfo_reports_heap_address_as_rw(self): 697 self.build() 698 self.set_inferior_startup_launch() 699 700 # Start up the inferior. 701 procs = self.prep_debug_monitor_and_inferior( 702 inferior_args=["get-heap-address-hex:", "sleep:5"]) 703 704 # Run the process 705 self.test_sequence.add_log_lines( 706 [ 707 # Start running after initial stop. 708 "read packet: $c#63", 709 # Match output line that prints the memory address of the message buffer within the inferior. 710 # Note we require launch-only testing so we can get inferior otuput. 711 {"type": "output_match", "regex": self.maybe_strict_output_regex(r"heap address: 0x([0-9a-fA-F]+)\r\n"), 712 "capture": {1: "heap_address"}}, 713 # Now stop the inferior. 714 "read packet: {}".format(chr(3)), 715 # And wait for the stop notification. 716 {"direction": "send", "regex": r"^\$T([0-9a-fA-F]{2})thread:([0-9a-fA-F]+);", "capture": {1: "stop_signo", 2: "stop_thread_id"}}], 717 True) 718 719 # Run the packet stream. 720 context = self.expect_gdbremote_sequence() 721 self.assertIsNotNone(context) 722 723 # Grab the address. 724 self.assertIsNotNone(context.get("heap_address")) 725 heap_address = int(context.get("heap_address"), 16) 726 727 # Grab memory region info from the inferior. 728 self.reset_test_sequence() 729 self.add_query_memory_region_packets(heap_address) 730 731 # Run the packet stream. 732 context = self.expect_gdbremote_sequence() 733 self.assertIsNotNone(context) 734 mem_region_dict = self.parse_memory_region_packet(context) 735 736 # Ensure there are no errors reported. 737 self.assertNotIn("error", mem_region_dict) 738 739 # Ensure address is readable and executable. 740 self.assertIn("permissions", mem_region_dict) 741 self.assertIn("r", mem_region_dict["permissions"]) 742 self.assertIn("w", mem_region_dict["permissions"]) 743 744 # Ensure the start address and size encompass the address we queried. 745 self.assert_address_within_memory_region(heap_address, mem_region_dict) 746 747 def breakpoint_set_and_remove_work(self, want_hardware): 748 # Start up the inferior. 749 procs = self.prep_debug_monitor_and_inferior( 750 inferior_args=[ 751 "get-code-address-hex:hello", 752 "sleep:1", 753 "call-function:hello"]) 754 755 # Run the process 756 self.add_register_info_collection_packets() 757 self.add_process_info_collection_packets() 758 self.test_sequence.add_log_lines( 759 [ # Start running after initial stop. 760 "read packet: $c#63", 761 # Match output line that prints the memory address of the function call entry point. 762 # Note we require launch-only testing so we can get inferior otuput. 763 {"type": "output_match", "regex": self.maybe_strict_output_regex(r"code address: 0x([0-9a-fA-F]+)\r\n"), 764 "capture": {1: "function_address"}}, 765 # Now stop the inferior. 766 "read packet: {}".format(chr(3)), 767 # And wait for the stop notification. 768 {"direction": "send", "regex": r"^\$T([0-9a-fA-F]{2})thread:([0-9a-fA-F]+);", "capture": {1: "stop_signo", 2: "stop_thread_id"}}], 769 True) 770 771 # Run the packet stream. 772 context = self.expect_gdbremote_sequence() 773 self.assertIsNotNone(context) 774 775 # Gather process info - we need endian of target to handle register 776 # value conversions. 777 process_info = self.parse_process_info_response(context) 778 endian = process_info.get("endian") 779 self.assertIsNotNone(endian) 780 781 # Gather register info entries. 782 reg_infos = self.parse_register_info_packets(context) 783 (pc_lldb_reg_index, pc_reg_info) = self.find_pc_reg_info(reg_infos) 784 self.assertIsNotNone(pc_lldb_reg_index) 785 self.assertIsNotNone(pc_reg_info) 786 787 # Grab the function address. 788 self.assertIsNotNone(context.get("function_address")) 789 function_address = int(context.get("function_address"), 16) 790 791 # Get current target architecture 792 target_arch = self.getArchitecture() 793 794 # Set the breakpoint. 795 if (target_arch == "arm") or (target_arch == "aarch64"): 796 # TODO: Handle case when setting breakpoint in thumb code 797 BREAKPOINT_KIND = 4 798 else: 799 BREAKPOINT_KIND = 1 800 801 # Set default packet type to Z0 (software breakpoint) 802 z_packet_type = 0 803 804 # If hardware breakpoint is requested set packet type to Z1 805 if want_hardware == True: 806 z_packet_type = 1 807 808 self.reset_test_sequence() 809 self.add_set_breakpoint_packets( 810 function_address, 811 z_packet_type, 812 do_continue=True, 813 breakpoint_kind=BREAKPOINT_KIND) 814 815 # Run the packet stream. 816 context = self.expect_gdbremote_sequence() 817 self.assertIsNotNone(context) 818 819 # Verify the stop signal reported was the breakpoint signal number. 820 stop_signo = context.get("stop_signo") 821 self.assertIsNotNone(stop_signo) 822 self.assertEqual(int(stop_signo, 16), 823 lldbutil.get_signal_number('SIGTRAP')) 824 825 # Ensure we did not receive any output. If the breakpoint was not set, we would 826 # see output (from a launched process with captured stdio) printing a hello, world message. 827 # That would indicate the breakpoint didn't take. 828 self.assertEqual(len(context["O_content"]), 0) 829 830 # Verify that the PC for the main thread is where we expect it - right at the breakpoint address. 831 # This acts as a another validation on the register reading code. 832 self.reset_test_sequence() 833 self.test_sequence.add_log_lines( 834 [ 835 # Print the PC. This should match the breakpoint address. 836 "read packet: $p{0:x}#00".format(pc_lldb_reg_index), 837 # Capture $p results. 838 {"direction": "send", 839 "regex": r"^\$([0-9a-fA-F]+)#", 840 "capture": {1: "p_response"}}, 841 ], True) 842 843 context = self.expect_gdbremote_sequence() 844 self.assertIsNotNone(context) 845 846 # Verify the PC is where we expect. Note response is in endianness of 847 # the inferior. 848 p_response = context.get("p_response") 849 self.assertIsNotNone(p_response) 850 851 # Convert from target endian to int. 852 returned_pc = lldbgdbserverutils.unpack_register_hex_unsigned( 853 endian, p_response) 854 self.assertEqual(returned_pc, function_address) 855 856 # Verify that a breakpoint remove and continue gets us the expected 857 # output. 858 self.reset_test_sequence() 859 860 # Add breakpoint remove packets 861 self.add_remove_breakpoint_packets( 862 function_address, 863 z_packet_type, 864 breakpoint_kind=BREAKPOINT_KIND) 865 866 self.test_sequence.add_log_lines( 867 [ 868 # Continue running. 869 "read packet: $c#63", 870 # We should now receive the output from the call. 871 {"type": "output_match", "regex": r"^hello, world\r\n$"}, 872 # And wait for program completion. 873 {"direction": "send", "regex": r"^\$W00(.*)#[0-9a-fA-F]{2}$"}, 874 ], True) 875 876 context = self.expect_gdbremote_sequence() 877 self.assertIsNotNone(context) 878 879 @skipIfWindows # No pty support to test any inferior output 880 def test_software_breakpoint_set_and_remove_work(self): 881 if self.getArchitecture() == "arm": 882 # TODO: Handle case when setting breakpoint in thumb code 883 self.build(dictionary={'CFLAGS_EXTRAS': '-marm'}) 884 else: 885 self.build() 886 self.set_inferior_startup_launch() 887 self.breakpoint_set_and_remove_work(want_hardware=False) 888 889 @skipUnlessPlatform(oslist=['linux']) 890 @skipIf(archs=no_match(['arm', 'aarch64'])) 891 def test_hardware_breakpoint_set_and_remove_work(self): 892 if self.getArchitecture() == "arm": 893 # TODO: Handle case when setting breakpoint in thumb code 894 self.build(dictionary={'CFLAGS_EXTRAS': '-marm'}) 895 else: 896 self.build() 897 self.set_inferior_startup_launch() 898 self.breakpoint_set_and_remove_work(want_hardware=True) 899 900 def test_qSupported_returns_known_stub_features(self): 901 self.build() 902 self.set_inferior_startup_launch() 903 904 # Start up the stub and start/prep the inferior. 905 procs = self.prep_debug_monitor_and_inferior() 906 self.add_qSupported_packets() 907 908 # Run the packet stream. 909 context = self.expect_gdbremote_sequence() 910 self.assertIsNotNone(context) 911 912 # Retrieve the qSupported features. 913 supported_dict = self.parse_qSupported_response(context) 914 self.assertIsNotNone(supported_dict) 915 self.assertTrue(len(supported_dict) > 0) 916 917 @skipIfWindows # No pty support to test any inferior output 918 def test_written_M_content_reads_back_correctly(self): 919 self.build() 920 self.set_inferior_startup_launch() 921 922 TEST_MESSAGE = "Hello, memory" 923 924 # Start up the stub and start/prep the inferior. 925 procs = self.prep_debug_monitor_and_inferior( 926 inferior_args=[ 927 "set-message:xxxxxxxxxxxxxX", 928 "get-data-address-hex:g_message", 929 "sleep:1", 930 "print-message:"]) 931 self.test_sequence.add_log_lines( 932 [ 933 # Start running after initial stop. 934 "read packet: $c#63", 935 # Match output line that prints the memory address of the message buffer within the inferior. 936 # Note we require launch-only testing so we can get inferior otuput. 937 {"type": "output_match", "regex": self.maybe_strict_output_regex(r"data address: 0x([0-9a-fA-F]+)\r\n"), 938 "capture": {1: "message_address"}}, 939 # Now stop the inferior. 940 "read packet: {}".format(chr(3)), 941 # And wait for the stop notification. 942 {"direction": "send", "regex": r"^\$T([0-9a-fA-F]{2})thread:([0-9a-fA-F]+);", "capture": {1: "stop_signo", 2: "stop_thread_id"}}], 943 True) 944 context = self.expect_gdbremote_sequence() 945 self.assertIsNotNone(context) 946 947 # Grab the message address. 948 self.assertIsNotNone(context.get("message_address")) 949 message_address = int(context.get("message_address"), 16) 950 951 # Hex-encode the test message, adding null termination. 952 hex_encoded_message = seven.hexlify(TEST_MESSAGE) 953 954 # Write the message to the inferior. Verify that we can read it with the hex-encoded (m) 955 # and binary (x) memory read packets. 956 self.reset_test_sequence() 957 self.test_sequence.add_log_lines( 958 ["read packet: $M{0:x},{1:x}:{2}#00".format(message_address, len(TEST_MESSAGE), hex_encoded_message), 959 "send packet: $OK#00", 960 "read packet: $m{0:x},{1:x}#00".format(message_address, len(TEST_MESSAGE)), 961 "send packet: ${0}#00".format(hex_encoded_message), 962 "read packet: $x{0:x},{1:x}#00".format(message_address, len(TEST_MESSAGE)), 963 "send packet: ${0}#00".format(TEST_MESSAGE), 964 "read packet: $m{0:x},4#00".format(message_address), 965 "send packet: ${0}#00".format(hex_encoded_message[0:8]), 966 "read packet: $x{0:x},4#00".format(message_address), 967 "send packet: ${0}#00".format(TEST_MESSAGE[0:4]), 968 "read packet: $c#63", 969 {"type": "output_match", "regex": r"^message: (.+)\r\n$", "capture": {1: "printed_message"}}, 970 "send packet: $W00#00", 971 ], True) 972 context = self.expect_gdbremote_sequence() 973 self.assertIsNotNone(context) 974 975 # Ensure what we read from inferior memory is what we wrote. 976 printed_message = context.get("printed_message") 977 self.assertIsNotNone(printed_message) 978 self.assertEqual(printed_message, TEST_MESSAGE + "X") 979 980 # Note: as of this moment, a hefty number of the GPR writes are failing with E32 (everything except rax-rdx, rdi, rsi, rbp). 981 # Come back to this. I have the test rigged to verify that at least some 982 # of the bit-flip writes work. 983 def test_P_writes_all_gpr_registers(self): 984 self.build() 985 self.set_inferior_startup_launch() 986 987 # Start inferior debug session, grab all register info. 988 procs = self.prep_debug_monitor_and_inferior(inferior_args=["sleep:2"]) 989 self.add_register_info_collection_packets() 990 self.add_process_info_collection_packets() 991 992 context = self.expect_gdbremote_sequence() 993 self.assertIsNotNone(context) 994 995 # Process register infos. 996 reg_infos = self.parse_register_info_packets(context) 997 self.assertIsNotNone(reg_infos) 998 self.add_lldb_register_index(reg_infos) 999 1000 # Process endian. 1001 process_info = self.parse_process_info_response(context) 1002 endian = process_info.get("endian") 1003 self.assertIsNotNone(endian) 1004 1005 # Pull out the register infos that we think we can bit flip 1006 # successfully,. 1007 gpr_reg_infos = [ 1008 reg_info for reg_info in reg_infos if self.is_bit_flippable_register(reg_info)] 1009 self.assertTrue(len(gpr_reg_infos) > 0) 1010 1011 # Write flipped bit pattern of existing value to each register. 1012 (successful_writes, failed_writes) = self.flip_all_bits_in_each_register_value( 1013 gpr_reg_infos, endian) 1014 self.trace("successful writes: {}, failed writes: {}".format(successful_writes, failed_writes)) 1015 self.assertTrue(successful_writes > 0) 1016 1017 # Note: as of this moment, a hefty number of the GPR writes are failing 1018 # with E32 (everything except rax-rdx, rdi, rsi, rbp). 1019 @skipIfWindows 1020 def test_P_and_p_thread_suffix_work(self): 1021 self.build() 1022 self.set_inferior_startup_launch() 1023 1024 # Startup the inferior with three threads. 1025 procs = self.prep_debug_monitor_and_inferior( 1026 inferior_args=["thread:new", "thread:new"]) 1027 self.add_thread_suffix_request_packets() 1028 self.add_register_info_collection_packets() 1029 self.add_process_info_collection_packets() 1030 1031 context = self.expect_gdbremote_sequence() 1032 self.assertIsNotNone(context) 1033 1034 process_info = self.parse_process_info_response(context) 1035 self.assertIsNotNone(process_info) 1036 endian = process_info.get("endian") 1037 self.assertIsNotNone(endian) 1038 1039 reg_infos = self.parse_register_info_packets(context) 1040 self.assertIsNotNone(reg_infos) 1041 self.add_lldb_register_index(reg_infos) 1042 1043 reg_index = self.select_modifiable_register(reg_infos) 1044 self.assertIsNotNone(reg_index) 1045 reg_byte_size = int(reg_infos[reg_index]["bitsize"]) // 8 1046 self.assertTrue(reg_byte_size > 0) 1047 1048 # Run the process a bit so threads can start up, and collect register 1049 # info. 1050 context = self.run_process_then_stop(run_seconds=1) 1051 self.assertIsNotNone(context) 1052 1053 # Wait for 3 threads to be present. 1054 threads = self.wait_for_thread_count(3) 1055 self.assertEqual(len(threads), 3) 1056 1057 expected_reg_values = [] 1058 register_increment = 1 1059 next_value = None 1060 1061 # Set the same register in each of 3 threads to a different value. 1062 # Verify each one has the unique value. 1063 for thread in threads: 1064 # If we don't have a next value yet, start it with the initial read 1065 # value + 1 1066 if not next_value: 1067 # Read pre-existing register value. 1068 self.reset_test_sequence() 1069 self.test_sequence.add_log_lines( 1070 ["read packet: $p{0:x};thread:{1:x}#00".format(reg_index, thread), 1071 {"direction": "send", "regex": r"^\$([0-9a-fA-F]+)#", "capture": {1: "p_response"}}, 1072 ], True) 1073 context = self.expect_gdbremote_sequence() 1074 self.assertIsNotNone(context) 1075 1076 # Set the next value to use for writing as the increment plus 1077 # current value. 1078 p_response = context.get("p_response") 1079 self.assertIsNotNone(p_response) 1080 next_value = lldbgdbserverutils.unpack_register_hex_unsigned( 1081 endian, p_response) 1082 1083 # Set new value using P and thread suffix. 1084 self.reset_test_sequence() 1085 self.test_sequence.add_log_lines( 1086 [ 1087 "read packet: $P{0:x}={1};thread:{2:x}#00".format( 1088 reg_index, 1089 lldbgdbserverutils.pack_register_hex( 1090 endian, 1091 next_value, 1092 byte_size=reg_byte_size), 1093 thread), 1094 "send packet: $OK#00", 1095 ], 1096 True) 1097 context = self.expect_gdbremote_sequence() 1098 self.assertIsNotNone(context) 1099 1100 # Save the value we set. 1101 expected_reg_values.append(next_value) 1102 1103 # Increment value for next thread to use (we want them all 1104 # different so we can verify they wrote to each thread correctly 1105 # next.) 1106 next_value += register_increment 1107 1108 # Revisit each thread and verify they have the expected value set for 1109 # the register we wrote. 1110 thread_index = 0 1111 for thread in threads: 1112 # Read pre-existing register value. 1113 self.reset_test_sequence() 1114 self.test_sequence.add_log_lines( 1115 ["read packet: $p{0:x};thread:{1:x}#00".format(reg_index, thread), 1116 {"direction": "send", "regex": r"^\$([0-9a-fA-F]+)#", "capture": {1: "p_response"}}, 1117 ], True) 1118 context = self.expect_gdbremote_sequence() 1119 self.assertIsNotNone(context) 1120 1121 # Get the register value. 1122 p_response = context.get("p_response") 1123 self.assertIsNotNone(p_response) 1124 read_value = lldbgdbserverutils.unpack_register_hex_unsigned( 1125 endian, p_response) 1126 1127 # Make sure we read back what we wrote. 1128 self.assertEqual(read_value, expected_reg_values[thread_index]) 1129 thread_index += 1 1130