Enable the creation of mTLS encrypted gRPC channels leveraging NI TLS via nitlsconfig (Python)

.config/cspell/project-software-terms.txt

@@ -37,6 +37,7 @@ NISHAREDDIR       # NI Linux install directory environment variable
 nicai             # NI-DAQmx C library filename (Windows DLL)
 nicaiu            # NI-DAQmx C library filename (Windows DLL, Unicode)
 nidevice          # NI package name; ni-device-grpc
+nitlsconfig       # Dependency: nitlsconfig
 nitypes           # Dependency: nitypes
 nptdms            # Dependency: nptdms
 numpy             # Dependency: numpy

CHANGELOG.md

@@ -38,6 +38,7 @@ All notable changes to this project will be documented in this file.
 
 * ### Major Changes
     * Add support for Analog Path Delay
+    * Added `nitlsconfig[grpc]` to the `grpc` extra to enable using NI TLS to create TLS-encrypted gRPC channels. `nitlsconfig` is the Python API for reading the NI TLS configuration installed on the system and creating NI gRPC Device client channels from it.
 
 * ### Known Issues
     * ...

generated/nidaqmx/_grpc_interpreter.py

@@ -12,6 +12,7 @@
 
 import google.protobuf.message
 import grpc
+import nitlsconfig
 import numpy
 
 from . import errors as errors
@@ -101,14 +102,20 @@ class GrpcStubInterpreter(BaseInterpreter):
     def __init__(self, grpc_options):
         self._grpc_options = grpc_options
         self._client = nidaqmx_grpc.NiDAQmxStub(grpc_options.grpc_channel)
+        # Querying the driver version is the first contact with the server, so it decides
+        # whether this interpreter connected.
+        connected = False
         try:
             major_version = self.get_system_info_attribute_uint32(0x1272)
             minor_version = self.get_system_info_attribute_uint32(0x1923)
             update_version = self.get_system_info_attribute_uint32(0x2f22)
+            connected = True
         except Exception:
             major_version = 0
             minor_version = 0
             update_version = 0
+        finally:
+            nitlsconfig.audit_session_connect('NI-DAQmx', grpc_options.grpc_channel, connected)
         self._driver_version = DriverVersion(major_version, minor_version, update_version)
 
     def _invoke(self, func, request, metadata=None):
@@ -141,7 +148,11 @@ def _handle_rpc_error(self, rpc_error):
                     error_message += f'\nSamples per channel written: {entry.value}'
         grpc_error = rpc_error.code()
         if grpc_error == grpc.StatusCode.UNAVAILABLE:
