| import datetime | |
| import http.server | |
| import ipaddress | |
| import os | |
| import shutil | |
| import socketserver | |
| import ssl | |
| import sys | |
| import threading | |
| import time | |
| from functools import partial | |
| import lldb | |
| from lldbsuite.test.decorators import * | |
| from lldbsuite.test.lldbtest import * | |
| """ | |
| Test debug symbol acquisition from a local SymStore repository. This can work | |
| cross-platform and for arbitrary debug info formats. We only support PDB | |
| currently. | |
| """ | |
| class MockedSymStore: | |
| """ | |
| Context Manager to populate a file structure equivalent to SymStore.exe | |
| """ | |
| def __init__(self, test: TestBase, exe, pdb): | |
| self._test = test | |
| self._exe = exe | |
| self._pdb = pdb | |
| self.cache_dir = test.getBuildArtifact("cache") | |
| def get_key_pdb(self, exe): | |
| """ | |
| Module UUID: 12345678-1234-5678-9ABC-DEF012345678-00000001 | |
| To SymStore key: 12345678123456789ABCDEF0123456781 | |
| """ | |
| spec = lldb.SBModuleSpec() | |
| spec.SetFileSpec(lldb.SBFileSpec(self._test.getBuildArtifact(exe))) | |
| module = lldb.SBModule(spec) | |
| raw = module.GetUUIDString().replace("-", "").upper() | |
| if len(raw) != 40: | |
| raise RuntimeError("Unexpected number of bytes in embedded UUID") | |
| guid_hex = raw[:32] | |
| age = int(raw[32:], 16) | |
| return guid_hex + str(age) | |
| def __enter__(self): | |
| """ | |
| Mock local symstore directory tree, move PDB there and report path. | |
| """ | |
| key = None | |
| if self._test.getDebugInfo() == "pdb": | |
| key = self.get_key_pdb(self._exe) | |
| self._test.assertIsNotNone(key) | |
| symstore_dir = self._test.getBuildArtifact("symstore") | |
| pdb_dir = os.path.join(symstore_dir, self._pdb, key) | |
| os.makedirs(pdb_dir, exist_ok=True) | |
| shutil.move( | |
| self._test.getBuildArtifact(self._pdb), | |
| os.path.join(pdb_dir, self._pdb), | |
| ) | |
| # We always configure a valid fallback cache, because we might not have | |
| # permission to write outside the test directory. | |
| self._test.runCmd( | |
| f"settings set plugin.symbol-locator.symstore.cache {self.cache_dir}" | |
| ) | |
| return symstore_dir | |
| def __exit__(self, *exc_info): | |
| """ | |
| Reset settings | |
| """ | |
| self._test.runCmd("settings clear plugin.symbol-locator.symstore") | |
| class NtSymbolPath: | |
| """ | |
| Context Manager to temporarily set the _NT_SYMBOL_PATH environment variable. | |
| """ | |
| def __init__(self, value): | |
| self._value = value | |
| self._saved = None | |
| def __enter__(self): | |
| self._saved = os.environ.get("_NT_SYMBOL_PATH") | |
| os.environ["_NT_SYMBOL_PATH"] = self._value | |
| def __exit__(self, *exc_info): | |
| if self._saved is None: | |
| os.environ.pop("_NT_SYMBOL_PATH", None) | |
| else: | |
| os.environ["_NT_SYMBOL_PATH"] = self._saved | |
| class HTTPServer: | |
| """ | |
| Context Manager to serve a local directory tree via HTTP. | |
| """ | |
| def __init__(self, dir, handler=None): | |
| address = ("localhost", 0) # auto-select free port | |
| if handler is None: | |
| handler = partial(http.server.SimpleHTTPRequestHandler, directory=dir) | |
| self._server = socketserver.ThreadingTCPServer(address, handler) | |
| self._thread = threading.Thread(target=self._server.serve_forever, daemon=True) | |
| def __enter__(self): | |
| self._thread.start() | |
| host, port = self._server.server_address | |
| return f"http://{host}:{port}" | |
| def __exit__(self, *exc_info): | |
| if self._server: | |
| self._server.shutdown() | |
| self._server.server_close() | |
| if self._thread: | |
| self._thread.join() | |
| class HTTPSServer: | |
| """ | |
| Context Manager to serve a local directory tree via HTTPS. | |
| """ | |
| class ErrorAwareTCPServer(socketserver.ThreadingTCPServer): | |
| """TCP layer that will suppress errors of the given type.""" | |
| def __init__(self, address, handler, err): | |
| self.err = err | |
| super().__init__(address, handler) | |
| def handle_error(self, request, client_address): | |
| if isinstance(sys.exc_info()[1], self.err): | |
| return | |
| super().handle_error(request, client_address) | |
| def __init__(self, dir=None, handler=None, cert=None): | |
| if handler is None: | |
| handler = partial(http.server.SimpleHTTPRequestHandler, directory=dir) | |
| ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) | |
| ctx.load_cert_chain(cert.file, cert.key_file) | |
| address = ("localhost", 0) # auto-select free port | |
| self._server = self.ErrorAwareTCPServer(address, handler, ssl.SSLError) | |
| self._server.socket = ctx.wrap_socket(self._server.socket, server_side=True) | |
| self._thread = threading.Thread(target=self._server.serve_forever, daemon=True) | |
| def __enter__(self): | |
| self._thread.start() | |
| host, port = self._server.server_address | |
| return f"https://{host}:{port}" | |
| def __exit__(self, *exc_info): | |
| if self._server: | |
| self._server.shutdown() | |
| self._server.server_close() | |
| if self._thread: | |
| self._thread.join() | |
| class SelfSignedCert: | |
| """ | |
| Self-signed cert/key pair for localhost. | |
| """ | |
| def __init__(self, tmpdir): | |
| from cryptography import x509 | |
| from cryptography.x509.oid import NameOID | |
| from cryptography.hazmat.primitives import hashes, serialization | |
| from cryptography.hazmat.primitives.asymmetric import rsa | |
| key = rsa.generate_private_key(public_exponent=65537, key_size=2048) | |
| name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "localhost")]) | |
| now = datetime.datetime.now(datetime.timezone.utc) | |
| cert = ( | |
| x509.CertificateBuilder() | |
| .subject_name(name) | |
| .issuer_name(name) | |
| .public_key(key.public_key()) | |
| .serial_number(x509.random_serial_number()) | |
| .not_valid_before(now) | |
| .not_valid_after(now + datetime.timedelta(days=1)) | |
| .add_extension( | |
| x509.SubjectAlternativeName( | |
| [ | |
| x509.DNSName("localhost"), | |
| x509.IPAddress(ipaddress.IPv4Address("127.0.0.1")), | |
| ] | |
| ), | |
| critical=False, | |
| ) | |
| .sign(key, hashes.SHA256()) | |
| ) | |
| os.makedirs(tmpdir, exist_ok=False) | |
| self.key_file = os.path.join(tmpdir, "server.key") | |
| self.file = os.path.join(tmpdir, "server.crt") | |
| with open(self.key_file, "wb") as f: | |
| f.write( | |
| key.private_bytes( | |
| serialization.Encoding.PEM, | |
| serialization.PrivateFormat.TraditionalOpenSSL, | |
| serialization.NoEncryption(), | |
| ) | |
| ) | |
| with open(self.file, "wb") as f: | |
| f.write(cert.public_bytes(serialization.Encoding.PEM)) | |
| self.fingerprint = cert.fingerprint(hashes.SHA256()).hex() | |
| class RedirectHandler(http.server.BaseHTTPRequestHandler): | |
| base_url = None | |
| def do_GET(self): | |
| self.send_response(301) | |
| self.send_header("Location", self.base_url + self.path) | |
| self.end_headers() | |
| def log_message(self, *args): | |
| pass # suppress request logs | |
| class RequestCounter(http.server.SimpleHTTPRequestHandler): | |
| requests = 0 # class-level so all instances share one counter | |
| def __init__(self, *args, directory=None, **kwargs): | |
| super().__init__(*args, directory=directory, **kwargs) | |
| def do_GET(self): | |
| RequestCounter.requests += 1 | |
| super().do_GET() | |
| class NeverRespondsHTTPHandler(http.server.BaseHTTPRequestHandler): | |
| """HTTP request handler that never responds.""" | |
| can_finish = threading.Event() | |
| def do_GET(self): | |
| self.can_finish.wait() | |
| def log_message(self, *args): | |
| pass # suppress server-side output | |
| class PartialResponseHTTPHandler(http.server.BaseHTTPRequestHandler): | |
| can_finish = threading.Event() | |
| def do_GET(self): | |
| self.send_response(200) | |
| self.send_header("Content-Type", "application/octet-stream") | |
| self.send_header("Content-Length", "1000000") | |
| self.end_headers() | |
| for _ in range(10): | |
| self.wfile.write(b"a" * 4096) | |
| self.wfile.flush() | |
| # Wait until the client has finished running. | |
| self.can_finish.wait() | |
| def log_message(self, *args): | |
| pass # suppress server-side output | |
| class SymStoreTests(TestBase): | |
| # Build artifacts can't be shared across test functions because | |
| # MockedSymStore moves the built PDB out of the build directory to simulate | |
| # it being absent from the normal lookup path. | |
| SHARED_BUILD_TESTCASE = False | |
| def build_inferior(self): | |
| if self.getDebugInfo() != "pdb": | |
| self.skipTest("Non-PDB debug info variants not yet supported") | |
| self.build() | |
| exe_file = "a.out" | |
| sym_file = "a.pdb" | |
| self.assertTrue(os.path.isfile(self.getBuildArtifact(exe_file))) | |
| self.assertTrue(os.path.isfile(self.getBuildArtifact(sym_file))) | |
| return exe_file, sym_file | |
| def assertFiles(self, dir, expected): | |
| actual = sum(len(f) for _, _, f in os.walk(dir)) | |
| self.assertEqual(actual, expected) | |
| def try_breakpoint(self, exe, should_have_loc, ext_lookup=True): | |
| enable = "true" if ext_lookup else "false" | |
| self.runCmd(f"settings set symbols.enable-external-lookup {enable}") | |
| target = self.dbg.CreateTarget(self.getBuildArtifact(exe)) | |
| self.assertTrue(target and target.IsValid(), "Target is valid") | |
| bp = target.BreakpointCreateByName("func") | |
| self.assertTrue(bp and bp.IsValid(), "Breakpoint is valid") | |
| self.assertEqual(bp.GetNumLocations(), 1 if should_have_loc else 0) | |
| self.dbg.DeleteTarget(target) | |
| def test_no_symstore_url(self): | |
| """ | |
| Check that breakpoint doesn't resolve without SymStore. | |
| """ | |
| exe, sym = self.build_inferior() | |
| with MockedSymStore(self, exe, sym): | |
| self.try_breakpoint(exe, should_have_loc=False) | |
| def test_external_lookup_off(self): | |
| """ | |
| Check that breakpoint doesn't resolve with external lookup disabled. | |
| """ | |
| exe, sym = self.build_inferior() | |
| with MockedSymStore(self, exe, sym) as dir: | |
| self.runCmd(f"settings set plugin.symbol-locator.symstore.urls {dir}") | |
| self.try_breakpoint(exe, ext_lookup=False, should_have_loc=False) | |
| def test_local_dir(self): | |
| """ | |
| Check that breakpoint resolves with local SymStore. | |
| """ | |
| exe, sym = self.build_inferior() | |
| with MockedSymStore(self, exe, sym) as dir: | |
| self.runCmd(f"settings set plugin.symbol-locator.symstore.urls {dir}") | |
| self.try_breakpoint(exe, should_have_loc=True) | |
| def test_http_not_found(self): | |
| """ | |
| Check that we don't issue a warning for a 404 response from a symbol server. | |
| """ | |
| exe, sym = self.build_inferior() | |
| with MockedSymStore(self, exe, sym) as symstore_dir: | |
| os.makedirs(f"{symstore_dir}_empty", exist_ok=False) | |
| with HTTPServer(f"{symstore_dir}_empty") as url: | |
| self.runCmd(f"settings set plugin.symbol-locator.symstore.urls {url}") | |
| warnings = "" | |
| with open(self.getBuildArtifact("stderr.txt"), "w+b") as err_file: | |
| self.dbg.SetErrorFileHandle(err_file, False) | |
| self.try_breakpoint(exe, should_have_loc=False) | |
| self.dbg.SetErrorFileHandle(sys.stderr, False) | |
| err_file.seek(0) | |
| warnings = err_file.read().decode() | |
| self.assertEqual(warnings, "") | |
| def test_http(self): | |
| """ | |
| Check that breakpoint resolves with remote SymStore. | |
| """ | |
| exe, sym = self.build_inferior() | |
| with MockedSymStore(self, exe, sym) as dir: | |
| with HTTPServer(dir) as url: | |
| self.runCmd(f"settings set plugin.symbol-locator.symstore.urls {url}") | |
| self.try_breakpoint(exe, should_have_loc=True) | |
| def test_sympath_local_dir(self): | |
| """ | |
| Check that breakpoint resolves with plain directory in _NT_SYMBOL_PATH. | |
| The PDB is not copied to the configured cache. | |
| """ | |
| exe, sym = self.build_inferior() | |
| symstore = MockedSymStore(self, exe, sym) | |
| with symstore as dir: | |
| with NtSymbolPath(dir): | |
| self.try_breakpoint(exe, should_have_loc=True) | |
| self.assertFiles(symstore.cache_dir, 0) | |
| def test_sympath_local_srv(self): | |
| """ | |
| Check that breakpoint resolves with local directory in server notation | |
| in _NT_SYMBOL_PATH. The PDB is not copied to the configured cache. | |
| """ | |
| exe, sym = self.build_inferior() | |
| symstore = MockedSymStore(self, exe, sym) | |
| with symstore as dir: | |
| with NtSymbolPath(f"srv*{dir}"): | |
| self.try_breakpoint(exe, should_have_loc=True) | |
| self.assertFiles(symstore.cache_dir, 0) | |
| def test_sympath_srv(self): | |
| """ | |
| Check that breakpoint resolves with an HTTP symbol server in _NT_SYMBOL_PATH | |
| using the srv* notation. The PDB is stored in the configured cache. | |
| """ | |
| exe, sym = self.build_inferior() | |
| symstore = MockedSymStore(self, exe, sym) | |
| with symstore as dir: | |
| self.assertFiles(symstore.cache_dir, 0) | |
| with HTTPServer(dir) as url: | |
| with NtSymbolPath(f"srv*{url}"): | |
| self.try_breakpoint(exe, should_have_loc=True) | |
| key = symstore.get_key_pdb(exe) | |
| cache_file = os.path.join(symstore.cache_dir, sym, key, sym) | |
| self.assertTrue(os.path.isfile(cache_file)) | |
| self.assertFiles(symstore.cache_dir, 1) | |
| def test_sympath_cache_explicit(self): | |
| """ | |
| Check PDB storage with explicit cache in _NT_SYMBOL_PATH. | |
| """ | |
| exe, sym = self.build_inferior() | |
| symstore = MockedSymStore(self, exe, sym) | |
| with symstore as dir: | |
| with HTTPServer(dir) as url: | |
| explicit_cache = self.getBuildArtifact("explicit_cache") | |
| with NtSymbolPath(f"srv*{explicit_cache}*{url}"): | |
| self.try_breakpoint(exe, should_have_loc=True) | |
| self.assertFiles(symstore.cache_dir, 0) | |
| self.assertFiles(explicit_cache, 1) | |
| def test_sympath_cache_implicit(self): | |
| """ | |
| Check PDB storage with implicit cache in _NT_SYMBOL_PATH. | |
| """ | |
| exe, sym = self.build_inferior() | |
| symstore = MockedSymStore(self, exe, sym) | |
| with symstore as dir: | |
| with HTTPServer(dir) as url: | |
| implicit_cache = self.getBuildArtifact("implicit_cache") | |
| with NtSymbolPath(f"cache*{implicit_cache};srv*{url}"): | |
| self.try_breakpoint(exe, should_have_loc=True) | |
| self.assertFiles(symstore.cache_dir, 0) | |
| self.assertFiles(implicit_cache, 1) | |
| def test_sympath_cache_invalid(self): | |
| """ | |
| Check that PDB is stored in configured default cache | |
| if path in _NT_SYMBOL_PATH is invalid. | |
| """ | |
| exe, sym = self.build_inferior() | |
| symstore = MockedSymStore(self, exe, sym) | |
| with symstore as dir: | |
| with HTTPServer(dir) as url: | |
| invalid_cache = ":\\<invalid_path>" | |
| self.assertFiles(symstore.cache_dir, 0) | |
| with NtSymbolPath(f"cache*{invalid_cache};srv*{url}"): | |
| self.try_breakpoint(exe, should_have_loc=True) | |
| self.assertFiles(symstore.cache_dir, 1) | |
| def test_sympath_cache_empty(self): | |
| """ | |
| Check that PDB is stored in configured default cache | |
| if path in _NT_SYMBOL_PATH is empty. | |
| """ | |
| exe, sym = self.build_inferior() | |
| symstore = MockedSymStore(self, exe, sym) | |
| with symstore as dir: | |
| with HTTPServer(dir) as url: | |
| self.assertFiles(symstore.cache_dir, 0) | |
| with NtSymbolPath(f"cache*;srv*{url}"): | |
| self.try_breakpoint(exe, should_have_loc=True) | |
| self.assertFiles(symstore.cache_dir, 1) | |
| def test_lookup_order(self): | |
| """ | |
| Check that _NT_SYMBOL_PATH takes precedence over symstore.urls setting. | |
| """ | |
| exe, sym = self.build_inferior() | |
| symstore = MockedSymStore(self, exe, sym) | |
| RequestCounter.requests = 0 | |
| with symstore as dir: | |
| with HTTPServer(dir, RequestCounter) as url: | |
| self.runCmd(f"settings set plugin.symbol-locator.symstore.urls {url}") | |
| with NtSymbolPath(dir): | |
| self.try_breakpoint(exe, should_have_loc=True) | |
| self.assertEqual(RequestCounter.requests, 0) | |
| def test_http_timeout(self): | |
| """ | |
| Check that a warning is emitted and no symbol is found when the server | |
| takes longer to respond than the configured timeout. | |
| """ | |
| exe, sym = self.build_inferior() | |
| with MockedSymStore(self, exe, sym) as dir: | |
| NeverRespondsHTTPHandler.can_finish.clear() | |
| self.runCmd("settings set plugin.symbol-locator.symstore.timeout 1") | |
| with HTTPServer(dir, NeverRespondsHTTPHandler) as url: | |
| try: | |
| self.runCmd( | |
| f"settings set plugin.symbol-locator.symstore.urls {url}" | |
| ) | |
| with open(self.getBuildArtifact("stderr.txt"), "w+b") as err_file: | |
| self.dbg.SetErrorFileHandle(err_file, False) | |
| self.try_breakpoint(exe, should_have_loc=False) | |
| self.dbg.SetErrorFileHandle(sys.stderr, False) | |
| err_file.seek(0) | |
| warnings = err_file.read().decode() | |
| self.assertIn("failed to download", warnings) | |
| finally: | |
| NeverRespondsHTTPHandler.can_finish.set() | |
| def test_http_partial_response(self): | |
| """ | |
| Check that a warning is emitted and no symbol is found when the server | |
| sends a partial response but it ultimately times out. | |
| """ | |
| exe, sym = self.build_inferior() | |
| store = MockedSymStore(self, exe, sym) | |
| with store as dir: | |
| PartialResponseHTTPHandler.can_finish.clear() | |
| self.runCmd("settings set plugin.symbol-locator.symstore.timeout 1") | |
| with HTTPServer(dir, PartialResponseHTTPHandler) as url: | |
| try: | |
| self.runCmd( | |
| f"settings set plugin.symbol-locator.symstore.urls {url}" | |
| ) | |
| with open(self.getBuildArtifact("stderr.txt"), "w+b") as err_file: | |
| self.dbg.SetErrorFileHandle(err_file, False) | |
| self.try_breakpoint(exe, should_have_loc=False) | |
| self.dbg.SetErrorFileHandle(sys.stderr, False) | |
| err_file.seek(0) | |
| warnings = err_file.read().decode() | |
| self.assertIn("failed to download", warnings) | |
| self.assertFiles(store.cache_dir, 0) | |
| finally: | |
| PartialResponseHTTPHandler.can_finish.set() | |
| @skipUnlessPackageAvailable("cryptography") | |
| def test_https(self): | |
| """ | |
| Check that breakpoint resolves with remote SymStore via HTTPS. | |
| """ | |
| exe, sym = self.build_inferior() | |
| with MockedSymStore(self, exe, sym) as symstore_dir: | |
| cert = SelfSignedCert(self.getBuildArtifact("cert")) | |
| with HTTPSServer(dir=symstore_dir, cert=cert) as https_url: | |
| # We accept only the self-signed certificate with this fingerprint | |
| self.runCmd( | |
| f"settings set plugin.symbol-locator.symstore.tls-cert-fingerprint {cert.fingerprint}" | |
| ) | |
| self.runCmd( | |
| f"settings set plugin.symbol-locator.symstore.urls {https_url}" | |
| ) | |
| self.try_breakpoint(exe, should_have_loc=True) | |
| @skipUnlessPackageAvailable("cryptography") | |
| def test_https_reject_selfsigned_cert(self): | |
| """ | |
| Check that LLDB rejects an HTTPS server with an untrusted self-signed cert. | |
| """ | |
| exe, sym = self.build_inferior() | |
| with MockedSymStore(self, exe, sym) as symstore_dir: | |
| cert = SelfSignedCert(self.getBuildArtifact("cert")) | |
| with HTTPSServer(dir=symstore_dir, cert=cert) as https_url: | |
| # No fingerprint set | |
| self.runCmd( | |
| f"settings set plugin.symbol-locator.symstore.urls {https_url}" | |
| ) | |
| self.try_breakpoint(exe, should_have_loc=False) | |
| # Incorrect fingerprint set | |
| bogus = "DEADBEEFCAFEBABE" | |
| self.runCmd( | |
| f"settings set plugin.symbol-locator.symstore.tls-cert-fingerprint {bogus}{bogus}{bogus}{bogus}" | |
| ) | |
| self.try_breakpoint(exe, should_have_loc=False) | |
| @skipUnlessPackageAvailable("cryptography") | |
| def test_https_reject_redirect_http(self): | |
| """ | |
| Check that LLDB does not retrieve symbols from servers that redirect with security downgrades. | |
| """ | |
| exe, sym = self.build_inferior() | |
| with MockedSymStore(self, exe, sym) as symstore_dir: | |
| with HTTPServer(symstore_dir) as http_url: | |
| RedirectHandler.base_url = http_url | |
| cert = SelfSignedCert(self.getBuildArtifact("cert")) | |
| with HTTPSServer(handler=RedirectHandler, cert=cert) as https_url: | |
| self.runCmd( | |
| f"settings set plugin.symbol-locator.symstore.urls {https_url}" | |
| ) | |
| self.runCmd( | |
| f"settings set plugin.symbol-locator.symstore.tls-cert-fingerprint {cert.fingerprint}" | |
| ) | |
| self.try_breakpoint(exe, should_have_loc=False) |