diff --git a/.gitignore b/.gitignore index 65f0a15..0a368c6 100644 --- a/.gitignore +++ b/.gitignore @@ -16,6 +16,9 @@ venv.bak/ # macOS .DS_Store +# Local git worktrees +.worktrees/ + # Python __pycache__/ *.py[cod] @@ -26,3 +29,5 @@ dist/ .eggs/ *.egg-info poetry.lock + +.worktrees/ diff --git a/README.md b/README.md index 1976e0d..15ecf34 100644 --- a/README.md +++ b/README.md @@ -29,7 +29,7 @@ pip install netboxlabs-diode-sdk * `DIODE_CLIENT_SECRET` - Client Secret for OAuth2 authentication * `DIODE_MAX_AUTH_RETRIES` - Maximum attempts for OAuth2 token fetch and gRPC re-authentication on `Unauthenticated` (default: `3`). Token fetch retries with exponential backoff on `429`, `500`, `502`, and `503`, honouring `Retry-After` when present on `429`/`503`. * `DIODE_CERT_FILE` - Path to custom certificate file for TLS connections -* `DIODE_SKIP_TLS_VERIFY` - Skip TLS verification (default: `false`) +* `DIODE_SKIP_TLS_VERIFY` - Skip TLS certificate verification for `grpcs://` / `https://` targets (default: `false`). Traffic stays encrypted; only chain validation is disabled. Prefer `DIODE_CERT_FILE` for self-signed production servers. * `DIODE_DRY_RUN_OUTPUT_DIR` - Directory where `DiodeDryRunClient` will write JSON files ### Example @@ -232,7 +232,7 @@ export NO_PROXY=localhost,127.0.0.1,.example.com **Important notes for proxy usage:** -1. **Proxy with SKIP_TLS_VERIFY**: When using HTTP(S) proxies, the SDK **always uses secure channels** because proxies require TLS for the CONNECT tunnel. Setting `DIODE_SKIP_TLS_VERIFY=true` with a proxy will log a warning and use a secure channel anyway. +1. **Proxy with SKIP_TLS_VERIFY**: For `grpcs://` / `https://` targets the SDK uses a **secure gRPC channel** (including when `DIODE_SKIP_TLS_VERIFY=true`). Proxies use an HTTP CONNECT tunnel; skip-verify disables certificate checks only, not TLS. Plaintext `grpc://` targets stay on insecure channels and use `HTTP_PROXY`. 2. **MITM proxies (like mitmproxy)**: To use an intercepting proxy, you must provide the proxy's CA certificate: ```bash @@ -268,10 +268,14 @@ export DIODE_CERT_FILE=/path/to/cert.pem #### Disabling TLS verification +Use this only as a development or break-glass escape hatch. The connection remains TLS-encrypted; the SDK skips validating the server certificate (same idea as Go `InsecureSkipVerify`). + ```bash export DIODE_SKIP_TLS_VERIFY=true ``` +For self-signed or private-CA servers in production, mount the CA or server cert via `DIODE_CERT_FILE` instead. + #### For legacy certificates (CN-only, no SANs) ```python diff --git a/netboxlabs/diode/sdk/client.py b/netboxlabs/diode/sdk/client.py index 3645769..a3049ea 100644 --- a/netboxlabs/diode/sdk/client.py +++ b/netboxlabs/diode/sdk/client.py @@ -8,6 +8,8 @@ import os import platform import random +import socket +import ssl import sys import tempfile import time @@ -86,21 +88,13 @@ def _load_certs(cert_file: str | None = None) -> bytes: return f.read() -def _should_verify_tls(scheme: str) -> bool: - """Determine if TLS verification should be enabled based on scheme and environment variable.""" - # Check if scheme is insecure - insecure_scheme = scheme in ["grpc", "http"] - - # Check environment variable +def _skip_tls_verify_from_env() -> bool: skip_tls_env = os.getenv(_DIODE_SKIP_TLS_VERIFY_ENVVAR_NAME, "").lower() - skip_tls_from_env = skip_tls_env in ["true", "1", "yes", "on"] - - # TLS verification is enabled by default, disabled only for insecure schemes or env var - return not (insecure_scheme or skip_tls_from_env) + return skip_tls_env in ["true", "1", "yes", "on"] -def parse_target(target: str) -> tuple[str, str, bool]: - """Parse the target into authority, path and tls_verify.""" +def parse_target(target: str) -> tuple[str, str, bool, bool]: + """Parse the target into authority, path, is_plaintext, and tls_verify.""" parsed_target = urlparse(target) if parsed_target.scheme not in ["grpc", "grpcs", "http", "https"]: @@ -108,8 +102,8 @@ def parse_target(target: str) -> tuple[str, str, bool]: "target should start with grpc://, grpcs://, http:// or https://" ) - # Determine if TLS verification should be enabled - tls_verify = _should_verify_tls(parsed_target.scheme) + is_plaintext = parsed_target.scheme in ("grpc", "http") + tls_verify = (not is_plaintext) and (not _skip_tls_verify_from_env()) authority = parsed_target.netloc @@ -119,7 +113,128 @@ def parse_target(target: str) -> tuple[str, str, bool]: elif parsed_target.scheme in ["grpcs", "https"]: authority += ":443" - return authority, parsed_target.path, tls_verify + return authority, parsed_target.path, is_plaintext, tls_verify + + +def _tls_server_name_from_cert_pem(pem: bytes) -> str | None: + with tempfile.NamedTemporaryFile(mode="wb", delete=False, suffix=".pem") as cert_file: + cert_file.write(pem) + cert_path = cert_file.name + try: + decoded = ssl._ssl._test_decode_cert(cert_path) + except ssl.SSLError: + return None + finally: + os.unlink(cert_path) + + san = decoded.get("subjectAltName") + if san: + for name_type, value in san: + if name_type == "DNS": + return value + + for rdn in decoded.get("subject", ()): + for key, value in rdn: + if key == "commonName": + return value + return None + + +def _connect_socket(authority: str, proxy_url: str | None) -> socket.socket: + host, port_str = authority.rsplit(":", 1) + port = int(port_str) + + if not proxy_url: + return socket.create_connection((host, port), timeout=10) + + parsed_proxy = urlparse(proxy_url) + if not parsed_proxy.hostname: + raise DiodeConfigError(f"Invalid proxy URL: {proxy_url}") + proxy_port = parsed_proxy.port or (443 if parsed_proxy.scheme == "https" else 80) + sock = socket.create_connection((parsed_proxy.hostname, proxy_port), timeout=10) + connect_request = ( + f"CONNECT {host}:{port} HTTP/1.1\r\n" + f"Host: {host}:{port}\r\n\r\n" + ) + sock.sendall(connect_request.encode()) + response = b"" + while b"\r\n\r\n" not in response: + chunk = sock.recv(4096) + if not chunk: + break + response += chunk + status_line = response.split(b"\r\n", 1)[0] + if b" 200 " not in status_line: + sock.close() + raise DiodeConfigError(f"Proxy CONNECT failed for {authority}") + return sock + + +def _fetch_peer_leaf_certificate( + authority: str, proxy_url: str | None = None +) -> tuple[bytes, str]: + host, _ = authority.rsplit(":", 1) + raw_sock = _connect_socket(authority, proxy_url) + try: + context = ssl.create_default_context() + context.check_hostname = False + context.verify_mode = ssl.CERT_NONE + tls_sock = context.wrap_socket(raw_sock, server_hostname=host) + der_cert = tls_sock.getpeercert(binary_form=True) + if not der_cert: + raise DiodeConfigError( + f"No peer certificate returned from {authority}" + ) + pem = ssl.DER_cert_to_PEM_cert(der_cert).encode() + finally: + raw_sock.close() + + server_name = _tls_server_name_from_cert_pem(pem) or host + return pem, server_name + + +def _skip_verify_channel_credentials( + authority: str, proxy_url: str | None +) -> tuple[grpc.ChannelCredentials, tuple[tuple[str, str], ...]]: + pem, server_name = _fetch_peer_leaf_certificate(authority, proxy_url) + credentials = grpc.ssl_channel_credentials(root_certificates=pem) + return credentials, (("grpc.ssl_target_name_override", server_name),) + + +def _open_grpc_channel( + target: str, + *, + is_plaintext: bool, + tls_verify: bool, + certificates: bytes | None, + channel_options: tuple, + proxy_url: str | None, + proxy_ssl_target_name_override: bool = False, +) -> grpc.Channel: + opts = list(channel_options) + if is_plaintext: + _LOGGER.debug("Setting up gRPC insecure channel") + return grpc.insecure_channel(target=target, options=tuple(opts)) + + if tls_verify: + credentials = ( + grpc.ssl_channel_credentials(root_certificates=certificates) + if certificates + else grpc.ssl_channel_credentials() + ) + if proxy_url and proxy_ssl_target_name_override: + opts.append(("grpc.ssl_target_name_override", target.split(":")[0])) + _LOGGER.debug( + f"Setting up gRPC secure channel with " + f"{'custom certificates' if certificates else 'system certificates'}" + f"{' via proxy' if proxy_url else ''}" + ) + return grpc.secure_channel(target, credentials, options=tuple(opts)) + + credentials, extra_opts = _skip_verify_channel_credentials(target, proxy_url) + opts.extend(extra_opts) + _LOGGER.debug("Setting up gRPC secure channel with TLS verification disabled") + return grpc.secure_channel(target, credentials, options=tuple(opts)) def _get_sentry_dsn(sentry_dsn: str | None = None) -> str | None: @@ -334,6 +449,7 @@ def __init__( sentry_profiles_sample_rate: float = 1.0, max_auth_retries: int = 3, cert_file: str | None = None, + skip_tls_verify: bool = False, ): """Initiate a new client.""" log_level = os.getenv(_DIODE_SDK_LOG_LEVEL_ENVVAR_NAME, "INFO").upper() @@ -346,13 +462,11 @@ def __init__( self._cert_file = _get_optional_config_value( _DIODE_CERT_FILE_ENVVAR_NAME, cert_file ) - self._target, self._path, self._tls_verify = parse_target(target) - # Whether the target scheme is secure (grpcs/https). Kept separately from - # tls_verify, which only controls certificate verification: tls_verify is - # False both for an insecure grpc:// target and for a grpcs:// target with - # verification disabled, so it cannot by itself tell the auth endpoint - # which scheme to use. - self._secure = urlparse(target).scheme in ("grpcs", "https") + self._target, self._path, self._is_plaintext, self._tls_verify = parse_target( + target + ) + if skip_tls_verify: + self._tls_verify = False # Load certificates once if needed self._certificates = ( @@ -382,37 +496,20 @@ def __init__( f"{self._name}/{self._version} {self._app_name}/{self._app_version}" ) - proxy_url = _get_grpc_proxy_url(self._target, self._tls_verify) + use_tls = not self._is_plaintext + proxy_url = _get_grpc_proxy_url(self._target, use_tls) if proxy_url: channel_opts.append(("grpc.http_proxy", proxy_url)) _LOGGER.debug(f"Configured gRPC proxy: {proxy_url}") - channel_opts = tuple(channel_opts) - - # Channel creation logic - if self._tls_verify: - credentials = ( - grpc.ssl_channel_credentials(root_certificates=self._certificates) - if self._certificates - else grpc.ssl_channel_credentials() - ) - - _LOGGER.debug( - f"Setting up gRPC secure channel with " - f"{'custom certificates' if self._certificates else 'system certificates'}" - f"{' via proxy' if proxy_url else ''}" - ) - self._channel = grpc.secure_channel( - self._target, - credentials, - options=channel_opts, - ) - else: - _LOGGER.debug("Setting up gRPC insecure channel") - self._channel = grpc.insecure_channel( - target=self._target, - options=channel_opts, - ) + self._channel = _open_grpc_channel( + self._target, + is_plaintext=self._is_plaintext, + tls_verify=self._tls_verify, + certificates=self._certificates, + channel_options=tuple(channel_opts), + proxy_url=proxy_url, + ) channel = self._channel @@ -542,6 +639,7 @@ def _authenticate(self, scope: str): authentication_client = _DiodeAuthentication( self._target, self._path, + self._is_plaintext, self._tls_verify, self._client_id, self._client_secret, @@ -553,7 +651,6 @@ def _authenticate(self, scope: str): self._certificates, self._cert_file, max_retries=self._max_auth_retries, - secure=self._secure, ) access_token = authentication_client.authenticate() self._metadata = list( @@ -652,6 +749,7 @@ def __init__( timeout: float = 10.0, metadata: dict[str, str] | Iterable[tuple[str, str]] | None = None, cert_file: str | None = None, + skip_tls_verify: bool = False, ): """Initiate a new Diode OTLP client.""" log_level = os.getenv(_DIODE_SDK_LOG_LEVEL_ENVVAR_NAME, "INFO").upper() @@ -663,7 +761,11 @@ def __init__( self._python_version = platform.python_version() self._timeout = timeout - self._target, self._path, self._tls_verify = parse_target(target) + self._target, self._path, self._is_plaintext, self._tls_verify = parse_target( + target + ) + if skip_tls_verify: + self._tls_verify = False self._cert_file = _get_optional_config_value( _DIODE_CERT_FILE_ENVVAR_NAME, cert_file ) @@ -677,41 +779,21 @@ def __init__( f"{self._name}/{self._version} {self._app_name}/{self._app_version}" ) - proxy_url = _get_grpc_proxy_url(self._target, self._tls_verify) + use_tls = not self._is_plaintext + proxy_url = _get_grpc_proxy_url(self._target, use_tls) if proxy_url: channel_opts.append(("grpc.http_proxy", proxy_url)) - # Extract hostname for SSL target name override - target_host = self._target.split(":")[0] - channel_opts.append(("grpc.ssl_target_name_override", target_host)) _LOGGER.debug(f"Configured gRPC proxy: {proxy_url}") - _LOGGER.debug(f"SSL target name override: {target_host}") - - channel_opts = tuple(channel_opts) - # Channel creation logic - if self._tls_verify: - credentials = ( - grpc.ssl_channel_credentials(root_certificates=self._certificates) - if self._certificates - else grpc.ssl_channel_credentials() - ) - - _LOGGER.debug( - f"Setting up gRPC secure channel with " - f"{'custom certificates' if self._certificates else 'system certificates'}" - f"{' via proxy' if proxy_url else ''}" - ) - base_channel = grpc.secure_channel( - self._target, - credentials, - options=channel_opts, - ) - else: - _LOGGER.debug("Setting up gRPC insecure channel") - base_channel = grpc.insecure_channel( - target=self._target, - options=channel_opts, - ) + base_channel = _open_grpc_channel( + self._target, + is_plaintext=self._is_plaintext, + tls_verify=self._tls_verify, + certificates=self._certificates, + channel_options=tuple(channel_opts), + proxy_url=proxy_url, + proxy_ssl_target_name_override=True, + ) self._base_channel = base_channel channel = base_channel @@ -763,6 +845,11 @@ def target(self) -> str: """Retrieve the export target.""" return self._target + @property + def tls_verify(self) -> bool: + """Retrieve whether TLS certificate verification is enabled.""" + return self._tls_verify + def __enter__(self): """Enter the runtime context.""" return self @@ -935,6 +1022,7 @@ def __init__( self, target: str, path: str, + is_plaintext: bool, tls_verify: bool, client_id: str, client_secret: str, @@ -949,11 +1037,10 @@ def __init__( initial_retry_delay: float | None = None, max_retry_delay: float | None = None, sleep: Callable[[float], None] | None = None, - secure: bool = True, ): self._target = target + self._is_plaintext = is_plaintext self._tls_verify = tls_verify - self._secure = secure self._client_id = client_id self._client_secret = client_secret self._path = path @@ -1076,7 +1163,7 @@ def _get_full_auth_url(self) -> str: via the session's verify setting. This keeps an insecure grpc:// target on HTTP even when DIODE_SKIP_TLS_VERIFY is set. """ - scheme = "https" if self._secure else "http" + scheme = "http" if self._is_plaintext else "https" path = self._path.rstrip("/") if self._path else "" return f"{scheme}://{self._target}{path}/auth/token" diff --git a/tests/test_client.py b/tests/test_client.py index a89457f..566a27d 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -38,6 +38,11 @@ from netboxlabs.diode.sdk.ingester import Entity from netboxlabs.diode.sdk.version import version_semver +_MOCK_PEER_CERT = ( + b"-----BEGIN CERTIFICATE-----\nTEST\n-----END CERTIFICATE-----\n", + "example.com", +) + def test_init(mock_diode_authentication): """Check we can initiate a client configuration.""" @@ -122,48 +127,69 @@ def test_parse_target_handles_ftp_prefix(): def test_parse_target_parses_authority_correctly(): """Check that parse_target parses the authority correctly.""" - authority, path, tls_verify = parse_target("grpc://localhost:8081") + authority, path, is_plaintext, tls_verify = parse_target("grpc://localhost:8081") assert authority == "localhost:8081" assert path == "" + assert is_plaintext is True assert tls_verify is False def test_parse_target_adds_default_port_if_missing(): """Check that parse_target adds the default port if missing.""" - authority, _, _ = parse_target("grpc://localhost") + authority, _, _, _ = parse_target("grpc://localhost") assert authority == "localhost:80" - authority, _, _ = parse_target("http://localhost") + authority, _, _, _ = parse_target("http://localhost") assert authority == "localhost:80" - authority, _, _ = parse_target("grpcs://localhost") + authority, _, _, _ = parse_target("grpcs://localhost") assert authority == "localhost:443" - authority, _, _ = parse_target("https://localhost") + authority, _, _, _ = parse_target("https://localhost") assert authority == "localhost:443" def test_parse_target_parses_path_correctly(): """Check that parse_target parses the path correctly.""" - _, path, _ = parse_target("grpc://localhost:8081/my/path") + _, path, _, _ = parse_target("grpc://localhost:8081/my/path") assert path == "/my/path" def test_parse_target_handles_no_path(): """Check that parse_target handles no path.""" - _, path, _ = parse_target("grpc://localhost:8081") + _, path, _, _ = parse_target("grpc://localhost:8081") assert path == "" -def test_parse_target_parses_tls_verify_correctly(): - """Check that parse_target parses tls_verify correctly.""" - _, _, tls_verify = parse_target("grpc://localhost:8081") +def test_parse_target_parses_plaintext_and_tls_verify(): + """Check that parse_target splits scheme from verification.""" + _, _, is_plaintext, tls_verify = parse_target("grpc://localhost:8081") + assert is_plaintext is True assert tls_verify is False - _, _, tls_verify = parse_target("http://localhost:8081") + _, _, is_plaintext, tls_verify = parse_target("http://localhost:8081") + assert is_plaintext is True assert tls_verify is False - _, _, tls_verify = parse_target("grpcs://localhost:8081") + _, _, is_plaintext, tls_verify = parse_target("grpcs://localhost:8081") + assert is_plaintext is False assert tls_verify is True - _, _, tls_verify = parse_target("https://localhost:8081") + _, _, is_plaintext, tls_verify = parse_target("https://localhost:8081") + assert is_plaintext is False assert tls_verify is True +def test_parse_target_skip_tls_env_on_secure_scheme(monkeypatch): + """DIODE_SKIP_TLS_VERIFY disables verification but keeps TLS for grpcs://.""" + monkeypatch.setenv("DIODE_SKIP_TLS_VERIFY", "true") + _, _, is_plaintext, tls_verify = parse_target("grpcs://localhost:8081") + assert is_plaintext is False + assert tls_verify is False + + +def test_parse_target_skip_tls_env_ignored_on_plaintext(monkeypatch): + """Plaintext grpc:// stays plaintext when skip-verify is set.""" + monkeypatch.setenv("DIODE_SKIP_TLS_VERIFY", "true") + _, _, is_plaintext, tls_verify = parse_target("grpc://localhost:8081") + assert is_plaintext is True + assert tls_verify is False + + def test_get_sentry_dsn_returns_env_var_when_no_input(): """Check that _get_sentry_dsn returns the env var when no input is provided.""" os.environ[_DIODE_SENTRY_DSN_ENVVAR_NAME] = "env_var_dsn" @@ -602,6 +628,7 @@ def test_diode_authentication_success(mock_diode_authentication): auth = _DiodeAuthentication( target="localhost:8081", path="/diode", + is_plaintext=True, tls_verify=False, client_id="test_client_id", client_secret="test_client_secret", @@ -627,6 +654,7 @@ def test_diode_authentication_failure(mock_diode_authentication): auth = _DiodeAuthentication( target="localhost:8081", path="/diode", + is_plaintext=True, tls_verify=False, client_id="test_client_id", client_secret="test_client_secret", @@ -664,8 +692,8 @@ def test_diode_authentication_url_with_path(mock_diode_authentication, path): auth = _DiodeAuthentication( target="localhost:8081", path=path, + is_plaintext=True, tls_verify=False, - secure=False, client_id="test_client_id", client_secret="test_client_secret", scope="diode:ingest", @@ -692,23 +720,18 @@ def test_diode_authentication_url_with_path(mock_diode_authentication, path): @pytest.mark.parametrize( - ("secure", "tls_verify", "skip_tls_env", "expected_scheme"), + ("is_plaintext", "tls_verify", "skip_tls_env", "expected_scheme"), [ - # (scheme secure?, verify certs?, DIODE_SKIP_TLS_VERIFY, expected auth scheme) - (False, False, None, "http"), # grpc:// - # grpc:// + skip must stay HTTP (the #101 fix), not flip to HTTPS. - (False, False, "true", "http"), - (True, True, None, "https"), # grpcs:// - # grpcs:// + skip is preserved: HTTPS with cert verification disabled. - (True, False, "true", "https"), - # Independence guards: the scheme follows `secure`, never `tls_verify`. - # A scheme = self._tls_verify regression would fail both of these. - (True, False, None, "https"), - (False, True, None, "http"), + (True, False, None, "http"), + (True, False, "true", "http"), + (False, True, None, "https"), + (False, False, "true", "https"), + (False, False, None, "https"), + (True, True, None, "http"), ], ) def test_auth_url_scheme_follows_target( - mock_diode_authentication, monkeypatch, secure, tls_verify, skip_tls_env, expected_scheme + mock_diode_authentication, monkeypatch, is_plaintext, tls_verify, skip_tls_env, expected_scheme ): """Auth endpoint scheme follows the target scheme, independent of tls_verify/env.""" if skip_tls_env is None: @@ -719,6 +742,7 @@ def test_auth_url_scheme_follows_target( auth = _DiodeAuthentication( target="host:8080", path="", + is_plaintext=is_plaintext, tls_verify=tls_verify, client_id="test_client_id", client_secret="test_client_secret", @@ -727,36 +751,37 @@ def test_auth_url_scheme_follows_target( sdk_version="0.1.0", app_name="test-app", app_version="1.0.0", - secure=secure, ) assert auth._get_full_auth_url() == f"{expected_scheme}://host:8080/auth/token" @pytest.mark.parametrize( - ("target", "expected_secure"), + ("target", "expected_is_plaintext"), [ - ("grpc://localhost:8081", False), - ("http://localhost:8081", False), - ("grpcs://localhost:8081", True), - ("https://localhost:8081", True), + ("grpc://localhost:8081", True), + ("http://localhost:8081", True), + ("grpcs://localhost:8081", False), + ("https://localhost:8081", False), ], ) -def test_client_secure_flag_follows_target_scheme( - mock_diode_authentication, monkeypatch, target, expected_secure +def test_client_is_plaintext_follows_target_scheme( + mock_diode_authentication, monkeypatch, target, expected_is_plaintext ): - """DiodeClient._secure reflects the target scheme even with DIODE_SKIP_TLS_VERIFY set.""" + """DiodeClient._is_plaintext reflects the target scheme even with skip-verify set.""" monkeypatch.setenv("DIODE_SKIP_TLS_VERIFY", "true") - client = DiodeClient( - target=target, - app_name="my-producer", - app_version="0.0.1", - client_id="abcde", - client_secret="123456", - ) - assert client._secure is expected_secure - # tls_verify is driven off skip-verify and is False here regardless of scheme, - # which is exactly why it cannot be reused to pick the auth scheme. + with patch( + "netboxlabs.diode.sdk.client._fetch_peer_leaf_certificate", + return_value=(b"-----BEGIN CERTIFICATE-----\nTEST\n-----END CERTIFICATE-----\n", "localhost"), + ): + client = DiodeClient( + target=target, + app_name="my-producer", + app_version="0.0.1", + client_id="abcde", + client_secret="123456", + ) + assert client._is_plaintext is expected_is_plaintext assert client.tls_verify is False @@ -765,6 +790,7 @@ def test_diode_authentication_request_exception(mock_diode_authentication): auth = _DiodeAuthentication( target="localhost:8081", path="/diode", + is_plaintext=True, tls_verify=False, client_id="test_client_id", client_secret="test_client_secret", @@ -850,6 +876,7 @@ def test_diode_authentication_retries_retriable_status(mock_diode_authentication auth = _DiodeAuthentication( target="localhost:8081", path="/diode", + is_plaintext=True, tls_verify=False, client_id="test_client_id", client_secret="test_client_secret", @@ -888,6 +915,7 @@ def test_diode_authentication_fails_fast_on_401(mock_diode_authentication): auth = _DiodeAuthentication( target="localhost:8081", path="/diode", + is_plaintext=True, tls_verify=False, client_id="test_client_id", client_secret="test_client_secret", @@ -916,6 +944,7 @@ def test_diode_authentication_exhausts_retries(mock_diode_authentication): auth = _DiodeAuthentication( target="localhost:8081", path="/diode", + is_plaintext=True, tls_verify=False, client_id="test_client_id", client_secret="test_client_secret", @@ -1114,6 +1143,34 @@ def test_otlp_client_grpcs_uses_secure_channel(): base_channel.close.assert_called_once() +def test_otlp_client_grpcs_skip_tls_uses_secure_channel(): + """DiodeOTLPClient keeps TLS when skip-verify is set.""" + with ( + patch( + "netboxlabs.diode.sdk.client._fetch_peer_leaf_certificate", + return_value=_MOCK_PEER_CERT, + ), + patch("netboxlabs.diode.sdk.client.grpc.secure_channel") as mock_secure_channel, + patch("netboxlabs.diode.sdk.client.grpc.insecure_channel") as mock_insecure_channel, + patch( + "netboxlabs.diode.sdk.client.grpc.intercept_channel", + return_value=mock.Mock(), + ), + patch("netboxlabs.diode.sdk.client.logs_service_pb2_grpc.LogsServiceStub"), + ): + mock_secure_channel.return_value = mock.Mock() + client = DiodeOTLPClient( + target="grpcs://collector.example:4317", + app_name="orb-producer", + app_version="1.2.3", + skip_tls_verify=True, + ) + assert client.tls_verify is False + mock_secure_channel.assert_called_once() + mock_insecure_channel.assert_not_called() + client.close() + + def test_otlp_insecure_channel_options_exclude_diode_keepalive(): """OTLP targets arbitrary collectors; only user-agent is forced (Codex/OBS-2873).""" with ( @@ -1148,6 +1205,7 @@ def test_diode_authentication_with_custom_certificates(): auth = _DiodeAuthentication( target="example.com:443", path="/api/v1", + is_plaintext=False, tls_verify=True, client_id="test_client", client_secret="test_secret", @@ -1418,97 +1476,108 @@ def test_client_without_cert_file_uses_default_certs(mock_diode_authentication): mock_secure_channel.assert_called_once() -def test_should_verify_tls_with_different_schemes(): - """Test _should_verify_tls with different URL schemes.""" - from netboxlabs.diode.sdk.client import ( - _DIODE_SKIP_TLS_VERIFY_ENVVAR_NAME, - _should_verify_tls, - ) +def test_skip_tls_verify_from_env(monkeypatch): + """Test DIODE_SKIP_TLS_VERIFY truthy values match Go SDK.""" + from netboxlabs.diode.sdk.client import _skip_tls_verify_from_env - # Clear environment variable to avoid interference - if _DIODE_SKIP_TLS_VERIFY_ENVVAR_NAME in os.environ: - del os.environ[_DIODE_SKIP_TLS_VERIFY_ENVVAR_NAME] + for skip_value in ["true", "True", "TRUE", "1", "yes", "on"]: + monkeypatch.setenv("DIODE_SKIP_TLS_VERIFY", skip_value) + assert _skip_tls_verify_from_env() is True - assert _should_verify_tls("grpc") is False # insecure scheme - assert _should_verify_tls("http") is False # insecure scheme - assert _should_verify_tls("grpcs") is True # secure scheme - assert _should_verify_tls("https") is True # secure scheme + for verify_value in ["false", "0", "no", "off", "", "random"]: + monkeypatch.setenv("DIODE_SKIP_TLS_VERIFY", verify_value) + assert _skip_tls_verify_from_env() is False -def test_should_verify_tls_with_skip_env_var(): - """Test _should_verify_tls with DIODE_SKIP_TLS_VERIFY environment variable.""" - from netboxlabs.diode.sdk.client import ( - _DIODE_SKIP_TLS_VERIFY_ENVVAR_NAME, - _should_verify_tls, - ) +def test_tls_server_name_from_cert_pem_prefers_san(): + """Extract DNS SAN for grpc.ssl_target_name_override when skipping verify.""" + from netboxlabs.diode.sdk.client import _tls_server_name_from_cert_pem - original_env = os.environ.get(_DIODE_SKIP_TLS_VERIFY_ENVVAR_NAME) + pem = b"-----BEGIN CERTIFICATE-----\nTEST\n-----END CERTIFICATE-----\n" + with patch( + "netboxlabs.diode.sdk.client.ssl._ssl._test_decode_cert", + return_value={"subjectAltName": [("DNS", "traefik.local")]}, + ): + assert _tls_server_name_from_cert_pem(pem) == "traefik.local" - try: - # Test truthy values that should skip TLS verification - for skip_value in ["true", "True", "TRUE", "1", "yes", "on"]: - os.environ[_DIODE_SKIP_TLS_VERIFY_ENVVAR_NAME] = skip_value - assert ( - _should_verify_tls("grpcs") is False - ) # Should skip even for secure schemes - - # Test falsy values that should NOT skip TLS verification - for verify_value in ["false", "0", "no", "off", "", "random"]: - os.environ[_DIODE_SKIP_TLS_VERIFY_ENVVAR_NAME] = verify_value - assert ( - _should_verify_tls("grpcs") is True - ) # Should verify for secure schemes - finally: - # Clean up environment variable - if original_env is not None: - os.environ[_DIODE_SKIP_TLS_VERIFY_ENVVAR_NAME] = original_env - else: - if _DIODE_SKIP_TLS_VERIFY_ENVVAR_NAME in os.environ: - del os.environ[_DIODE_SKIP_TLS_VERIFY_ENVVAR_NAME] +def test_tls_server_name_from_cert_pem_falls_back_to_cn(): + """Use commonName when the certificate has no DNS SAN.""" + from netboxlabs.diode.sdk.client import _tls_server_name_from_cert_pem + + pem = b"-----BEGIN CERTIFICATE-----\nTEST\n-----END CERTIFICATE-----\n" + with patch( + "netboxlabs.diode.sdk.client.ssl._ssl._test_decode_cert", + return_value={"subject": [[("commonName", "TRAEFIK")]]}, + ): + assert _tls_server_name_from_cert_pem(pem) == "TRAEFIK" def test_client_with_skip_tls_verify_env_var(mock_diode_authentication): - """Test DiodeClient with DIODE_SKIP_TLS_VERIFY environment variable.""" + """grpcs:// with DIODE_SKIP_TLS_VERIFY keeps a secure channel.""" from netboxlabs.diode.sdk.client import _DIODE_SKIP_TLS_VERIFY_ENVVAR_NAME original_env = os.environ.get(_DIODE_SKIP_TLS_VERIFY_ENVVAR_NAME) try: - # Set environment variable to skip TLS verification os.environ[_DIODE_SKIP_TLS_VERIFY_ENVVAR_NAME] = "true" - with mock.patch("grpc.insecure_channel") as mock_insecure_channel: + with ( + patch( + "netboxlabs.diode.sdk.client._fetch_peer_leaf_certificate", + return_value=_MOCK_PEER_CERT, + ), + mock.patch("grpc.insecure_channel") as mock_insecure_channel, + mock.patch("grpc.secure_channel") as mock_secure_channel, + ): client = DiodeClient( - target="grpcs://localhost:8081", # Note: grpcs:// but TLS should be skipped + target="grpcs://localhost:8081", app_name="my-producer", app_version="0.0.1", client_id="abcde", client_secret="123456", ) - # Should skip TLS verification due to environment variable assert client.tls_verify is False - - # Should use insecure channel even with grpcs:// - mock_insecure_channel.assert_called_once() + mock_insecure_channel.assert_not_called() + mock_secure_channel.assert_called_once() finally: - # Clean up environment variable if original_env is not None: os.environ[_DIODE_SKIP_TLS_VERIFY_ENVVAR_NAME] = original_env else: - if _DIODE_SKIP_TLS_VERIFY_ENVVAR_NAME in os.environ: - del os.environ[_DIODE_SKIP_TLS_VERIFY_ENVVAR_NAME] + os.environ.pop(_DIODE_SKIP_TLS_VERIFY_ENVVAR_NAME, None) + + +def test_client_with_skip_tls_verify_constructor(mock_diode_authentication): + """skip_tls_verify=True on DiodeClient uses secure_channel for grpcs://.""" + with ( + patch( + "netboxlabs.diode.sdk.client._fetch_peer_leaf_certificate", + return_value=_MOCK_PEER_CERT, + ), + mock.patch("grpc.secure_channel") as mock_secure_channel, + mock.patch("grpc.insecure_channel") as mock_insecure_channel, + ): + client = DiodeClient( + target="grpcs://localhost:8081", + app_name="my-producer", + app_version="0.0.1", + client_id="abcde", + client_secret="123456", + skip_tls_verify=True, + ) + assert client.tls_verify is False + mock_secure_channel.assert_called_once() + mock_insecure_channel.assert_not_called() def test_client_cert_file_with_skip_tls_verify_env_var( mock_diode_authentication, tmp_path ): - """Test cert_file parameter with DIODE_SKIP_TLS_VERIFY environment variable.""" + """cert_file with skip-verify still opens a secure channel.""" from netboxlabs.diode.sdk.client import _DIODE_SKIP_TLS_VERIFY_ENVVAR_NAME - # Create a dummy certificate file cert_content = ( b"-----BEGIN CERTIFICATE-----\nTEST CERT\n-----END CERTIFICATE-----\n" ) @@ -1518,10 +1587,16 @@ def test_client_cert_file_with_skip_tls_verify_env_var( original_skip_env = os.environ.get(_DIODE_SKIP_TLS_VERIFY_ENVVAR_NAME) try: - # Set environment variable to skip TLS verification os.environ[_DIODE_SKIP_TLS_VERIFY_ENVVAR_NAME] = "true" - with mock.patch("grpc.insecure_channel") as mock_insecure_channel: + with ( + patch( + "netboxlabs.diode.sdk.client._fetch_peer_leaf_certificate", + return_value=_MOCK_PEER_CERT, + ), + mock.patch("grpc.insecure_channel") as mock_insecure_channel, + mock.patch("grpc.secure_channel") as mock_secure_channel, + ): client = DiodeClient( target="grpcs://localhost:8081", app_name="my-producer", @@ -1531,22 +1606,36 @@ def test_client_cert_file_with_skip_tls_verify_env_var( cert_file=str(cert_file), ) - # Should respect DIODE_SKIP_TLS_VERIFY=true even with cert_file assert client.tls_verify is False - - # Should use insecure channel due to environment variable - mock_insecure_channel.assert_called_once() - - # Certificate should still be loaded for potential use + mock_insecure_channel.assert_not_called() + mock_secure_channel.assert_called_once() assert client._certificates == cert_content finally: - # Clean up environment variable if original_skip_env is not None: os.environ[_DIODE_SKIP_TLS_VERIFY_ENVVAR_NAME] = original_skip_env else: - if _DIODE_SKIP_TLS_VERIFY_ENVVAR_NAME in os.environ: - del os.environ[_DIODE_SKIP_TLS_VERIFY_ENVVAR_NAME] + os.environ.pop(_DIODE_SKIP_TLS_VERIFY_ENVVAR_NAME, None) + + +def test_auth_session_verify_false_when_skip_tls(mock_diode_authentication): + """Token fetch uses HTTPS with verify=False when tls_verify is disabled.""" + auth = _DiodeAuthentication( + target="localhost:443", + path="", + is_plaintext=False, + tls_verify=False, + client_id="test_client_id", + client_secret="test_client_secret", + scope="diode:ingest", + sdk_name="diode-sdk-python", + sdk_version="0.1.0", + app_name="test-app", + app_version="1.0.0", + ) + session = mock.Mock() + auth._configure_auth_session(session) + assert session.verify is False def test_certificate_loading_efficiency(tmp_path): @@ -2072,14 +2161,21 @@ def test_diode_client_configures_proxy_option(mock_diode_authentication): del os.environ["HTTP_PROXY"] -def test_diode_client_uses_insecure_channel_with_proxy_when_skip_tls( +def test_diode_client_uses_secure_channel_with_proxy_when_skip_tls( mock_diode_authentication, ): - """Test DiodeClient uses insecure channel with proxy when SKIP_TLS_VERIFY is set.""" + """grpcs:// with proxy and skip-verify stays on a secure channel.""" os.environ["HTTP_PROXY"] = "http://proxy.example.com:8080" os.environ["DIODE_SKIP_TLS_VERIFY"] = "true" try: - with mock.patch("grpc.insecure_channel") as mock_insecure_channel: + with ( + patch( + "netboxlabs.diode.sdk.client._fetch_peer_leaf_certificate", + return_value=_MOCK_PEER_CERT, + ), + mock.patch("grpc.insecure_channel") as mock_insecure_channel, + mock.patch("grpc.secure_channel") as mock_secure_channel, + ): DiodeClient( target="grpcs://example.com:443", app_name="my-producer", @@ -2088,12 +2184,11 @@ def test_diode_client_uses_insecure_channel_with_proxy_when_skip_tls( client_secret="123456", ) - # Should use insecure channel when SKIP_TLS_VERIFY is set, even with proxy - mock_insecure_channel.assert_called_once() - _, kwargs = mock_insecure_channel.call_args + mock_insecure_channel.assert_not_called() + mock_secure_channel.assert_called_once() + _, kwargs = mock_secure_channel.call_args options = kwargs["options"] - # Verify proxy option is set proxy_option = next( (opt for opt in options if opt[0] == "grpc.http_proxy"), None )