diff --git a/README.md b/README.md index f8418ac..89d4e0d 100644 --- a/README.md +++ b/README.md @@ -50,8 +50,11 @@ channel = nitlsconfig.create_grpc_client_channel( ``` `TlsConfigurationError` is raised when TLS is enabled but the configuration is -unusable. Accessing any of these names without the `grpc` extra installed raises -`ImportError` telling you which extra to install. +unusable. It is always importable, since handling it does not require grpcio. + +`create_grpc_client_channel` and `RetryPolicy` do require grpcio; accessing them +without the `grpc` extra installed raises `ImportError` telling you which extra +to install. ## Reading configurations ```python diff --git a/src/nitlsconfig/__init__.py b/src/nitlsconfig/__init__.py index 7a1e3c9..93a5461 100644 --- a/src/nitlsconfig/__init__.py +++ b/src/nitlsconfig/__init__.py @@ -21,24 +21,28 @@ ClientCertMode, ClientConfig, ClientServerMode, - CommandFailedError, - ExecutableNotFoundError, - InvalidOutputError, LocationScheme, - NitlsconfigCliError, ServerCertMode, ServerClientMode, ServerConfig, TrustedCertificateData, KnownServerData, ) +from nitlsconfig.errors import ( + CommandFailedError, + CommandTimeoutError, + ExecutableNotFoundError, + InvalidOutputError, + NitlsconfigCliError, + NitlsconfigError, + TlsConfigurationError, +) if TYPE_CHECKING: # Imported eagerly for type checkers and editors, which do not run __getattr__. from nitlsconfig.grpc_channel import ( DEFAULT_SERVICE_NAME, RetryPolicy, - TlsConfigurationError, create_grpc_client_channel, ) @@ -50,7 +54,6 @@ _GRPC_EXPORTS = [ "DEFAULT_SERVICE_NAME", "RetryPolicy", - "TlsConfigurationError", "create_grpc_client_channel", ] @@ -65,10 +68,13 @@ "ServerCertMode", "ServerClientMode", "ServerConfig", + "NitlsconfigError", "NitlsconfigCliError", "ExecutableNotFoundError", "CommandFailedError", + "CommandTimeoutError", "InvalidOutputError", + "TlsConfigurationError", "TrustedCertificateData", "KnownServerData", ] diff --git a/src/nitlsconfig/cli.py b/src/nitlsconfig/cli.py index b54f885..a5363f4 100644 --- a/src/nitlsconfig/cli.py +++ b/src/nitlsconfig/cli.py @@ -10,6 +10,14 @@ from enum import Enum from typing import Any, Optional, Tuple, TypeVar +from nitlsconfig.errors import ( + CommandFailedError, + CommandTimeoutError, + ExecutableNotFoundError, + InvalidOutputError, + NitlsconfigCliError, +) + ALLOWED_SCOPES: Tuple[str, ...] = ("client", "server") # Expected JSON root keys from nitlsconfig output. @@ -81,22 +89,6 @@ ) -class NitlsconfigCliError(RuntimeError): - """Base error for nitlsconfig command invocation failures.""" - - -class ExecutableNotFoundError(NitlsconfigCliError): - """Raised when a usable nitlsconfig executable cannot be found.""" - - -class CommandFailedError(NitlsconfigCliError): - """Raised when nitlsconfig exits with a non-zero return code.""" - - -class InvalidOutputError(NitlsconfigCliError): - """Raised when command output cannot be parsed as expected.""" - - class ServerCertMode(str, Enum): "Server TLS certificate mode." @@ -319,6 +311,7 @@ def run_nitlsconfig_command( """ executable = "nitlsconfig" argv = [executable, *command_args] + timeout_seconds = 30 try: completed = subprocess.run( @@ -326,11 +319,16 @@ def run_nitlsconfig_command( capture_output=True, text=True, check=False, - timeout=30, + timeout=timeout_seconds, ) # nosec B603 - argv is passed shell-free and executable selection is controlled except FileNotFoundError as ex: raise ExecutableNotFoundError( - "Unable to find nitlsconfig executable. " f"Tried {executable!r}." + f"Could not find an installation of {executable}. Please ensure that {executable} " + "is installed on this machine or contact National Instruments for support." + ) from ex + except subprocess.TimeoutExpired as ex: + raise CommandTimeoutError( + f"nitlsconfig command timed out after {timeout_seconds} seconds: {' '.join(argv)}." ) from ex if completed.returncode != 0: diff --git a/src/nitlsconfig/errors.py b/src/nitlsconfig/errors.py new file mode 100644 index 0000000..dc17b53 --- /dev/null +++ b/src/nitlsconfig/errors.py @@ -0,0 +1,44 @@ +"""Exceptions raised by this package.""" + +from __future__ import annotations + + +class NitlsconfigError(RuntimeError): + """Base error for every failure raised by this package.""" + + +class NitlsconfigCliError(NitlsconfigError): + """Base error for nitlsconfig command invocation failures.""" + + +class ExecutableNotFoundError(NitlsconfigCliError): + """Raised when a usable nitlsconfig executable cannot be found.""" + + +class CommandFailedError(NitlsconfigCliError): + """Raised when nitlsconfig exits with a non-zero return code.""" + + +class CommandTimeoutError(NitlsconfigCliError): + """Raised when nitlsconfig does not exit within the allotted time.""" + + +class InvalidOutputError(NitlsconfigCliError): + """Raised when command output cannot be parsed as expected.""" + + +class TlsConfigurationError(NitlsconfigError): + """Raised when the NI-TLS configuration was read successfully but is invalid. + + Deliberately not a :class:`NitlsconfigCliError`: the CLI worked, and the fix + is to provision or try again to provision this machine rather than to install + or repair installation. + """ + + #: Static message text, matching the wording used elsewhere in the product. Call + #: sites append the specific detail after it, as the C++ loader does. + _MESSAGE = ( + "A TLS configuration error occurred. Use NI Hardware Manager to verify that certificates " + "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." + ) diff --git a/src/nitlsconfig/grpc_channel.py b/src/nitlsconfig/grpc_channel.py index 98a3b18..90d47b6 100644 --- a/src/nitlsconfig/grpc_channel.py +++ b/src/nitlsconfig/grpc_channel.py @@ -21,17 +21,14 @@ The name carries the ``grpc`` prefix because this package also re-exports the factory from its root, alongside any potential future non-gRPC transports. -A client ``server_mode`` of ``TrustAlways`` is not currently supported and raises -:class:`TlsConfigurationError`. - -A client ``server_mode`` of ``SkipHostnameValidation`` is treated exactly like -``TrustedCertificates``: the server certificate chain is verified *and* the -hostname is checked. gRPC's Python API exposes no way to skip only the hostname -check: doing so requires a custom certificate verifier, which grpcio does not -bind in Python, where the TLS surface is limited to -``grpc.ssl_channel_credentials``. Verifying when asked not to fails closed, so -this is safe, but a caller who sets the mode gets no relaxation of the hostname -check. +``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 +independently, since that requires a custom certificate verifier which grpcio +does not bind in Python, where the TLS surface is limited to +``grpc.ssl_channel_credentials``. Verifying when asked not to fails closed, so a +caller who sets ``TrustAlways`` or ``SkipHostnameValidation`` gets a stricter +connection than requested rather than a weaker one. When the server certificate's CN/SAN does not match the dialed host, pass ``grpc.ssl_target_name_override`` via ``options`` instead. That substitutes the @@ -52,13 +49,12 @@ tag_channel_target, ) from nitlsconfig.cli import ( - CertificateLocation, ClientCertMode, ClientConfig, ClientServerMode, LocationScheme, - NitlsconfigCliError, ) +from nitlsconfig.errors import TlsConfigurationError __all__ = [ "DEFAULT_SERVICE_NAME", @@ -91,19 +87,6 @@ def _format_target(server_address: str, server_port: int) -> str: return f"{server_address}:{server_port}" -class TlsConfigurationError(NitlsconfigCliError): - """Raised when the NI-TLS configuration was read successfully but is invalid.""" - - #: Shared remedy text appended to messages whose fix is to provision - #: certificates. Kept in one place so the guidance stays consistent with the - #: wording used elsewhere in the product. - _REMEDY = ( - "Use NI Hardware Manager to verify that certificates 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." - ) - - @dataclass(frozen=True) class RetryPolicy: """Client retry behavior, realized as a gRPC service config. @@ -201,39 +184,19 @@ def _apply_retry_policy( return channel_options -def _require_file_scheme( - location: CertificateLocation, description: str, service_name: str -) -> None: - """Validate that a certificate or key location is a usable File:// path. - - The ni-grpc-device client capabilities declare support for the File scheme - only, so any other scheme is rejected rather than silently ignored. - """ - if location.scheme != LocationScheme.File: - raise TlsConfigurationError( - f"Client {description} must use the File scheme for service " - f"{service_name!r}, got {location.scheme.value!r}." - ) - if not location.path: - raise TlsConfigurationError( - f"TLS is enabled but the client {description} path is missing for " - f"service {service_name!r}." - ) - - def _require_contents(contents: str, description: str, service_name: str) -> str: """Validate that configured certificate material is actually present. Empty contents mean the material could not be produced (missing, unreadable, - or not yet provisioned), never that the client opted out. Opting out is - expressed by the configuration itself: ``certificate_mode`` Disabled for the - client identity, and the SystemDefault scheme for trust anchors. Neither - reaches this check. + not yet provisioned, or named by a scheme nitlsconfig cannot resolve), never + that the client opted out. Opting out is expressed by the configuration + itself: ``certificate_mode`` Disabled for the client identity, and the + SystemDefault scheme for trust anchors. Neither reaches this check. """ if not contents: raise TlsConfigurationError( - f"TLS is configured for service {service_name!r} but the client " - f"{description} is missing on this system. {TlsConfigurationError._REMEDY}" + f"{TlsConfigurationError._MESSAGE} The client {description} is missing on this " + f"system for service {service_name!r}." ) return contents @@ -246,30 +209,20 @@ def _load_client_tls_settings(config: ClientConfig) -> Optional[_ClientTlsSettin """ service_name = config.service_name - server_mode = config.server_mode - if server_mode == ClientServerMode.Disabled: + # Disabled is the only mode that turns TLS off. TrustAlways, SkipHostnameValidation, + # and Unknown all fall through to full chain and hostname verification: relaxing + # either check is a policy decision for NI-TLS, so this defaults to secure. + if config.server_mode == ClientServerMode.Disabled: return None - if server_mode == ClientServerMode.TrustAlways: - # ni-grpc-device.client.caps.yml declares supports_server_mode_trust_always: false, - # so this mode is not offered for this service. - raise TlsConfigurationError( - f"Client server_mode TrustAlways is not supported for service {service_name!r}." - ) - if server_mode == ClientServerMode.Unknown: - raise TlsConfigurationError(f"Unsupported client server_mode for service {service_name!r}.") - # certificate_mode only decides mTLS versus one-way TLS. - if config.certificate_mode == ClientCertMode.Unknown: - raise TlsConfigurationError( - f"Unsupported client certificate_mode for service {service_name!r}." - ) + # certificate_mode only decides mTLS versus one-way TLS. Disabled is the only mode + # that skips the client certificate, so Unknown presents one and fails closed if + # the material is not provisioned. present_client_cert = config.certificate_mode != ClientCertMode.Disabled certificate_chain_contents = "" private_key_contents = "" if present_client_cert: - _require_file_scheme(config.certificate_chain_location, "certificate chain", service_name) - _require_file_scheme(config.certificate_key_location, "certificate key", service_name) certificate_chain_contents = _require_contents( config.certificate_chain_contents, "certificate chain", service_name ) @@ -279,20 +232,13 @@ def _load_client_tls_settings(config: ClientConfig) -> Optional[_ClientTlsSettin # Trust anchors are always required: the client must verify the server. # SystemDefault means "use the platform certificate store" and carries no - # contents. Every other usable scheme (File, Directory) is resolved by - # nitlsconfig into a single PEM bundle, so the scheme itself does not need - # to be special-cased here; only Unknown is rejected. + # contents. Every other scheme names specific anchors that nitlsconfig + # resolves into a single PEM bundle, so the contents check below covers an + # unrecognized scheme without rejecting one this version has yet to learn. trusted_location = config.trusted_certificates_location - if trusted_location.scheme == LocationScheme.Unknown: - raise TlsConfigurationError( - f"TLS is configured for service {service_name!r} but the client trusted " - f"certificates location is missing or unrecognized. {TlsConfigurationError._REMEDY}" - ) trusted_contents = "" if trusted_location.scheme != LocationScheme.SystemDefault: - # Any scheme other than SystemDefault names specific anchors, so they - # must actually be present. trusted_contents = _require_contents( config.trusted_certificates_contents, "trusted certificate bundle", service_name ) diff --git a/tests/unit/test_grpc_channel.py b/tests/unit/test_grpc_channel.py index a61f3d7..2f51798 100644 --- a/tests/unit/test_grpc_channel.py +++ b/tests/unit/test_grpc_channel.py @@ -213,11 +213,19 @@ def test_directory_trust_anchors_are_supported( assert channel.credentials["root_certificates"] == b"ROOT_A\nROOT_B" -def test_skip_hostname_validation_matches_trusted_certificates( - monkeypatch: pytest.MonkeyPatch, +@pytest.mark.parametrize( + "server_mode", + [ + ClientServerMode.SkipHostnameValidation, + ClientServerMode.TrustAlways, + ClientServerMode.Unknown, + ], +) +def test_non_disabled_server_modes_match_trusted_certificates( + monkeypatch: pytest.MonkeyPatch, server_mode: ClientServerMode ) -> None: - # grpc's Python API cannot skip only the hostname check, so this mode is - # deliberately treated as TrustedCertificates. + # grpc's Python API cannot relax verification, so every mode that is not + # Disabled fails closed as TrustedCertificates. def config(server_mode: ClientServerMode) -> FakeClientConfig: return FakeClientConfig( server_mode=server_mode, @@ -227,24 +235,18 @@ def config(server_mode: ClientServerMode) -> FakeClientConfig: ) strict = create_channel(monkeypatch, config(ClientServerMode.TrustedCertificates)) - skipped = create_channel(monkeypatch, config(ClientServerMode.SkipHostnameValidation)) + relaxed = create_channel(monkeypatch, config(server_mode)) - assert skipped.secure - assert skipped.credentials == strict.credentials - assert skipped.options == strict.options + assert relaxed.secure + assert relaxed.credentials == strict.credentials + assert relaxed.options == strict.options @pytest.mark.parametrize( "config", [ - pytest.param( - FakeClientConfig(server_mode=ClientServerMode.TrustAlways), - id="trust_always_unsupported", - ), - pytest.param( - FakeClientConfig(server_mode=ClientServerMode.Unknown), - id="unknown_server_mode", - ), + # Unknown presents a client certificate, so it fails on the unprovisioned + # material rather than on the mode itself. pytest.param( FakeClientConfig( server_mode=ClientServerMode.TrustedCertificates, @@ -253,26 +255,6 @@ def config(server_mode: ClientServerMode) -> FakeClientConfig: ), id="unknown_certificate_mode", ), - pytest.param( - FakeClientConfig( - server_mode=ClientServerMode.TrustedCertificates, - certificate_mode=ClientCertMode.Managed, - certificate_chain_location=CertificateLocation(LocationScheme.Directory, "d"), - certificate_key_location=FILE_KEY, - trusted_certificates_location=FILE_TRUST, - ), - id="non_file_certificate_scheme", - ), - pytest.param( - FakeClientConfig( - server_mode=ClientServerMode.TrustedCertificates, - certificate_mode=ClientCertMode.Managed, - certificate_chain_location=CertificateLocation(LocationScheme.File), - certificate_key_location=FILE_KEY, - trusted_certificates_location=FILE_TRUST, - ), - id="empty_certificate_path", - ), pytest.param( FakeClientConfig( server_mode=ClientServerMode.TrustedCertificates, @@ -281,14 +263,6 @@ def config(server_mode: ClientServerMode) -> FakeClientConfig: ), id="missing_trust_anchors", ), - pytest.param( - FakeClientConfig( - server_mode=ClientServerMode.TrustedCertificates, - certificate_mode=ClientCertMode.Disabled, - trusted_certificates_location=CertificateLocation(LocationScheme.Unknown), - ), - id="unknown_trust_scheme", - ), # A File trust bundle that produced nothing must fail rather than fall back # to the platform trust store, which would silently widen trust far beyond # the configured anchors. diff --git a/tests/unit/test_nitlsconfig.py b/tests/unit/test_nitlsconfig.py index 483d0a9..7368e40 100644 --- a/tests/unit/test_nitlsconfig.py +++ b/tests/unit/test_nitlsconfig.py @@ -3,6 +3,7 @@ import json import pathlib import platform +import subprocess from typing import cast, Mapping, Optional, TypedDict import grpc @@ -16,6 +17,9 @@ CLIENT_FIXTURE_PATH = TEST_DIR / "nitlsconfig_client.json" SERVER_FIXTURE_PATH = TEST_DIR / "nitlsconfig_server.json" +# Captured before the autouse fixture below replaces it with a fixture-backed fake. +REAL_RUN_NITLSCONFIG_COMMAND = nitlsconfig_cli.run_nitlsconfig_command + class NitlsconfigJsonFixtures(TypedDict): """Data from nitlsconfig*.json fixtures.""" @@ -101,6 +105,22 @@ def fake_run_nitlsconfig_command(command_args: tuple[str, ...]) -> str: ) +def test_command_timeout_is_reported_as_a_package_error( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A hung CLI must surface through the package base class, not as a subprocess error.""" + + def fake_run(*args: object, **kwargs: object) -> object: + raise subprocess.TimeoutExpired(cmd="nitlsconfig", timeout=30) + + monkeypatch.setattr(subprocess, "run", fake_run) + + with pytest.raises(nitlsconfig.NitlsconfigError) as excinfo: + REAL_RUN_NITLSCONFIG_COMMAND(nitlsconfig_cli.build_list_command("client")) + + assert isinstance(excinfo.value, nitlsconfig.CommandTimeoutError) + + def test_list_client() -> None: """Test list_client.""" client_list = nitlsconfig.ClientConfig.list_services()