Add get_tls_connection_error_elaboration to allow clients to get more descriptive errors

src/nitlsconfig/__init__.py

@@ -36,6 +36,7 @@
     NitlsconfigCliError,
     NitlsconfigError,
     TlsConfigurationError,
+    get_tls_connection_error_elaboration,
 )
 
 if TYPE_CHECKING:
@@ -75,6 +76,7 @@
     "TlsConfigurationError",
     "TrustedCertificateData",
     "KnownServerData",
+    "get_tls_connection_error_elaboration",
 ]
 
 # The gRPC names are public API, but only on an install that can supply them.

src/nitlsconfig/audit.py

@@ -41,6 +41,7 @@
 import threading
 from enum import Enum
 
+from nitlsconfig.channel_tag import get_channel_target
 from nitlsconfig.service import SERVICE_NAME
 
 _ROLE = "Client"
@@ -60,11 +61,6 @@ class TransportSecurity(Enum):
 _logging_handler_lock = threading.Lock()
 _logging_handler_attached = False
 
-# Set on channels we create. Carries the address audit_session_connect reports, and
-# marks the channel as ours: the API layer issuing the initialize RPC holds only the
-# channel, and cannot otherwise tell whether NI-TLS had any part in building it.
-_TARGET_ATTR = "_nitls_audit_target"
-
 _MAX_FIELD_LENGTH = 256
 
 # The quote is here because messages wrap every field in single quotes; without it
@@ -97,17 +93,6 @@ def _audit_field(value: object) -> str:
     return truncated + "..."
 
 
-def tag_channel_target(channel: object, target: str) -> None:
-    """Record on a channel the ``host:port`` it was created for. Never raises."""
-    try:
-        setattr(channel, _TARGET_ATTR, target)
-    except Exception:
-        logging.getLogger(__name__).debug(
-            "Unable to tag channel for audit logging; the channel will go unaudited.",
-            exc_info=True,
-        )
-
-
 def _make_logging_handler() -> logging.Handler:
     """Create the platform audit logging handler.
 
@@ -203,7 +188,7 @@ def audit_session_connect(driver_name: str, channel: object, connected: bool) ->
     source on machines not using NI-TLS at all.
     """
     try:
-        target = getattr(channel, _TARGET_ATTR, "")
+        target = get_channel_target(channel)
         if not target:
             return
 

src/nitlsconfig/channel_tag.py

@@ -0,0 +1,39 @@
+"""Marks gRPC channels this package created.
+
+A caller holding only a channel cannot tell whether NI-TLS had any part in
+building it, so the channel factory tags what it creates. Two features read the
+tag: audit records name the address, and connection-error elaboration speaks only
+for channels we built.
+
+The tag is advisory, not a security control. It says where a channel came from,
+never that a connection is trustworthy.
+"""
+
+from __future__ import annotations
+
+import logging
+
+_TARGET_ATTR = "_nitls_channel_target"
+
+
+def tag_channel_target(channel: object, target: str) -> None:
+    """Record on a channel the ``host:port`` it was created for. Never raises."""
+    try:
+        setattr(channel, _TARGET_ATTR, target)
+    except Exception:
+        logging.getLogger(__name__).debug(
+            "Unable to tag channel; it will go unaudited and will not elaborate on "
+            "connection errors.",
+            exc_info=True,
+        )
+
+
+def get_channel_target(channel: object) -> str:
+    """Return the ``host:port`` a channel was created for, or empty if we did not create it."""
+    target = getattr(channel, _TARGET_ATTR, "")
+    return target if isinstance(target, str) else ""
+
+
+def is_nitls_channel(channel: object) -> bool:
+    """Return whether this package created the channel."""
+    return bool(get_channel_target(channel))

src/nitlsconfig/errors.py

@@ -1,7 +1,11 @@
-"""Exceptions raised by this package."""
+"""Exceptions raised by this package, and the text that explains them."""
 
 from __future__ import annotations
 
+from typing import Optional
+
+from nitlsconfig.channel_tag import is_nitls_channel
+
 
 class NitlsconfigError(RuntimeError):
     """Base error for every failure raised by this package."""