-            error_message = 'Failed to connect to server'
+            # gRPC reports a rejected TLS handshake and an unreachable server with the
+            # same code, so ask nitlsconfig whether it built this channel and can say more.
+            error_message = nitlsconfig.get_tls_connection_error_elaboration(
+                self._grpc_options.grpc_channel
+            ) or 'Failed to connect to server'
         elif grpc_error == grpc.StatusCode.UNIMPLEMENTED:
             error_message = (
                 'This operation is not supported by the NI gRPC Device Server being used. Upgrade NI gRPC Device Server.'

poetry.toml

@@ -9,5 +9,6 @@ min-release-age-exclude = [
     "ni-grpcdevice-v1-proto",
     "ni-protobuf-types",
     "ni-python-styleguide",
+    "nitlsconfig",
     "nitypes",
 ]

pyproject.toml

@@ -43,7 +43,7 @@ documentation = "https://nidaqmx-python.readthedocs.io"
 nidaqmx = 'nidaqmx.__main__:main'
 
 [project.optional-dependencies]
-grpc = ['grpcio (>=1.49.0,<2.0)', 'protobuf (>=4.21)', 'ni-grpcdevice-v1-proto (>=1.0.0)', 'ni-protobuf-types (>=1.0.0)']
+grpc = ['grpcio (>=1.49.0,<2.0)', 'protobuf (>=4.21)', 'ni-grpcdevice-v1-proto (>=1.0.0)', 'ni-protobuf-types (>=1.0.0)', 'nitlsconfig[grpc] (>=1.0.0a4)']
 
 [tool.poetry]
 

src/codegen/templates/_grpc_interpreter.py.mako

@@ -29,6 +29,7 @@ from collections.abc import Callable, Sequence
 
 import google.protobuf.message
 import grpc
+import nitlsconfig
 import numpy
 
 from . import errors as errors
@@ -118,14 +119,20 @@ class GrpcStubInterpreter(BaseInterpreter):
     def __init__(self, grpc_options):
         self._grpc_options = grpc_options
         self._client = nidaqmx_grpc.NiDAQmxStub(grpc_options.grpc_channel)
+        # Querying the driver version is the first contact with the server, so it decides
+        # whether this interpreter connected.
+        connected = False
         try:
             major_version = self.get_system_info_attribute_uint32(0x1272)
             minor_version = self.get_system_info_attribute_uint32(0x1923)
             update_version = self.get_system_info_attribute_uint32(0x2f22)
+            connected = True
         except Exception:
             major_version = 0
             minor_version = 0
             update_version = 0
+        finally:
+            nitlsconfig.audit_session_connect('NI-DAQmx', grpc_options.grpc_channel, connected)
         self._driver_version = DriverVersion(major_version, minor_version, update_version)
 
     def _invoke(self, func, request, metadata=None):
@@ -158,7 +165,11 @@ class GrpcStubInterpreter(BaseInterpreter):
                     error_message += f'\nSamples per channel written: {entry.value}'
         grpc_error = rpc_error.code()
         if grpc_error == grpc.StatusCode.UNAVAILABLE:
-            error_message = 'Failed to connect to server'
+            # gRPC reports a rejected TLS handshake and an unreachable server with the
+            # same code, so ask nitlsconfig whether it built this channel and can say more.
+            error_message = nitlsconfig.get_tls_connection_error_elaboration(
+                self._grpc_options.grpc_channel
+            ) or 'Failed to connect to server'
         elif grpc_error == grpc.StatusCode.UNIMPLEMENTED:
             error_message = (
                 'This operation is not supported by the NI gRPC Device Server being used. Upgrade NI gRPC Device Server.'

tests/unit/test_nitlsconfig_grpc_integration.py

@@ -0,0 +1,133 @@
+from __future__ import annotations
+
+from unittest.mock import Mock
+
+import pytest
+from pytest_mock import MockerFixture
+
+import nidaqmx
+from nidaqmx import Task
+from tests.unit._grpc_utils import create_grpc_options
+from tests.unit._task_utils import expect_create_task, expect_get_task_name
+
+# _FakeRpcError subclasses grpc.RpcError, so it must be defined here or importing this
+# module raises NameError when the grpc extra isn't installed.
+try:
+    import grpc
+    import nitlsconfig
+
+    from nidaqmx._grpc_interpreter import GrpcStubInterpreter
+
+    class _FakeRpcError(grpc.RpcError):
+        def __init__(self, code):
+            self._code = code
+
+        def code(self):
+            return self._code
+
+        def details(self):
+            return "original details"
+
+        def trailing_metadata(self):
+            return []
+
+except ImportError:
+    grpc = None  # type: ignore
+
+
+@pytest.fixture
+def nitlsconfig_tagged_channel():
+    """A gRPC channel tagged the way nitlsconfig.create_grpc_device_channel tags one.
+
+    create_grpc_device_channel needs the nitlsconfig CLI installed on the system, so tagging a
+    plain channel is the closest we can get to one here. Nothing connects over it.
+    """
+    if grpc is None:
+        pytest.skip("The grpc module is not available.")
+    target = "localhost:31763"
+    with grpc.insecure_channel(target) as channel:
+        nitlsconfig.channel_tag.tag_channel_target(channel, target)
+        yield channel
+
+
+def _create_interpreter(mocker: MockerFixture, grpc_options, version_error=None):
+    mocker.patch("nidaqmx._grpc_interpreter.nidaqmx_grpc.NiDAQmxStub", autospec=True)
+    mocker.patch.object(
+        GrpcStubInterpreter,
+        "get_system_info_attribute_uint32",
+        side_effect=version_error,
+        return_value=None if version_error else 1,
+    )
+    return GrpcStubInterpreter(grpc_options)
+
+
+def test___untagged_channel___handle_unavailable___raises_failed_to_connect(
+    mocker: MockerFixture,
+):
+    interpreter = _create_interpreter(mocker, create_grpc_options(mocker))
+
+    with pytest.raises(nidaqmx.errors.RpcError) as exc_info:
+        interpreter._handle_rpc_error(_FakeRpcError(grpc.StatusCode.UNAVAILABLE))
+
+    assert exc_info.value.rpc_code == grpc.StatusCode.UNAVAILABLE
+    assert exc_info.value.description == "Failed to connect to server"
+
+
+def test___nitlsconfig_tagged_channel___handle_unavailable___raises_tls_elaboration(
+    mocker: MockerFixture, nitlsconfig_tagged_channel
+):
+    # Derived from nitlsconfig itself, so this fails if it stops recognizing our channel.
+    expected_message = nitlsconfig.get_tls_connection_error_elaboration(nitlsconfig_tagged_channel)
+    assert expected_message is not None
+    assert expected_message != "Failed to connect to server"
+    grpc_options = nidaqmx.GrpcSessionOptions(nitlsconfig_tagged_channel, "")
+    interpreter = _create_interpreter(mocker, grpc_options)
+
+    with pytest.raises(nidaqmx.errors.RpcError) as exc_info:
+        interpreter._handle_rpc_error(_FakeRpcError(grpc.StatusCode.UNAVAILABLE))
+
+    assert exc_info.value.rpc_code == grpc.StatusCode.UNAVAILABLE
+    assert exc_info.value.description == expected_message
+
+
+def test___nitlsconfig_tagged_channel___handle_other_status_code___preserves_original_details(
+    mocker: MockerFixture, nitlsconfig_tagged_channel
+):
+    # Tagged, so an elaboration is available: this fails if we stop limiting it to UNAVAILABLE.
+    grpc_options = nidaqmx.GrpcSessionOptions(nitlsconfig_tagged_channel, "")
+    interpreter = _create_interpreter(mocker, grpc_options)
+
+    with pytest.raises(nidaqmx.errors.RpcError) as exc_info:
+        interpreter._handle_rpc_error(_FakeRpcError(grpc.StatusCode.INTERNAL))
+
+    assert exc_info.value.rpc_code == grpc.StatusCode.INTERNAL
+    assert exc_info.value.description == "original details"
+
+
+def test___server_reachable___create_interpreter___audits_connected(mocker: MockerFixture):
+    patched_audit = mocker.patch("nitlsconfig.audit_session_connect", autospec=True)
+    grpc_options = create_grpc_options(mocker)
+
+    _create_interpreter(mocker, grpc_options)
+
+    patched_audit.assert_called_once_with("NI-DAQmx", grpc_options.grpc_channel, True)
+
+
+def test___server_unreachable___create_interpreter___audits_not_connected(mocker: MockerFixture):
+    patched_audit = mocker.patch("nitlsconfig.audit_session_connect", autospec=True)
+    grpc_options = create_grpc_options(mocker)
+
+    _create_interpreter(mocker, grpc_options, version_error=Exception("unreachable"))
+
+    patched_audit.assert_called_once_with("NI-DAQmx", grpc_options.grpc_channel, False)
+
+
+def test___no_grpc_options___create_task___does_not_audit(interpreter: Mock, mocker: MockerFixture):
+    patched_audit = mocker.patch("nitlsconfig.audit_session_connect", autospec=True)
+    expect_create_task(interpreter)
+    expect_get_task_name(interpreter, "MyTask")
+
+    with Task("MyTask"):
+        pass
+
+    patched_audit.assert_not_called()