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