Add new TLS System tests

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
+    } 
+ }

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

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()

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)