Add new TLS tests to DMM's 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
 
@@ -16,13 +17,14 @@
 import system_test_utilities  # noqa: E402
 
 
-class SystemTests:
+# Defines a subset of system tests to validate basic DMM functionality. This is run as a part of the full SystemTests class, and
+# independently for test classes which do not require running the entire suite (TLS-enabled gRPC tests today).
+class BasicValidationTests:
     @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
 
-    # Basic usability tests
     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.
@@ -40,6 +42,8 @@ def test_multi_point_acquisition(self, session):
         measurements = session.read_multi_point(8)
         assert len(measurements) == 8
 
+
+class SystemTests(BasicValidationTests):
     # Attribute tests
     def test_vi_string_attribute(self, session):
         assert session.instrument_model == 'NI PXIe-4082'
@@ -312,7 +316,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 +332,19 @@ def test_fetch_waveform_into(self, session):
             assert not math.isnan(sample)
 
 
-class TestGrpc(SystemTests):
+class TestGrpcNoTLS(SystemTests):
     @pytest.fixture(scope='class')
-    def grpc_channel(self):
+    @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.json')
+        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')
-    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 +376,44 @@ 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}'
+
+
+@pytest.mark.skipif(sys.maxsize < 2**32, reason="TLS configuration and certificate exchange scripts are not supported in 32-bit processes")
+class TestGrpcSecuredTLS(BasicValidationTests):
+    @pytest.fixture(scope='class')
+    @classmethod
+    def grpc_channel(cls):
+        system_test_utilities.configure_tls_modes_secure(service="ni-grpc-device", server_host="localhost")
+        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_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}
+
+
+@pytest.mark.skipif(sys.maxsize < 2**32, reason="TLS configuration and certificate exchange scripts are not supported in 32-bit processes")
+class TestGrpcUnsecuredTLS(BasicValidationTests):
+    @pytest.fixture(scope='class')
+    @classmethod
+    def grpc_channel(cls):
+        system_test_utilities.configure_tls_modes_insecure(service="ni-grpc-device", server_host="localhost")
+
+        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}

src/shared/system_test_utilities.py

@@ -1,8 +1,10 @@
+import json
 import os
 import pathlib
 import pytest
 import re
 import subprocess
+import sys
 import threading
 import time
 
@@ -104,3 +106,131 @@ 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")
+
+    subprocess.run(command, check=True, env=env)
+
+
+def configure_tls_modes(
+    service: str,
+    server_host: str,
+    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 = "--local-server"
+    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 (
+            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")
+
+    subprocess.run(command, check=True, env=env)
+
+
+def configure_tls_modes_secure(
+    service: str,
+    server_host: str
+):
+    configure_tls_modes(
+        service=service,
+        server_host=server_host,
+        server_cert_mode="ManagedSelfSigned",
+        server_client_mode="ManagedSelfSigned",
+        client_cert_mode="Managed",
+        client_server_mode="TrustedCertificates"
+    )
+
+
+def configure_tls_modes_insecure(
+    service: str,
+    server_host: str
+):
+    configure_tls_modes(
+        service=service,
+        server_host=server_host,
+        server_cert_mode="Disabled",
+        server_client_mode="Disabled",
+        client_cert_mode="Disabled",
+        client_server_mode="Disabled"
+    )