From 568fce975975be09022a08d3793fc64333cfa77b Mon Sep 17 00:00:00 2001 From: Ryan Wixon Date: Wed, 9 Sep 2026 12:21:16 -0500 Subject: [PATCH 1/3] Add new TLS system tests --- src/nidmm/system_tests/test_system_nidmm.py | 56 ++++++++++- src/shared/system_test_utilities.py | 102 ++++++++++++++++++++ 2 files changed, 157 insertions(+), 1 deletion(-) diff --git a/src/nidmm/system_tests/test_system_nidmm.py b/src/nidmm/system_tests/test_system_nidmm.py index 55cdea3988..0a1d2781ec 100644 --- a/src/nidmm/system_tests/test_system_nidmm.py +++ b/src/nidmm/system_tests/test_system_nidmm.py @@ -327,9 +327,63 @@ 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): + system_test_utilities.write_grpc_device_server_config(use_tls_config=True) + system_test_utilities.exchange_certificates("localhost") + system_test_utilities.configure_tls_modes( + "ni-grpc-device-server", + "localhost", + "Disabled", + "Disabled", + "Disabled", + "Disabled" + ) + + current_directory = os.path.dirname(os.path.abspath(__file__)) + config_file_path = os.path.join(current_directory, 'grpc_server_config.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): + grpc_options = nidmm.GrpcSessionOptions(grpc_channel, '') + return {'grpc_options': grpc_options} + + +class TestGrpcUnsecuredTLS(SystemTests): + @pytest.fixture(scope='class') + def grpc_channel(self): + system_test_utilities.write_grpc_device_server_config(use_tls_config=True) + system_test_utilities.exchange_certificates("localhost") + system_test_utilities.configure_tls_modes( + "ni-grpc-device-server", + "localhost", + "ManagedSelfSigned", + "ManagedSelfSigned", + "Managed", + "TrustedCertificates" + ) + + current_directory = os.path.dirname(os.path.abspath(__file__)) + config_file_path = os.path.join(current_directory, 'grpc_server_config.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): + grpc_options = nidmm.GrpcSessionOptions(grpc_channel, '') + return {'grpc_options': grpc_options} + + +class TestGrpcNoTLS(SystemTests): + @pytest.fixture(scope='class') + def grpc_channel(self): + system_test_utilities.write_grpc_device_server_config(use_tls_config=False) + current_directory = os.path.dirname(os.path.abspath(__file__)) config_file_path = os.path.join(current_directory, 'grpc_server_config.json') with system_test_utilities.GrpcServerProcess(config_file_path) as proc: diff --git a/src/shared/system_test_utilities.py b/src/shared/system_test_utilities.py index dea3b2d1cd..a6130223f3 100644 --- a/src/shared/system_test_utilities.py +++ b/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,103 @@ 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, +): + script_path = ( + r"C:\NITests\nitlsconfigtest\exchange_certificates.py" + if sys.platform == "win32" else + r"/opt/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) + subprocess.run(command, check=True) + + +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, +): + script_path = ( + r"C:\NITests\nitlsconfigtest\configure_tls_modes.py" + if sys.platform == "win32" else + r"/opt/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 + ) + subprocess.run(command, check=True) + + +def write_grpc_device_server_config(use_tls_config: bool = True): + config_path = ( + r"C:\Program Files\National Instruments\Shared\NI gRPC Device Server\server_config.json" + if sys.platform == "win32" else + r"/etc/ni_grpc_device_server/server_config.json" + ) + if not os.path.isfile(config_path): + raise FileNotFoundError(f"NI gRPC Device Server config file not found: {config_path}") + + config = { + "address": "[::]", + "port": 31763, + } + if use_tls_config: + config["security"] = "ni-tls-config" + config["feature_toggles"] = {"ni-tls-config": True} + + with open(config_path, "w", encoding="utf-8") as config_file: + json.dump(config, config_file, indent=4) + config_file.write("\n") \ No newline at end of file From e1f346b3bdd355a3776ddaf7bad590e8d9eb274e Mon Sep 17 00:00:00 2001 From: Ryan Wixon Date: Thu, 10 Sep 2026 11:17:11 -0500 Subject: [PATCH 2/3] Add sad path tests --- src/nidmm/system_tests/test_system_nidmm.py | 103 +++++++++++++++++--- src/shared/system_test_utilities.py | 6 +- 2 files changed, 94 insertions(+), 15 deletions(-) diff --git a/src/nidmm/system_tests/test_system_nidmm.py b/src/nidmm/system_tests/test_system_nidmm.py index 0a1d2781ec..d880977c4d 100644 --- a/src/nidmm/system_tests/test_system_nidmm.py +++ b/src/nidmm/system_tests/test_system_nidmm.py @@ -333,12 +333,12 @@ def grpc_channel(self): system_test_utilities.write_grpc_device_server_config(use_tls_config=True) system_test_utilities.exchange_certificates("localhost") system_test_utilities.configure_tls_modes( - "ni-grpc-device-server", - "localhost", - "Disabled", - "Disabled", - "Disabled", - "Disabled" + service="ni-grpc-device-server", + server_host="localhost", + server_cert_mode="ManagedSelfSigned", + server_client_mode="ManagedSelfSigned", + client_cert_mode="Managed", + client_server_mode="TrustedCertificates" ) current_directory = os.path.dirname(os.path.abspath(__file__)) @@ -351,6 +351,85 @@ def grpc_channel(self): def session_creation_kwargs(self, grpc_channel): grpc_options = nidmm.GrpcSessionOptions(grpc_channel, '') return {'grpc_options': grpc_options} + + def test_unsecured_client(self, grpc_channel): + system_test_utilities.configure_tls_modes( + service="ni-grpc-device-server", + server_host="localhost", + server_cert_mode="ManagedSelfSigned", + server_client_mode="ManagedSelfSigned", + client_cert_mode="Disabled", + client_server_mode="Disabled" + ) + + grpc_options = nidmm.GrpcSessionOptions(grpc_channel, '') + try: + with pytest.raises(nidmm.Error) as exc_info: + with nidmm.Session('FakeDevice', False, True, 'Simulate=1, DriverSetup=Model:4082; BoardType:PXIe', grpc_options=grpc_options): + pass + + assert exc_info.value.rpc_code == grpc.StatusCode.UNAVAILABLE + assert exc_info.value.description == 'Failed to connect to server' + finally: + # Restore the normal TLS configuration + system_test_utilities.configure_tls_modes( + service="ni-grpc-device-server", + server_host="localhost", + server_cert_mode="ManagedSelfSigned", + server_client_mode="ManagedSelfSigned", + client_cert_mode="Managed", + client_server_mode="TrustedCertificates" + ) + + def test_unsecured_server(self, grpc_channel): + system_test_utilities.configure_tls_modes( + service="ni-grpc-device-server", + server_host="localhost", + server_cert_mode="Disabled", + server_client_mode="Disabled", + client_cert_mode="Managed", + client_server_mode="TrustedCertificates" + ) + + grpc_options = nidmm.GrpcSessionOptions(grpc_channel, '') + try: + with pytest.raises(nidmm.Error) as exc_info: + with nidmm.Session('FakeDevice', False, True, 'Simulate=1, DriverSetup=Model:4082; BoardType:PXIe', grpc_options=grpc_options): + pass + + assert exc_info.value.rpc_code == grpc.StatusCode.UNAVAILABLE + assert exc_info.value.description == 'Failed to connect to server' + finally: + # Restore the normal TLS configuration + system_test_utilities.configure_tls_modes( + service="ni-grpc-device-server", + server_host="localhost", + server_cert_mode="ManagedSelfSigned", + server_client_mode="ManagedSelfSigned", + client_cert_mode="Managed", + client_server_mode="TrustedCertificates" + ) + + def test_no_certificates(self, grpc_channel): + trusted_client_folder = ( + r"C:/ProgramData/National Instruments/nitlsconfig/server.d/ni-grpc-device/trusted.d" + if sys.platform == "win32" else + r"/etc/nitlsconfig/server.d/ni-grpc-device/trusted.d" + ) + if os.path.exists(trusted_client_folder): + shutil.rmtree(trusted_client_folder) + + grpc_options = nidmm.GrpcSessionOptions(grpc_channel, '') + try: + with pytest.raises(nidmm.Error) as exc_info: + with nidmm.Session('FakeDevice', False, True, 'Simulate=1, DriverSetup=Model:4082; BoardType:PXIe', grpc_options=grpc_options): + pass + + assert exc_info.value.rpc_code == grpc.StatusCode.UNAVAILABLE + assert exc_info.value.description == 'Failed to connect to server' + finally: + # Reprovision to restore the deleted certificate + system_test_utilities.exchange_certificates("localhost") class TestGrpcUnsecuredTLS(SystemTests): @@ -359,12 +438,12 @@ def grpc_channel(self): system_test_utilities.write_grpc_device_server_config(use_tls_config=True) system_test_utilities.exchange_certificates("localhost") system_test_utilities.configure_tls_modes( - "ni-grpc-device-server", - "localhost", - "ManagedSelfSigned", - "ManagedSelfSigned", - "Managed", - "TrustedCertificates" + service="ni-grpc-device-server", + 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__)) diff --git a/src/shared/system_test_utilities.py b/src/shared/system_test_utilities.py index a6130223f3..5addeaa76b 100644 --- a/src/shared/system_test_utilities.py +++ b/src/shared/system_test_utilities.py @@ -116,7 +116,7 @@ def exchange_certificates( verbosity: int = 2, ): script_path = ( - r"C:\NITests\nitlsconfigtest\exchange_certificates.py" + r"C:/NITests/nitlsconfigtest/exchange_certificates.py" if sys.platform == "win32" else r"/opt/NITests/nitlsconfigtest/exchange_certificates.py" ) @@ -153,7 +153,7 @@ def configure_tls_modes( client_server_mode: str | None = None, ): script_path = ( - r"C:\NITests\nitlsconfigtest\configure_tls_modes.py" + r"C:/NITests/nitlsconfigtest/configure_tls_modes.py" if sys.platform == "win32" else r"/opt/NITests/nitlsconfigtest/configure_tls_modes.py" ) @@ -188,7 +188,7 @@ def configure_tls_modes( def write_grpc_device_server_config(use_tls_config: bool = True): config_path = ( - r"C:\Program Files\National Instruments\Shared\NI gRPC Device Server\server_config.json" + r"C:/Program Files/National Instruments/Shared/NI gRPC Device Server/server_config.json" if sys.platform == "win32" else r"/etc/ni_grpc_device_server/server_config.json" ) From 40140b9174f84af68de522d1a80ffa25805717e1 Mon Sep 17 00:00:00 2001 From: Ryan Wixon Date: Thu, 10 Sep 2026 11:31:12 -0500 Subject: [PATCH 3/3] Resolve flake errors --- src/nidmm/system_tests/test_system_nidmm.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/nidmm/system_tests/test_system_nidmm.py b/src/nidmm/system_tests/test_system_nidmm.py index d880977c4d..edc51d7fad 100644 --- a/src/nidmm/system_tests/test_system_nidmm.py +++ b/src/nidmm/system_tests/test_system_nidmm.py @@ -1,6 +1,7 @@ import math import os import pathlib +import shutil import sys import tempfile import time @@ -351,7 +352,7 @@ def grpc_channel(self): def session_creation_kwargs(self, grpc_channel): grpc_options = nidmm.GrpcSessionOptions(grpc_channel, '') return {'grpc_options': grpc_options} - + def test_unsecured_client(self, grpc_channel): system_test_utilities.configure_tls_modes( service="ni-grpc-device-server", @@ -409,7 +410,7 @@ def test_unsecured_server(self, grpc_channel): client_cert_mode="Managed", client_server_mode="TrustedCertificates" ) - + def test_no_certificates(self, grpc_channel): trusted_client_folder = ( r"C:/ProgramData/National Instruments/nitlsconfig/server.d/ni-grpc-device/trusted.d"