@@ -42,3 +46,36 @@ class TlsConfigurationError(NitlsconfigError):
         "are configured and matching on both the host and remote target. Check that the remote "
         "target has a compatible TLS enabled configuration with the host."
     )
+
+
+_TLS_CONNECTION_ERROR_ELABORATION = (
+    "Connection to the remote system failed.\n\n"
+    "Check that the remote target is reachable. Verify in NI Hardware Manager that "
+    "the host has a compatible TLS enabled configuration with the remote target."
+)
+
+
+def get_tls_connection_error_elaboration(channel: object) -> Optional[str]:
+    """Return text elaborating on a connection failure, or None for a foreign channel.
+
+    gRPC reports a rejected TLS handshake as ``UNAVAILABLE``, the same code it
+    uses for an unreachable server, so a driver API catching that error cannot
+    tell the two apart and reports only that connecting failed. Pass the channel
+    here to recover the missing half of the diagnosis.
+
+    Text is returned for any channel
+    :func:`~nitlsconfig.grpc_channel.create_grpc_device_channel` built, including
+    one created unencrypted because ``server_mode`` is Disabled: a client
+    configured for plaintext against a TLS-enabled target fails to connect for
+    exactly the reason the text describes.
+
+    Substitute it for an existing message that only says connecting failed,
+    which it restates; append it to one that carries detail
+    of its own, such as the RPC that failed.
+
+    Never raises, so it is safe to call from an exception handler.
+    """
+    try:
+        return _TLS_CONNECTION_ERROR_ELABORATION if is_nitls_channel(channel) else None
+    except Exception:  # pragma: no cover - getattr on a hostile channel
+        return None

src/nitlsconfig/grpc_channel.py

@@ -22,6 +22,10 @@
 Other services can still be read through :class:`~nitlsconfig.cli.ClientConfig`;
 a configurable service name can be added later without breaking this signature.
 
+:func:`~nitlsconfig.errors.get_tls_connection_error_elaboration` recognizes a
+channel built here, so a driver API can tell the caller that a failure to connect
+may be TLS-related, which gRPC's status codes cannot express on their own.
+
 ``server_mode`` Disabled selects a plain connection. Every other mode is treated
 exactly like ``TrustedCertificates``: the server certificate chain is verified
 *and* the hostname is checked. gRPC's Python API cannot relax either check
@@ -47,8 +51,8 @@
 from nitlsconfig.audit import (
     TransportSecurity,
     audit_transport_posture,
-    tag_channel_target,
 )
+from nitlsconfig.channel_tag import tag_channel_target
 from nitlsconfig.cli import (
     ClientCertMode,
     ClientConfig,

tests/unit/test_audit.py

@@ -11,6 +11,7 @@
     audit_session_connect,
     audit_transport_posture,
 )
+from nitlsconfig.channel_tag import tag_channel_target
 
 SERVICE = "ni-grpc-device"
 HOST = "localhost"
@@ -169,7 +170,7 @@ class FakeChannel:
 
 def test___session_connected___logs_connect_as_info(recorded: RecordingHandler) -> None:
     channel = FakeChannel()
-    audit.tag_channel_target(channel, "localhost:31763")
+    tag_channel_target(channel, "localhost:31763")
 
     audit_session_connect("NI-DCPower", channel, True)
 
@@ -181,7 +182,7 @@ def test___session_connected___logs_connect_as_info(recorded: RecordingHandler)
 
 def test___session_not_connected___logs_failure_as_error(recorded: RecordingHandler) -> None:
     channel = FakeChannel()
-    audit.tag_channel_target(channel, "localhost:31763")
+    tag_channel_target(channel, "localhost:31763")
 
     audit_session_connect("NI-DCPower", channel, False)
 
@@ -195,7 +196,7 @@ def test___session_fields_with_newlines___escape_record_breaking_characters(
     recorded: RecordingHandler,
 ) -> None:
     channel = FakeChannel()
-    audit.tag_channel_target(channel, "localhost\r\nforged-entry")
+    tag_channel_target(channel, "localhost\r\nforged-entry")
 
     audit_session_connect("driver\rname", channel, True)
 
