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