From ea8be57d6eec56067ad8c48a677e1e4458f5462c Mon Sep 17 00:00:00 2001 From: Alex Dubois Date: Wed, 2 Sep 2026 11:54:22 -0500 Subject: [PATCH 1/3] Get nitlsconfig repo ready for publishing documentation via readthedocs --- .readthedocs.yml | 16 ++++ README.md | 23 ++++-- docs/index.rst | 7 +- pyproject.toml | 5 +- src/nitlsconfig/__init__.py | 14 +--- src/nitlsconfig/{service.py => _service.py} | 4 +- src/nitlsconfig/audit.py | 75 ++++++++---------- src/nitlsconfig/channel_tag.py | 4 +- src/nitlsconfig/cli.py | 86 ++++++++++----------- src/nitlsconfig/errors.py | 2 +- src/nitlsconfig/grpc_channel.py | 51 ++++++------ tests/unit/test_audit.py | 4 +- tests/unit/test_nitlsconfig.py | 14 ++-- 13 files changed, 153 insertions(+), 152 deletions(-) create mode 100644 .readthedocs.yml rename src/nitlsconfig/{service.py => _service.py} (77%) diff --git a/.readthedocs.yml b/.readthedocs.yml new file mode 100644 index 0000000..a3a9bcf --- /dev/null +++ b/.readthedocs.yml @@ -0,0 +1,16 @@ +# .readthedocs.yml + +version: 2 + +build: + os: ubuntu-24.04 + tools: + python: "3.11" + jobs: + post_create_environment: + - pip install poetry==2.1.4 + post_install: + - VIRTUAL_ENV=$READTHEDOCS_VIRTUALENV_PATH poetry install --only main,docs + +sphinx: + configuration: docs/conf.py diff --git a/README.md b/README.md index f3af3e9..4129be4 100644 --- a/README.md +++ b/README.md @@ -1,7 +1,7 @@ # nitlsconfig Python API that reads nitlsconfig configurations through the `nitlsconfig` command line, -and builds gRPC client channels from them. +and builds NI gRPC Device client channels from them. Installed and imported as `nitlsconfig`; developed at [ni/nitlsconfig-python](https://github.com/ni/nitlsconfig-python). @@ -12,7 +12,7 @@ Installed and imported as `nitlsconfig`; developed at ## Install -Reading NI-TLS configuration is pure Python and has no third-party dependencies: +Reading NI TLS configuration is pure Python and has no third-party dependencies: - `pip install nitlsconfig` @@ -20,11 +20,16 @@ The gRPC channel factory additionally needs grpcio, which is an optional extra: - `pip install nitlsconfig[grpc]` +Neither install provides the `nitlsconfig` runtime itself. This package reads +configuration by invoking the `nitlsconfig` command line interface, which ships with NI +driver software products that support NI TLS. Install a driver that provides it before using +this package; without it, calls raise `ExecutableNotFoundError`. + ## Creating a gRPC channel -`create_grpc_device_channel` reads the local NI-TLS client configuration for the NI +`create_grpc_device_channel` reads the local NI TLS client configuration for the NI gRPC Device Server and returns a `grpc.Channel` secured accordingly. The -`server_address` hostname or address is used to select matching target-specific NI-TLS +`server_address` hostname or address is used to select matching target-specific NI TLS settings. Pass the channel straight to any NI gRPC Python API: ```python @@ -37,9 +42,12 @@ with nitlsconfig.create_grpc_device_channel("localhost", 31763) as channel: ... ``` -The channel is mutually authenticated, one-way TLS, or insecure depending on how -the machine is configured; no code change is needed to move between them. The -channel is owned by the caller - NI driver APIs never close it. +NI driver software provides the NI TLS configuration, and by default it expects mTLS, so the +channel is mutually authenticated. Falling back to one-way TLS or to an insecure connection +is an explicit change to that configuration, made through NI Hardware Manager. No code change +is needed to move between them. + +The channel is owned by the caller - NI driver APIs never close it. Retries are opt-in: @@ -89,7 +97,6 @@ if servers: print(server_info.certificate_key_location.scheme) print(server_info.trusted_certificates_location.scheme) print(server_info.trusted_certificates_contents) - print(server_info.certificate_key_contents) # Enumerate trusted certificates for cert in server_info.trusted_certificates: diff --git a/docs/index.rst b/docs/index.rst index 4f99be7..ba1afc1 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -1,11 +1,12 @@ -nitlsconfig Python API -====================== +.. mdinclude:: ../README.md + .. toctree:: :maxdepth: 3 + :caption: API Reference autoapi/index Indices and tables ------------------- +================== * :ref:`modindex` * :ref:`search` \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index 6ea858c..f126bba 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -2,7 +2,7 @@ name = "nitlsconfig" version = "1.0.0a3" license = "MIT" -description = "Python API for reading nitlsconfig configurations and creating gRPC client channels from them" +description = "Python API for reading nitlsconfig configurations and creating NI gRPC Device client channels from them" authors = [{name = "NI", email = "opensource@ni.com"}] maintainers = [ {name = "Philip Thong", email = "philip.thong@emerson.com"}, @@ -26,7 +26,7 @@ classifiers = [ requires-python = ">=3.9" dynamic = ["dependencies"] -# Reading NI-TLS configuration is pure Python. Only nitlsconfig.grpc_channel +# Reading NI TLS configuration is pure Python. Only nitlsconfig.grpc_channel # needs grpcio, so the binary wheel is opt-in. pywin32 rides along with it # because audit records are only produced when a channel is created, and the # Windows Event Log logging handler cannot be written to without it. @@ -35,6 +35,7 @@ grpc = ["grpcio>=1.49.0,<2.0", "pywin32>=306; sys_platform == 'win32'"] [project.urls] repository = "https://github.com/ni/nitlsconfig-python" +documentation = "https://nitlsconfig.readthedocs.io" [project.scripts] nitlsconfig-read = "nitlsconfig.cli:nitlsconfig_main" diff --git a/src/nitlsconfig/__init__.py b/src/nitlsconfig/__init__.py index b6f8f55..b84f35e 100644 --- a/src/nitlsconfig/__init__.py +++ b/src/nitlsconfig/__init__.py @@ -1,14 +1,8 @@ -"""Python package to read settings from nitlsconfig and build connections from them. +"""Python package to read settings from nitlsconfig and build NI gRPC Device channels from them. -Reading configuration is pure Python and has no third-party dependencies. The -gRPC channel factory needs grpcio, which is an optional extra:: - - pip install nitlsconfig[grpc] - -The gRPC names below are therefore resolved lazily: importing this package never -imports grpcio, so a caller that only reads NI-TLS configuration does not pay -for a binary dependency it will not use. Additional transports can be added the -same way without changing what a bare install requires. +The gRPC names below are resolved lazily: importing this package never imports +grpcio, so a caller that only reads NI TLS configuration does not pay for a +binary dependency it will not use. See the project README for install options. """ from importlib.metadata import version diff --git a/src/nitlsconfig/service.py b/src/nitlsconfig/_service.py similarity index 77% rename from src/nitlsconfig/service.py rename to src/nitlsconfig/_service.py index 649a438..37ad8d3 100644 --- a/src/nitlsconfig/service.py +++ b/src/nitlsconfig/_service.py @@ -1,4 +1,4 @@ -"""The NI-TLS services this package builds transports for. +"""The NI TLS services this package builds transports for. Only the NI gRPC Device Server is supported today. Anything else can still be read through :class:`~nitlsconfig.cli.ClientConfig`, but has no channel factory. @@ -6,7 +6,7 @@ from __future__ import annotations -# The NI-TLS registered service name for the NI gRPC Device Server: the file stem of +# The NI TLS registered service name for the NI gRPC Device Server: the file stem of # ni-grpc-device.client.caps.yml, the Event Log source, and the record tag are all this # one name, so records can be tied back to the configuration they describe. SERVICE_NAME = "ni-grpc-device" diff --git a/src/nitlsconfig/audit.py b/src/nitlsconfig/audit.py index 5fc69e2..a3dca7c 100644 --- a/src/nitlsconfig/audit.py +++ b/src/nitlsconfig/audit.py @@ -1,37 +1,12 @@ -"""Audit logging for NI-TLS client transports. +"""Audit logging for NI TLS client transports. -Writes to the platform audit log similarly to other NI gRPC clients: +Records the security posture of transports this package creates, and the outcome of a +driver's gRPC session initialize RPC, to the platform audit log: the Windows Event Log +on Windows, syslog on Linux. -Windows: Windows Event Log -Linux: syslog - -We use the record pattern ``[][] ``. - -A record is emitted only for something that attests to the security posture of a -connection that actually existed: transport posture and session connect results. - -Configuration errors are not audited, because no channel is created and nothing -is transmitted; :class:`TlsConfigurationError` carries that detail to the caller -directly. - -We also do not record *which* server we authenticated, for example its -certificate subject. gRPC's Python client API takes certificates as input only: -it offers no callback during the handshake and no way to read the server's -certificate afterward. - -Transport posture is emitted by the channel factory. -The session connect record cannot be: it reports the outcome of a driver's -initialize RPC, and a gRPC channel connects lazily, so the API layer that -issues that RPC has to call ``audit_session_connect`` itself. - -Auditing covers the NI gRPC Device Server only, so the service name is fixed -package-wide rather than accepted from callers. Should another service ever need -audit records, this module grows a service parameter again at that point. - -Records report what this package observed, assuming the hosting process is not -hostile. Nothing here can defend against code in the same process, which can -call the standard library logger directly. Untrusted *values* reaching a record -are bounded and escaped below, because those do cross a trust boundary. +Transport posture is recorded by the channel factory. The session connect outcome +cannot be, because a gRPC channel connects lazily, so the driver API layer that issues +the initialize RPC calls :func:`audit_session_connect` itself. """ from __future__ import annotations @@ -41,13 +16,30 @@ import threading from enum import Enum +from nitlsconfig._service import SERVICE_NAME from nitlsconfig.channel_tag import get_channel_target -from nitlsconfig.service import SERVICE_NAME + +# A record is emitted only for something that attests to the security posture of a +# connection that actually existed. Configuration errors are not audited, because no +# channel is created and nothing is transmitted; TlsConfigurationError carries that +# detail to the caller directly. We also do not record *which* server we authenticated, +# for example its certificate subject: gRPC's Python client API takes certificates as +# input only, offering no handshake callback and no way to read the server's certificate +# afterward. +# +# Auditing covers the NI gRPC Device Server only, so the service name is fixed +# package-wide rather than accepted from callers. Should another service ever need audit +# records, this module grows a service parameter again at that point. +# +# Records report what this package observed, assuming the hosting process is not hostile. +# Nothing here can defend against code in the same process, which can call the standard +# library logger directly. Untrusted *values* reaching a record are bounded and escaped +# below, because those do cross a trust boundary. _ROLE = "Client" -class TransportSecurity(Enum): +class _TransportSecurity(Enum): """Security posture of a created transport.""" Unencrypted = "unencrypted" @@ -114,7 +106,7 @@ def _make_logging_handler() -> logging.Handler: # The platform logging handler failed, not `logging` itself; this diagnostic # goes to the host application's ordinary logger, never to the audit channel. logging.getLogger(__name__).warning( - "NI-TLS audit logging is unavailable on this system; audit events " + "NI TLS audit logging is unavailable on this system; audit events " "will not be recorded.", exc_info=True, ) @@ -140,6 +132,7 @@ def _get_audit_logger() -> logging.Logger: logger.propagate = False handler = _make_logging_handler() + # Record pattern: [][] handler.setFormatter(logging.Formatter(f"[{SERVICE_NAME}][{_ROLE}] %(message)s")) logger.addHandler(handler) _logging_handler_attached = True @@ -147,7 +140,7 @@ def _get_audit_logger() -> logging.Logger: return logger -def audit_transport_posture(peer_host: str, security: TransportSecurity) -> None: +def _audit_transport_posture(peer_host: str, security: _TransportSecurity) -> None: """Record the security posture of a client transport. Never raises: auditing must not disrupt transport creation. @@ -158,16 +151,16 @@ def audit_transport_posture(peer_host: str, security: TransportSecurity) -> None if peer_host: message += f" to '{peer_host}'" - if security is TransportSecurity.Unencrypted: + if security is _TransportSecurity.Unencrypted: message += " is unencrypted (TLS disabled)." - elif security is TransportSecurity.ServerAuthenticatedTls: + elif security is _TransportSecurity.ServerAuthenticatedTls: message += " uses one-way TLS. Not presenting a client certificate." else: message += " uses mutual TLS. Presenting a client certificate." logger = _get_audit_logger() # Mutual TLS is the secure baseline; weaker postures are auditable warnings. - if security is TransportSecurity.MutualTls: + if security is _TransportSecurity.MutualTls: logger.info(message) else: logger.warning(message) @@ -183,9 +176,9 @@ def audit_session_connect(driver_name: str, channel: object, connected: bool) -> Channels this package did not create are ignored, so drivers can call this unconditionally. A caller who built their own channel never went through - NI-TLS, so there is no transport posture record to pair the outcome with and + NI TLS, so there is no transport posture record to pair the outcome with and nothing to attest to; auditing it anyway would also register an Event Log - source on machines not using NI-TLS at all. + source on machines not using NI TLS at all. """ try: target = get_channel_target(channel) diff --git a/src/nitlsconfig/channel_tag.py b/src/nitlsconfig/channel_tag.py index 7acb8ee..55e75f2 100644 --- a/src/nitlsconfig/channel_tag.py +++ b/src/nitlsconfig/channel_tag.py @@ -1,6 +1,6 @@ -"""Marks gRPC channels this package created. +"""Marks NI gRPC Device channels this package created. -A caller holding only a channel cannot tell whether NI-TLS had any part in +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. diff --git a/src/nitlsconfig/cli.py b/src/nitlsconfig/cli.py index a5363f4..3c02d96 100644 --- a/src/nitlsconfig/cli.py +++ b/src/nitlsconfig/cli.py @@ -18,19 +18,19 @@ NitlsconfigCliError, ) -ALLOWED_SCOPES: Tuple[str, ...] = ("client", "server") +_ALLOWED_SCOPES: Tuple[str, ...] = ("client", "server") # Expected JSON root keys from nitlsconfig output. -ROLE_TO_JSON_ROOT_KEY = { +_ROLE_TO_JSON_ROOT_KEY = { "client": "client", "server": "server", } # Command templates are based on nitlsconfigcli README and fixture scripts. -LIST_COMMAND_TEMPLATE: Tuple[str, ...] = ("{role}", "list") +_LIST_COMMAND_TEMPLATE: Tuple[str, ...] = ("{role}", "list") -CLIENT_BATCH_READ_COMMAND_TEMPLATE: Tuple[str, ...] = ( +_CLIENT_BATCH_READ_COMMAND_TEMPLATE: Tuple[str, ...] = ( "--output-format=json", "client", "batch-read", @@ -58,7 +58,7 @@ "trusted_certificates_contents", ) -SERVER_BATCH_READ_COMMAND_TEMPLATE: Tuple[str, ...] = ( +_SERVER_BATCH_READ_COMMAND_TEMPLATE: Tuple[str, ...] = ( "--output-format=json", "server", "batch-read", @@ -168,10 +168,10 @@ def __str__(self) -> str: return self.to_string() -EnumT = TypeVar("EnumT", bound=Enum) +_EnumT = TypeVar("_EnumT", bound=Enum) -def _parse_enum(value: str, enum_cls: type[EnumT], unknown_member: EnumT) -> EnumT: +def _parse_enum(value: str, enum_cls: type[_EnumT], unknown_member: _EnumT) -> _EnumT: "Parse a string value into an Enum member, returning unknown_member if not found." try: return enum_cls(value) @@ -196,7 +196,7 @@ class KnownServerData: trusted_certificates_contents: str = field() @classmethod - def from_json_obj(cls, obj: dict[str, Any]) -> "KnownServerData": + def _from_json_obj(cls, obj: dict[str, Any]) -> "KnownServerData": "Parse a known server object from CLI JSON into KnownServerData." return cls( raw=obj, @@ -237,7 +237,7 @@ class TrustedCertificateData: trusted_certificate_contents: str = field() @classmethod - def from_json_obj(cls, obj: dict[str, Any]) -> "TrustedCertificateData": + def _from_json_obj(cls, obj: dict[str, Any]) -> "TrustedCertificateData": "Parse a trusted certificate object from CLI JSON into TrustedCertificateData." return cls( raw=obj, @@ -250,7 +250,7 @@ def from_json_obj(cls, obj: dict[str, Any]) -> "TrustedCertificateData": @dataclass(frozen=True) -class ServiceData: +class _ServiceData: """Typed wrapper for service objects from CLI JSON.""" raw: dict[str, Any] @@ -258,21 +258,21 @@ class ServiceData: trusted_certificates: list[TrustedCertificateData] = field(default_factory=list) @classmethod - def from_json_obj(cls, obj: dict[str, Any]) -> "ServiceData": - "Parse a service object from CLI JSON into a typed ServiceData object." + def from_json_obj(cls, obj: dict[str, Any]) -> "_ServiceData": + "Parse a service object from CLI JSON into a typed _ServiceData object." known_servers_raw = obj.get("known_servers", []) known_servers: list[KnownServerData] = [] if isinstance(known_servers_raw, list): for item in known_servers_raw: if isinstance(item, dict): - known_servers.append(KnownServerData.from_json_obj(item)) + known_servers.append(KnownServerData._from_json_obj(item)) trusted_certificates_raw = obj.get("trusted_certificates", []) trusted_certificates: list[TrustedCertificateData] = [] if isinstance(trusted_certificates_raw, list): for item in trusted_certificates_raw: if isinstance(item, dict): - trusted_certificates.append(TrustedCertificateData.from_json_obj(item)) + trusted_certificates.append(TrustedCertificateData._from_json_obj(item)) return cls(raw=obj, known_servers=known_servers, trusted_certificates=trusted_certificates) @@ -284,25 +284,25 @@ def value(self, key: str, default: str = "") -> str: return default -def build_list_command(role: str) -> Tuple[str, ...]: +def _build_list_command(role: str) -> Tuple[str, ...]: """Build argv for list mode only. This function does not include executable resolution. """ - return tuple(part.format(role=role) for part in LIST_COMMAND_TEMPLATE) + return tuple(part.format(role=role) for part in _LIST_COMMAND_TEMPLATE) -def build_batch_read_command(role: str) -> Tuple[str, ...]: +def _build_batch_read_command(role: str) -> Tuple[str, ...]: """Build argv template for batch-read mode only. The template mirrors existing fixture-generation scripts. """ if role == "client": - return CLIENT_BATCH_READ_COMMAND_TEMPLATE - return SERVER_BATCH_READ_COMMAND_TEMPLATE + return _CLIENT_BATCH_READ_COMMAND_TEMPLATE + return _SERVER_BATCH_READ_COMMAND_TEMPLATE -def run_nitlsconfig_command( +def _run_nitlsconfig_command( command_args: Tuple[str, ...], ) -> str: """Run nitlsconfig command and return stdout. @@ -345,11 +345,11 @@ def run_nitlsconfig_command( return completed.stdout -def run_nitlsconfig_json_command( +def _run_nitlsconfig_json_command( command_args: Tuple[str, ...], ) -> Any: """Run nitlsconfig command and parse stdout as JSON.""" - stdout = run_nitlsconfig_command( + stdout = _run_nitlsconfig_command( command_args=command_args, ) @@ -368,24 +368,24 @@ def _list_services(scope: str) -> list[str]: Output is parsed line-by-line and normalized by stripping whitespace and dropping empty lines. """ - stdout = run_nitlsconfig_command(command_args=build_list_command(scope)) + stdout = _run_nitlsconfig_command(command_args=_build_list_command(scope)) return [line.strip() for line in stdout.splitlines() if line.strip()] -def _read_services(scope: str) -> list[ServiceData]: +def _read_services(scope: str) -> list[_ServiceData]: """Read full service configurations for the requested scope. Parsed output preserves the original key casing and values from the CLI JSON. """ - payload = run_nitlsconfig_json_command(command_args=build_batch_read_command(scope)) + payload = _run_nitlsconfig_json_command(command_args=_build_batch_read_command(scope)) if not isinstance(payload, dict): raise InvalidOutputError("nitlsconfig JSON output root must be an object") - if scope not in ROLE_TO_JSON_ROOT_KEY: + if scope not in _ROLE_TO_JSON_ROOT_KEY: raise InvalidOutputError(f"Unsupported scope for nitlsconfig JSON output: {scope!r}") - root_key = ROLE_TO_JSON_ROOT_KEY[scope] + root_key = _ROLE_TO_JSON_ROOT_KEY[scope] if root_key not in payload: raise InvalidOutputError(f"nitlsconfig JSON output missing expected root key: {root_key!r}") @@ -393,10 +393,10 @@ def _read_services(scope: str) -> list[ServiceData]: if not isinstance(services, list): raise InvalidOutputError(f"nitlsconfig JSON root key {root_key!r} must contain a list") - normalized: list[ServiceData] = [] + normalized: list[_ServiceData] = [] for item in services: if isinstance(item, dict): - normalized.append(ServiceData.from_json_obj(item)) + normalized.append(_ServiceData.from_json_obj(item)) else: raise InvalidOutputError("nitlsconfig service entries must be objects") return normalized @@ -416,18 +416,18 @@ def list_services(cls) -> list[str]: return _list_services(cls._scope) @classmethod - def _read_all(cls) -> list[ServiceData]: + def _read_all(cls) -> list[_ServiceData]: return _read_services(cls._scope) @classmethod - def _find_service_data(cls, service_name: str) -> ServiceData: + def _find_service_data(cls, service_name: str) -> _ServiceData: for item in cls._read_all(): if item.value("service_name") == service_name: return item # Keep behavior compatible with minimal list-only service entries. if service_name in cls.list_services(): - return ServiceData(raw={"service_name": service_name}) + return _ServiceData(raw={"service_name": service_name}) raise InvalidOutputError(f"Service not found: {service_name!r}") @@ -438,17 +438,9 @@ def _location(self, key: str) -> CertificateLocation: return CertificateLocation.from_string(self._value(key)) @property - def certificate_mode_raw(self) -> str: + def _certificate_mode_raw(self) -> str: return self._value("certificate_mode") - @property - def certificate_chain_location_raw(self) -> str: - return self._value("certificate_chain_location") - - @property - def certificate_chain_contents_raw(self) -> str: - return self._value("certificate_chain_contents") - class ServerConfig(_BaseConfig): """Read-only view for server-side TLS configuration.""" @@ -459,7 +451,7 @@ class ServerConfig(_BaseConfig): def certificate_mode(self) -> ServerCertMode: "Parse certificate_mode string into ServerCertMode enum, defaulting to Unknown." return _parse_enum( - self.certificate_mode_raw, + self._certificate_mode_raw, ServerCertMode, ServerCertMode.Unknown, ) @@ -481,7 +473,7 @@ def certificate_key_location(self) -> CertificateLocation: @property def certificate_key_contents(self) -> str: - "Return the raw certificate_key_contents string from the service configuration." + "Return the server private key as PEM. Secret material: do not log or persist it." return self._value("certificate_key_contents") @property @@ -538,7 +530,7 @@ def __init__(self, service_name: str, server_address: Optional[str] = None) -> N None, ) if known_server is not None: - self._resolved_data = ServiceData(raw=known_server.raw) + self._resolved_data = _ServiceData(raw=known_server.raw) def _value(self, key: str, default: str = "") -> str: return self._resolved_data.value(key, default) @@ -547,7 +539,7 @@ def _value(self, key: str, default: str = "") -> str: def certificate_mode(self) -> ClientCertMode: "Parse certificate_mode string into ClientCertMode enum, defaulting to Unknown." return _parse_enum( - self.certificate_mode_raw, + self._certificate_mode_raw, ClientCertMode, ClientCertMode.Unknown, ) @@ -568,7 +560,7 @@ def certificate_key_location(self) -> CertificateLocation: @property def certificate_key_contents(self) -> str: - "Return the raw certificate_key_contents string from the service configuration." + "Return the client private key as PEM. Secret material: do not log or persist it." return self._value("certificate_key_contents") @property @@ -604,7 +596,7 @@ def __str__(self) -> str: def nitlsconfig_main(argv: Optional[list[str]] = None) -> int: """Console entry point for read-only service listing.""" parser = argparse.ArgumentParser(prog="nitlsconfig-read") - parser.add_argument("scope", choices=ALLOWED_SCOPES) + parser.add_argument("scope", choices=_ALLOWED_SCOPES) parser.add_argument("command", choices=["list"], nargs="?", default="list") args = parser.parse_args(argv) diff --git a/src/nitlsconfig/errors.py b/src/nitlsconfig/errors.py index f60f37e..29975b9 100644 --- a/src/nitlsconfig/errors.py +++ b/src/nitlsconfig/errors.py @@ -32,7 +32,7 @@ class InvalidOutputError(NitlsconfigCliError): class TlsConfigurationError(NitlsconfigError): - """Raised when the NI-TLS configuration was read successfully but is invalid. + """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 diff --git a/src/nitlsconfig/grpc_channel.py b/src/nitlsconfig/grpc_channel.py index 225c0ee..bf6b19a 100644 --- a/src/nitlsconfig/grpc_channel.py +++ b/src/nitlsconfig/grpc_channel.py @@ -1,8 +1,8 @@ -"""Create gRPC channels to the NI gRPC Device Server from NI-TLS (nitlsconfig) configuration. +"""Create gRPC channels to the NI gRPC Device Server from NI TLS (nitlsconfig) configuration. -Reads the local NI-TLS client configuration for the NI gRPC Device Server and -produces a :class:`grpc.Channel` that is either secured with TLS/mTLS or, when -TLS is not configured, a plain insecure channel. +Reads the local NI TLS client configuration for the NI gRPC Device Server and +produces a :class:`grpc.Channel` secured with TLS/mTLS, or a plain insecure +channel when TLS has been explicitly toggled off. The resulting channel is a normal ``grpc.Channel``. It can be handed directly to any NI gRPC Python API, for example:: @@ -26,14 +26,8 @@ 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 -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. +``server_mode`` Disabled selects a plain connection. Every other mode verifies +the server certificate chain and checks the hostname. 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 @@ -48,9 +42,10 @@ import grpc +from nitlsconfig._service import SERVICE_NAME from nitlsconfig.audit import ( - TransportSecurity, - audit_transport_posture, + _TransportSecurity, + _audit_transport_posture, ) from nitlsconfig.channel_tag import tag_channel_target from nitlsconfig.cli import ( @@ -60,7 +55,6 @@ LocationScheme, ) from nitlsconfig.errors import TlsConfigurationError -from nitlsconfig.service import SERVICE_NAME __all__ = [ "RetryPolicy", @@ -210,8 +204,11 @@ def _load_client_tls_settings(config: ClientConfig) -> Optional[_ClientTlsSettin service_name = config.service_name # 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. + # and Unknown all fall through to full chain and hostname verification, because gRPC's + # Python API cannot relax either check independently: that needs 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 those + # callers get a stricter connection than requested rather than a weaker one. if config.server_mode == ClientServerMode.Disabled: return None @@ -283,17 +280,17 @@ def create_grpc_device_channel( options: ChannelOptions = (), retry_policy: Optional[RetryPolicy] = None, ) -> grpc.Channel: - """Create a gRPC channel to ``server_address:server_port`` using NI-TLS configuration. + """Create an NI gRPC Device channel to ``server_address:server_port``. - Reads the NI-TLS client configuration for the NI gRPC Device Server and builds + Reads the NI TLS client configuration for the NI gRPC Device Server and builds a channel that verifies the server certificate and, when the configuration - calls for mutual TLS, also presents the client certificate. Falls back to an - insecure channel when the client's ``server_mode`` is Disabled, which is the - default until the machine is configured. + calls for mutual TLS, also presents the client certificate. The channel is + insecure only when the client's ``server_mode`` has been explicitly set to + Disabled. Args: server_address: Host name or address of the NI gRPC Device Server. Also used - to resolve NI-TLS settings specific to this target. IPv6 literals may be + to resolve NI TLS settings specific to this target. IPv6 literals may be passed with or without brackets. server_port: Port of the NI gRPC Device Server. options: gRPC channel arguments, as ``(key, value)`` pairs. Use this to @@ -320,13 +317,13 @@ def create_grpc_device_channel( settings = _load_client_tls_settings(ClientConfig(SERVICE_NAME, server_address)) if settings is None: - security = TransportSecurity.Unencrypted + security = _TransportSecurity.Unencrypted elif settings.present_client_cert: - security = TransportSecurity.MutualTls + security = _TransportSecurity.MutualTls else: - security = TransportSecurity.ServerAuthenticatedTls + security = _TransportSecurity.ServerAuthenticatedTls - audit_transport_posture(server_address, security) + _audit_transport_posture(server_address, security) if settings is None: channel = grpc.insecure_channel(target, options=channel_options) diff --git a/tests/unit/test_audit.py b/tests/unit/test_audit.py index e6d9ae9..3d35932 100644 --- a/tests/unit/test_audit.py +++ b/tests/unit/test_audit.py @@ -7,9 +7,9 @@ from nitlsconfig import audit from nitlsconfig.audit import ( - TransportSecurity, + _TransportSecurity as TransportSecurity, + _audit_transport_posture as audit_transport_posture, audit_session_connect, - audit_transport_posture, ) from nitlsconfig.channel_tag import tag_channel_target diff --git a/tests/unit/test_nitlsconfig.py b/tests/unit/test_nitlsconfig.py index 1e33646..8542166 100644 --- a/tests/unit/test_nitlsconfig.py +++ b/tests/unit/test_nitlsconfig.py @@ -18,7 +18,7 @@ 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 +REAL_RUN_NITLSCONFIG_COMMAND = nitlsconfig_cli._run_nitlsconfig_command class NitlsconfigJsonFixtures(TypedDict): @@ -67,12 +67,12 @@ def mock_nitlsconfig_command( nitlsconfig_json_fixtures: NitlsconfigJsonFixtures, ) -> None: command_responses: dict[tuple[str, ...], str] = { - nitlsconfig_cli.build_list_command("client"): nitlsconfig_json_fixtures["client_list"], - nitlsconfig_cli.build_list_command("server"): nitlsconfig_json_fixtures["server_list"], - nitlsconfig_cli.build_batch_read_command("client"): nitlsconfig_json_fixtures[ + nitlsconfig_cli._build_list_command("client"): nitlsconfig_json_fixtures["client_list"], + nitlsconfig_cli._build_list_command("server"): nitlsconfig_json_fixtures["server_list"], + nitlsconfig_cli._build_batch_read_command("client"): nitlsconfig_json_fixtures[ "client_json" ], - nitlsconfig_cli.build_batch_read_command("server"): nitlsconfig_json_fixtures[ + nitlsconfig_cli._build_batch_read_command("server"): nitlsconfig_json_fixtures[ "server_json" ], } @@ -100,7 +100,7 @@ def fake_run_nitlsconfig_command(command_args: tuple[str, ...]) -> str: monkeypatch.setattr( nitlsconfig_cli, - "run_nitlsconfig_command", + "_run_nitlsconfig_command", fake_run_nitlsconfig_command, ) @@ -116,7 +116,7 @@ def fake_run(*args: object, **kwargs: object) -> object: monkeypatch.setattr(subprocess, "run", fake_run) with pytest.raises(nitlsconfig.NitlsconfigError) as excinfo: - REAL_RUN_NITLSCONFIG_COMMAND(nitlsconfig_cli.build_list_command("client")) + REAL_RUN_NITLSCONFIG_COMMAND(nitlsconfig_cli._build_list_command("client")) assert isinstance(excinfo.value, nitlsconfig.CommandTimeoutError) From 873332219cbced23cf4570113de51ff6b567bdf5 Mon Sep 17 00:00:00 2001 From: Alex Dubois Date: Wed, 2 Sep 2026 11:58:32 -0500 Subject: [PATCH 2/3] Fix readthedocs documentation to be nitlsconfig-python --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index f126bba..ae0ede2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -35,7 +35,7 @@ grpc = ["grpcio>=1.49.0,<2.0", "pywin32>=306; sys_platform == 'win32'"] [project.urls] repository = "https://github.com/ni/nitlsconfig-python" -documentation = "https://nitlsconfig.readthedocs.io" +documentation = "https://nitlsconfig-python.readthedocs.io" [project.scripts] nitlsconfig-read = "nitlsconfig.cli:nitlsconfig_main" From 39e76b7703940bacba69ea3e3764ae3245043217 Mon Sep 17 00:00:00 2001 From: Alex Dubois Date: Wed, 2 Sep 2026 13:43:32 -0500 Subject: [PATCH 3/3] Further improve documentation to provide better aligned documentation with what we intend to provide to nimi-python to make our details less vague --- README.md | 15 +++++++++++---- src/nitlsconfig/grpc_channel.py | 4 ++++ 2 files changed, 15 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 4129be4..8c58642 100644 --- a/README.md +++ b/README.md @@ -42,10 +42,17 @@ with nitlsconfig.create_grpc_device_channel("localhost", 31763) as channel: ... ``` -NI driver software provides the NI TLS configuration, and by default it expects mTLS, so the -channel is mutually authenticated. Falling back to one-way TLS or to an insecure connection -is an explicit change to that configuration, made through NI Hardware Manager. No code change -is needed to move between them. +NI driver software provides the NI TLS configuration. Using `create_grpc_device_channel` +opts into mTLS. + +Before `create_grpc_device_channel` can succeed, use NI Hardware Manager to perform a +certificate exchange with the remote system. See +[Managing mTLS](https://www.ni.com/docs/en-US/bundle/hardwaremanager/page/mtls-manage.html) +for details. + +Weakening the security posture to one-way TLS or to an insecure connection +requires explicitly changing that configuration in NI Hardware Manager. Either way, no +code change is needed. The channel is owned by the caller - NI driver APIs never close it. diff --git a/src/nitlsconfig/grpc_channel.py b/src/nitlsconfig/grpc_channel.py index bf6b19a..7d1d2ad 100644 --- a/src/nitlsconfig/grpc_channel.py +++ b/src/nitlsconfig/grpc_channel.py @@ -288,6 +288,10 @@ def create_grpc_device_channel( insecure only when the client's ``server_mode`` has been explicitly set to Disabled. + TLS requires a certificate exchange with the remote system, performed in + NI Hardware Manager. See `Managing mTLS + `_. + Args: server_address: Host name or address of the NI gRPC Device Server. Also used to resolve NI TLS settings specific to this target. IPv6 literals may be