diff --git a/src/nidmm/system_tests/grpc_server_config.json b/src/nidmm/system_tests/grpc_server_config_no_tls.json similarity index 100% rename from src/nidmm/system_tests/grpc_server_config.json rename to src/nidmm/system_tests/grpc_server_config_no_tls.json diff --git a/src/nidmm/system_tests/grpc_server_config_tls.json b/src/nidmm/system_tests/grpc_server_config_tls.json new file mode 100644 index 0000000000..4300dc473d --- /dev/null +++ b/src/nidmm/system_tests/grpc_server_config_tls.json @@ -0,0 +1,8 @@ +{ + "address": "[::]", + "port": 31762, + "security": "ni-tls-config", + "feature_toggles": { + "ni-tls-config": true + } + } diff --git a/src/nidmm/system_tests/test_system_nidmm.py b/src/nidmm/system_tests/test_system_nidmm.py index 55cdea3988..830fa5008b 100644 --- a/src/nidmm/system_tests/test_system_nidmm.py +++ b/src/nidmm/system_tests/test_system_nidmm.py @@ -7,6 +7,7 @@ import grpc import hightime +import nitlsconfig import numpy import pytest @@ -312,7 +313,8 @@ def test_multi_threading_ivi_synchronized_wrapper_releases_lock(self, session): class TestLibrary(SystemTests): @pytest.fixture(scope='class') - def session_creation_kwargs(self): + @classmethod + def session_creation_kwargs(cls): return {} def test_fetch_waveform_into(self, session): @@ -327,17 +329,29 @@ def test_fetch_waveform_into(self, session): assert not math.isnan(sample) -class TestGrpc(SystemTests): +class TestGrpcSecuredTLS(SystemTests): @pytest.fixture(scope='class') - def grpc_channel(self): + @classmethod + def grpc_channel(cls): + system_test_utilities.configure_tls_modes( + service="ni-grpc-device", + server_host="localhost", + server_cert_mode="ManagedSelfSigned", + server_client_mode="ManagedSelfSigned", + client_cert_mode="Managed", + client_server_mode="TrustedCertificates" + ) + system_test_utilities.exchange_certificates("localhost") + current_directory = os.path.dirname(os.path.abspath(__file__)) - config_file_path = os.path.join(current_directory, 'grpc_server_config.json') + config_file_path = os.path.join(current_directory, 'grpc_server_config_tls.json') with system_test_utilities.GrpcServerProcess(config_file_path) as proc: - channel = grpc.insecure_channel(f"localhost:{proc.server_port}") + channel = nitlsconfig.create_grpc_device_channel('localhost', proc.server_port) yield channel @pytest.fixture(scope='class') - def session_creation_kwargs(self, grpc_channel): + @classmethod + def session_creation_kwargs(cls, grpc_channel): grpc_options = nidmm.GrpcSessionOptions(grpc_channel, '') return {'grpc_options': grpc_options} @@ -369,3 +383,158 @@ def test_attach_to_non_existent_session(self, grpc_channel): assert e.rpc_code == expected_grpc_error assert e.description == expected_error_message assert str(e) == f'{expected_grpc_error}: {expected_error_message}' + + +class TestGrpcUnsecuredTLS: + @pytest.fixture(scope='function') + def session(self, session_creation_kwargs): + with nidmm.Session('FakeDevice', False, True, 'Simulate=1, DriverSetup=Model:4082; BoardType:PXIe', **session_creation_kwargs) as simulated_session: + yield simulated_session + + @pytest.fixture(scope='class') + @classmethod + def grpc_channel(cls): + system_test_utilities.configure_tls_modes( + service="ni-grpc-device", + server_host="localhost", + server_cert_mode="Disabled", + server_client_mode="Disabled", + client_cert_mode="Disabled", + client_server_mode="Disabled" + ) + + current_directory = os.path.dirname(os.path.abspath(__file__)) + config_file_path = os.path.join(current_directory, 'grpc_server_config_tls.json') + with system_test_utilities.GrpcServerProcess(config_file_path) as proc: + channel = nitlsconfig.create_grpc_device_channel('localhost', proc.server_port) + yield channel + + @pytest.fixture(scope='class') + @classmethod + def session_creation_kwargs(cls, grpc_channel): + grpc_options = nidmm.GrpcSessionOptions(grpc_channel, '') + return {'grpc_options': grpc_options} + + def test_take_simple_measurement_works(self, session): + session.configure_measurement_digits(nidmm.Function.DC_CURRENT, 1, 5.5) + assert session.read() != 0 # Assumes DMM reading is not exactly zero to support non-connected modules and simulated modules. + + def test_acquisition(self, session): + session.configure_measurement_digits(nidmm.Function.DC_CURRENT, 1, 5.5) + with session.initiate(): + session.fetch() + with session.initiate(): + session.fetch() + + def test_multi_point_acquisition(self, session): + session.configure_multi_point(4, 2) + session.configure_measurement_digits(nidmm.Function.DC_VOLTS, 1, 5.5) + measurements = session.read_multi_point(8) + assert len(measurements) == 8 + + +class TestGrpcNoTLS: + @pytest.fixture(scope='function') + def session(self, session_creation_kwargs): + with nidmm.Session('FakeDevice', False, True, 'Simulate=1, DriverSetup=Model:4082; BoardType:PXIe', **session_creation_kwargs) as simulated_session: + yield simulated_session + + @pytest.fixture(scope='class') + @classmethod + def grpc_channel(cls): + current_directory = os.path.dirname(os.path.abspath(__file__)) + config_file_path = os.path.join(current_directory, 'grpc_server_config_no_tls.json') + with system_test_utilities.GrpcServerProcess(config_file_path) as proc: + channel = grpc.insecure_channel(f"localhost:{proc.server_port}") + yield channel + + @pytest.fixture(scope='class') + @classmethod + def session_creation_kwargs(cls, grpc_channel): + grpc_options = nidmm.GrpcSessionOptions(grpc_channel, '') + return {'grpc_options': grpc_options} + + def test_take_simple_measurement_works(self, session): + session.configure_measurement_digits(nidmm.Function.DC_CURRENT, 1, 5.5) + assert session.read() != 0 # Assumes DMM reading is not exactly zero to support non-connected modules and simulated modules. + + def test_acquisition(self, session): + session.configure_measurement_digits(nidmm.Function.DC_CURRENT, 1, 5.5) + with session.initiate(): + session.fetch() + with session.initiate(): + session.fetch() + + def test_multi_point_acquisition(self, session): + session.configure_multi_point(4, 2) + session.configure_measurement_digits(nidmm.Function.DC_VOLTS, 1, 5.5) + measurements = session.read_multi_point(8) + assert len(measurements) == 8 + + +def test_unsecured_client(): + system_test_utilities.configure_tls_modes( + service="ni-grpc-device", + server_host="localhost", + server_cert_mode="ManagedSelfSigned", + server_client_mode="ManagedSelfSigned", + client_cert_mode="Managed", + client_server_mode="TrustedCertificates" + ) + system_test_utilities.exchange_certificates("localhost") + + system_test_utilities.configure_tls_modes( + service="ni-grpc-device", + server_host="localhost", + server_cert_mode="ManagedSelfSigned", + server_client_mode="ManagedSelfSigned", + client_cert_mode="Disabled", + client_server_mode="Disabled" + ) + + current_directory = os.path.dirname(os.path.abspath(__file__)) + config_file_path = os.path.join(current_directory, 'grpc_server_config_tls.json') + + # Attempt to connect to the server. Since it is expecting a TLS-enabled client, this should fail. + with system_test_utilities.GrpcServerProcess(config_file_path) as proc: + unsecured_client_channel = nitlsconfig.create_grpc_device_channel('localhost', proc.server_port) + grpc_options = nidmm.GrpcSessionOptions(unsecured_client_channel, '') + try: + with nidmm.Session('FakeDevice', False, True, 'Simulate=1, DriverSetup=Model:4082; BoardType:PXIe', grpc_options=grpc_options): + assert False + except nidmm.Error: + pass + + +def test_unsecured_server(): + system_test_utilities.configure_tls_modes( + service="ni-grpc-device", + server_host="localhost", + server_cert_mode="ManagedSelfSigned", + server_client_mode="ManagedSelfSigned", + client_cert_mode="Managed", + client_server_mode="TrustedCertificates" + ) + system_test_utilities.exchange_certificates("localhost") + + system_test_utilities.configure_tls_modes( + service="ni-grpc-device", + server_host="localhost", + server_cert_mode="Disabled", + server_client_mode="Disabled", + client_cert_mode="Managed", + client_server_mode="TrustedCertificates" + ) + + current_directory = os.path.dirname(os.path.abspath(__file__)) + config_file_path = os.path.join(current_directory, 'grpc_server_config_tls.json') + + # Attempt to connect to the server. Since the client is expecting a TLS-enabled server, this should fail. + with system_test_utilities.GrpcServerProcess(config_file_path) as proc: + unsecured_server_channel = nitlsconfig.create_grpc_device_channel('localhost', proc.server_port) + grpc_options = nidmm.GrpcSessionOptions(unsecured_server_channel, '') + try: + with nidmm.Session('FakeDevice', False, True, 'Simulate=1, DriverSetup=Model:4082; BoardType:PXIe', grpc_options=grpc_options): + assert False + except nidmm.Error: + pass diff --git a/src/shared/nitlsconfig_32_bit_patch.py b/src/shared/nitlsconfig_32_bit_patch.py new file mode 100644 index 0000000000..db22ac51e6 --- /dev/null +++ b/src/shared/nitlsconfig_32_bit_patch.py @@ -0,0 +1,32 @@ +import ctypes +import os +import subprocess + + +def _patch_subprocess_for_32_bit_nitlsconfig_lookup(): + # Because nitlsconfig lives in System32, and the 32-bit system tests are run on a 64-bit machine, the installation + # of nitlsconfig is invisible by default. To get around this, we can disable Wow64 redirection. In order to minimize + # the impact of this, we patch the subprocess initialization specifically for calls to nitlsconfig + if os.name != "nt": + return + + original_init = subprocess.Popen.__init__ + + def patched_init(self, args, *posargs, **kwargs): + command = args[0] if isinstance(args, (list, tuple)) else args + is_nitlsconfig = isinstance(command, str) and os.path.splitext(os.path.basename(command))[0].lower() == "nitlsconfig" + if not is_nitlsconfig: + return original_init(self, args, *posargs, **kwargs) + + old = ctypes.c_void_p() + disabled = bool(ctypes.windll.kernel32.Wow64DisableWow64FsRedirection(ctypes.byref(old))) + try: + return original_init(self, args, *posargs, **kwargs) + finally: + if disabled: + ctypes.windll.kernel32.Wow64RevertWow64FsRedirection(old) + + subprocess.Popen.__init__ = patched_init + + +_patch_subprocess_for_32_bit_nitlsconfig_lookup() diff --git a/src/shared/system_test_utilities.py b/src/shared/system_test_utilities.py index dea3b2d1cd..7c5580f17f 100644 --- a/src/shared/system_test_utilities.py +++ b/src/shared/system_test_utilities.py @@ -1,11 +1,15 @@ +import json import os import pathlib import pytest import re import subprocess +import sys import threading import time +import nitlsconfig_32_bit_patch # noqa: F401 + class GrpcServerProcess: def __init__(self, config_file_path): @@ -104,3 +108,121 @@ def impl_test_multi_threading_ivi_synchronized_wrapper_releases_lock(ivi_method_ t2.start() t2.join() assert not t2.is_alive() + + +def exchange_certificates( + server_host: str, + server_user: str | None = None, + client_host: str | None = None, + client_user: str | None = None, + verbosity: int = 2, +): + # gRPC tests only run on Windows, so this isn't necessary on Linux. + if os.name != "nt": + return + + # 26.5 versions of ni-grpc-device server installers do not properly create the trusted.d directory, + # which causes issues with the certificate exchange process. This has been fixed in the 26.8 version + # of the installer, but it has not yet been released. For now, we're creating it manually; this can + # be removed once nimibot system tests are updated to test against >= 26.8 versions of the drivers. + trusted_servers_path = pathlib.Path(r"C:/ProgramData/National Instruments/nitlsconfig/server.d/ni-grpc-device/trusted.d") + trusted_servers_path.mkdir(parents=True, exist_ok=True) + + # 26.5 versions of ni-grpc-device client configuration use a default certificate_mode of Disabled, + # which prevents client-side certificate generation from this script. In 26.8 and beyond, the default + # is Managed. We set it manually here; this can be removed once nimibot system tests are updated to + # test against >= 26.8 versions of the drivers. + client_config_path = ( + pathlib.Path(os.environ["LOCALAPPDATA"]) + / "National Instruments" / "nitlsconfig" / "client.d" / "ni-grpc-device.conf.yml" + ) + content = client_config_path.read_text() + content = re.sub(r"(?m)^certificate_mode:.*$", "certificate_mode: Managed", content) + client_config_path.write_text(content) + + script_path = r"C:/NITests/nitlsconfigtest/exchange_certificates.py" + if not pathlib.Path(script_path).is_file(): + raise FileNotFoundError(f"Certificate exchange script not found: {script_path}") + + server_host_arg = f"--server-host={server_host}" + server_user_arg = f"--server-user={server_user}" if server_user else "--local-server" + client_host_arg = f"--client-host={client_host}" if client_host else None + client_user_arg = f"--client-user={client_user}" if client_user else None + + verbosity = max(0, min(verbosity, 4)) + verbosity_arg = { + 0: "-qq", + 1: "-q", + 3: "-v", + 4: "-vv", + }.get(verbosity) + + command = [sys.executable, str(pathlib.Path(script_path)), server_host_arg, server_user_arg] + command.extend(arg for arg in (client_host_arg, client_user_arg, verbosity_arg) if arg is not None) + + # The script expects this environment variable to be set + env = os.environ.copy() + env.setdefault("USERNAME", "Administrator") + + _run_nitlsconfigtest_script_with_patch(script_path, command[2:], env) + + +def configure_tls_modes( + service: str, + server_host: str, + server_user: str | None = None, + client_host: str | None = None, + client_user: str | None = None, + server_cert_mode: str | None = None, + server_client_mode: str | None = None, + client_cert_mode: str | None = None, + client_server_mode: str | None = None, +): + # gRPC tests only run on Windows, so this isn't necessary on Linux. + if os.name != "nt": + return + + script_path = r"C:/NITests/nitlsconfigtest/configure_tls_modes.py" + if not pathlib.Path(script_path).is_file(): + raise FileNotFoundError(f"Configure TLS modes script not found: {script_path}") + + service_arg = f"--service={service}" + server_host_arg = f"--server-host={server_host}" + server_user_arg = f"--server-user={server_user}" if server_user else "--local-server" + client_host_arg = f"--client-host={client_host}" if client_host else None + client_user_arg = f"--client-user={client_user}" if client_user else None + server_cert_mode_arg = f"--server-certificate-mode={server_cert_mode}" if server_cert_mode else None + server_client_mode_arg = f"--server-client-mode={server_client_mode}" if server_client_mode else None + client_cert_mode_arg = f"--client-certificate-mode={client_cert_mode}" if client_cert_mode else None + client_server_mode_arg = f"--client-server-mode={client_server_mode}" if client_server_mode else None + + command = [sys.executable, str(pathlib.Path(script_path)), service_arg, server_host_arg, server_user_arg] + command.extend( + arg + for arg in ( + client_host_arg, + client_user_arg, + server_cert_mode_arg, + server_client_mode_arg, + client_cert_mode_arg, + client_server_mode_arg, + ) + if arg is not None + ) + + # The script expects this environment variable to be set + env = os.environ.copy() + env.setdefault("USERNAME", "Administrator") + + _run_nitlsconfigtest_script_with_patch(script_path, command[2:], env) + +def _run_nitlsconfigtest_script_with_patch(script_path: str, args: list, env: dict) -> None: + # A bootstrap script is used to import the patcher so that the scripts can see the nitlsconfig executable even if + # they are in a 32-bit context. + bootstrap = ( + "import runpy, sys\n" + "import nitlsconfig_32_bit_patch\n" + f"sys.argv = [{script_path!r}] + {args!r}\n" + f"runpy.run_path({script_path!r}, run_name='__main__')\n" + ) + subprocess.run([sys.executable, "-c", bootstrap], check=True, env=env) \ No newline at end of file