@@ -211,13 +212,6 @@ def test___untagged_channel___emits_no_record(recorded: RecordingHandler) -> Non
     assert recorded.records == []
 
 
-def test___channel_rejects_attributes___tagging_does_not_raise() -> None:
-    class Slotted:
-        __slots__ = ()
-
-    audit.tag_channel_target(Slotted(), "localhost:31763")
-
-
 def test___audit_logger___formats_with_service_and_role_prefix(
     recorded: RecordingHandler,
 ) -> None:
@@ -267,7 +261,7 @@ def handleError(self, record: logging.LogRecord) -> None:  # noqa: N802 - loggin
     reset_logger()
 
     channel = FakeChannel()
-    audit.tag_channel_target(channel, "localhost:31763")
+    tag_channel_target(channel, "localhost:31763")
 
     audit_transport_posture(HOST, TransportSecurity.MutualTls)
     audit_session_connect("NI-DCPower", channel, False)

tests/unit/test_channel_tag.py

@@ -0,0 +1,34 @@
+"Pytests for nitlsconfig.channel_tag."
+
+from nitlsconfig.channel_tag import get_channel_target, is_nitls_channel, tag_channel_target
+
+
+class FakeChannel:
+    """Stands in for a grpc.Channel, which is only an attribute holder here."""
+
+
+def test___tagged_channel___reports_its_target() -> None:
+    channel = FakeChannel()
+    tag_channel_target(channel, "localhost:31763")
+
+    assert get_channel_target(channel) == "localhost:31763"
+    assert is_nitls_channel(channel)
+
+
+def test___untagged_channel___is_not_recognized_as_ours() -> None:
+    assert get_channel_target(FakeChannel()) == ""
+    assert not is_nitls_channel(FakeChannel())
+
+
+def test___channel_carrying_a_foreign_attribute___is_not_recognized_as_ours() -> None:
+    channel = FakeChannel()
+    setattr(channel, "_nitls_channel_target", object())
+
+    assert not is_nitls_channel(channel)
+
+
+def test___channel_rejects_attributes___tagging_does_not_raise() -> None:
+    class Slotted:
+        __slots__ = ()
+
+    tag_channel_target(Slotted(), "localhost:31763")

tests/unit/test_errors.py

@@ -0,0 +1,22 @@
+"""Pytests for nitlsconfig.errors."""
+
+from nitlsconfig.channel_tag import tag_channel_target
+from nitlsconfig.errors import get_tls_connection_error_elaboration
+
+
+class FakeChannel:
+    """Stands in for a grpc.Channel, which the elaboration never calls into."""
+
+
+def test___channel_we_created___elaborates_on_connection_errors() -> None:
+    channel = FakeChannel()
+    tag_channel_target(channel, "localhost:31763")
+
+    elaboration = get_tls_connection_error_elaboration(channel)
+
+    assert elaboration is not None
+    assert "NI Hardware Manager" in elaboration
+
+
+def test___channel_we_did_not_create___does_not_elaborate() -> None:
+    assert get_tls_connection_error_elaboration(object()) is None

tests/unit/test_grpc_channel.py

@@ -14,6 +14,7 @@
 )
 
 from nitlsconfig import grpc_channel
+from nitlsconfig.channel_tag import is_nitls_channel
 from nitlsconfig.cli import (
     CertificateLocation,
     ClientCertMode,
@@ -177,6 +178,26 @@ def test_mutual_tls_credentials(
     }
 
 
+def test_created_channels_are_tagged_as_ours(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # Tagging is what lets a driver API elaborate on a later connection failure.
+    # Plaintext counts too: a client expecting plaintext cannot reach a TLS target.
+    tls = create_channel(
+        monkeypatch,
+        FakeClientConfig(
+            server_mode=ClientServerMode.TrustedCertificates,
+            certificate_mode=ClientCertMode.Disabled,
+            trusted_certificates_location=FILE_TRUST,
+            trusted_certificates_contents="ROOT",
+        ),
+    )
+    plaintext = create_channel(monkeypatch, FakeClientConfig())
+
+    assert is_nitls_channel(tls)
+    assert is_nitls_channel(plaintext)
+
+
 def test_one_way_tls_credentials(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None: