From f4da2d01dab86f97028653f646d1b0e8f0cc4b4b Mon Sep 17 00:00:00 2001 From: Arseniy Dugin Date: Tue, 14 Jul 2026 10:24:19 +0000 Subject: [PATCH] Add configurable TLS verification for all HTTP transports Introduce a single TLS verification contract shared by the requests client, the httpx schema/operation transport, and the dynamic-discovery client. TANGLE_API_CA_BUNDLE verifies against a custom CA bundle and TANGLE_API_VERIFY_TLS toggles verification, with verification enabled by default. Precedence is explicit argument, CA bundle, verify flag, then the secure default; unset settings preserve requests' environment and caller-supplied session behavior. --- README.md | 51 ++ packages/tangle-cli/src/tangle_cli/api_cli.py | 16 +- .../tangle-cli/src/tangle_cli/api_schema.py | 8 + .../src/tangle_cli/api_transport.py | 126 +++ packages/tangle-cli/src/tangle_cli/cli.py | 69 +- .../tangle-cli/src/tangle_cli/cli_options.py | 24 + packages/tangle-cli/src/tangle_cli/client.py | 10 +- .../tangle_cli/dynamic_discovery_client.py | 15 + tests/test_packaging.py | 7 +- tests/test_tls_verification.py | 748 ++++++++++++++++++ 10 files changed, 1066 insertions(+), 8 deletions(-) create mode 100644 tests/test_tls_verification.py diff --git a/README.md b/README.md index db3076e..20e44d8 100644 --- a/README.md +++ b/README.md @@ -60,6 +60,57 @@ API-backed commands commonly accept these options. Explicit CLI options win over | `--config` | YAML/JSON defaults. Many commands accept a single object, a list of objects, or `_defaults` + `configs`. | | `--log-type` | SDK progress logs: `console`, `none`, or `file`. Logs go to stderr or a temp log file so structured stdout stays parseable. | | `TANGLE_VERBOSE=1` | Redacted HTTP request/response diagnostics only. This is separate from normal progress logging. | +| `--ca-bundle` | Global CLI flag: path to a PEM CA bundle used as the TLS trust store for every transport. Overrides `TANGLE_API_CA_BUNDLE`. Place before the subcommand. | +| `--verify-tls` / `--no-verify-tls` | Global CLI flag: enable or disable TLS verification for every transport. Overrides `TANGLE_API_VERIFY_TLS`. `--no-verify-tls` is local-development only. Place before the subcommand. | +| `TANGLE_API_CA_BUNDLE` | Path to a PEM CA bundle used to verify TLS for every transport. Use this to trust a private or corporate CA without disabling verification. | +| `TANGLE_API_VERIFY_TLS` | TLS verification toggle. Values `0`, `false`, or `no` (case/space-insensitive) disable verification; any other nonempty value keeps it on. | + +### TLS verification + +TLS certificate verification is enabled by default for all HTTP transports (schema +fetches, `tangle api` calls, and the programmatic clients). The effective setting is +resolved with the following precedence, highest to lowest: + +1. An explicit `verify=` argument to the Python clients (a `bool` or a path to a CA bundle). +2. The global CLI flags `--ca-bundle` / `--verify-tls` / `--no-verify-tls`. +3. `TANGLE_API_CA_BUNDLE` — verify against the given CA bundle. +4. `TANGLE_API_VERIFY_TLS` — enable or disable verification. +5. The secure default: verification enabled against the system trust store. + +The global CLI flags are true root options that apply to every command — the static +`tangle sdk ...` clients, the dynamic `tangle api ...` commands, and `tangle api refresh`. +Place them **before** the subcommand, for example `tangle --ca-bundle ca.pem api ...` or +`tangle --no-verify-tls sdk ...`. They are honored even by the dynamic OpenAPI schema +discovery that runs before command dispatch. A defaulted (absent) flag does not override +the environment: when a flag is not supplied, the `TANGLE_API_*` variables and the standard +`REQUESTS_CA_BUNDLE` / `CURL_CA_BUNDLE` handling still apply. `--ca-bundle` combined with an +explicit `--no-verify-tls` is contradictory and fails fast before any request; `--ca-bundle` +with `--verify-tls` is redundant but accepted. + +If both env vars are set, `TANGLE_API_CA_BUNDLE` wins and TLS stays verified against the +bundle. Empty values are treated as unset. A `--ca-bundle` or `TANGLE_API_CA_BUNDLE` that +does not point to an existing file fails fast with an actionable error before any request is +made. When no Tangle-specific setting is provided, the standard `REQUESTS_CA_BUNDLE` / +`CURL_CA_BUNDLE` handling and any caller-supplied `requests.Session.verify` are left +untouched. + +For a private CA, prefer `--ca-bundle` / `TANGLE_API_CA_BUNDLE` over disabling verification: +it keeps certificates verified against a trusted root. `--no-verify-tls` / +`TANGLE_API_VERIFY_TLS=0` disables verification entirely and is intended for local +development only — never use it against production endpoints. + +```bash +# Trust a private CA with the global flag (recommended for internal/self-hosted APIs) +uv run tangle --ca-bundle /etc/ssl/private-ca.pem \ + api refresh --base-url https://internal.example + +# Or via environment variable +TANGLE_API_CA_BUNDLE=/etc/ssl/private-ca.pem \ + uv run tangle api refresh --base-url https://internal.example + +# Disable verification (local development only) +uv run tangle --no-verify-tls api refresh --base-url https://localhost:8443 +``` Examples for protected APIs: diff --git a/packages/tangle-cli/src/tangle_cli/api_cli.py b/packages/tangle-cli/src/tangle_cli/api_cli.py index 4a80863..db98df8 100644 --- a/packages/tangle-cli/src/tangle_cli/api_cli.py +++ b/packages/tangle-cli/src/tangle_cli/api_cli.py @@ -640,10 +640,22 @@ def _argv_dispatches_dynamic_command(argv: list[str]) -> bool: def _api_argv_tail(argv: list[str]) -> list[str] | None: - """Return args after the root `api` command, or None for non-API invocations.""" + """Return args after the root `api` command, or None for non-API invocations. + + Global TLS flags (``--ca-bundle``/``--verify-tls``/``--no-verify-tls``) may + precede the subcommand, so they are skipped before locating `api`. + """ args = list(argv[1:]) - for index, arg in enumerate(args): + index = 0 + while index < len(args): + arg = args[index] + if arg == "--ca-bundle" and index + 1 < len(args): + index += 2 + continue + if arg.startswith("--ca-bundle=") or arg in {"--verify-tls", "--no-verify-tls"}: + index += 1 + continue if arg == "--": if index + 1 < len(args) and args[index + 1] == "api": return args[index + 2 :] diff --git a/packages/tangle-cli/src/tangle_cli/api_schema.py b/packages/tangle-cli/src/tangle_cli/api_schema.py index 5475ea8..72ed2cc 100644 --- a/packages/tangle-cli/src/tangle_cli/api_schema.py +++ b/packages/tangle-cli/src/tangle_cli/api_schema.py @@ -18,7 +18,9 @@ _normalize_base_url, _openapi_url, _request_headers, + _VERIFY_UNSET, default_base_url, + httpx_verify, ) SUPPORTED_METHODS = {"get", "post", "put", "patch", "delete"} @@ -122,6 +124,7 @@ def fetch_schema( auth_header: str | None = None, headers: dict[str, str] | None = None, include_env_credentials: bool = True, + verify: Any = _VERIFY_UNSET, ) -> dict[str, Any]: """Fetch ``/openapi.json``, applying bearer and custom auth headers.""" @@ -136,6 +139,7 @@ def fetch_schema( include_env_credentials=include_env_credentials, ), timeout=DEFAULT_TIMEOUT_SECONDS, + verify=httpx_verify(verify), ) response.raise_for_status() payload = response.text @@ -152,6 +156,7 @@ def refresh_schema( auth_header: str | None = None, headers: dict[str, str] | None = None, include_env_credentials: bool = True, + verify: Any = _VERIFY_UNSET, ) -> tuple[dict[str, Any], Path]: """Fetch and cache the latest schema for a backend.""" @@ -163,6 +168,7 @@ def refresh_schema( auth_header, headers, include_env_credentials=include_env_credentials, + verify=verify, ) path = write_cached_schema(schema, base_url) return schema, path @@ -175,6 +181,7 @@ def load_or_fetch_schema( auth_header: str | None = None, headers: dict[str, str] | None = None, include_env_credentials: bool = True, + verify: Any = _VERIFY_UNSET, ) -> dict[str, Any]: """Use a cached schema when available, otherwise fetch once and cache it.""" @@ -188,6 +195,7 @@ def load_or_fetch_schema( auth_header, headers, include_env_credentials=include_env_credentials, + verify=verify, ) return schema diff --git a/packages/tangle-cli/src/tangle_cli/api_transport.py b/packages/tangle-cli/src/tangle_cli/api_transport.py index b128070..42c76f2 100644 --- a/packages/tangle-cli/src/tangle_cli/api_transport.py +++ b/packages/tangle-cli/src/tangle_cli/api_transport.py @@ -5,6 +5,7 @@ import json import os import re +import ssl import sys import urllib.parse from pathlib import Path @@ -16,6 +17,20 @@ DEFAULT_TIMEOUT_SECONDS = 30.0 _HEADER_NAME_RE = re.compile(r"^[!#$%&'*+.^_`|~0-9A-Za-z-]+$") _MISSING = object() + +# Canonical resolved TLS verification value: ``True``/``False`` or a CA bundle +# path. Both requests and httpx transports adapt this single contract. +VerifyValue = bool | str +VerifyArgument = bool | str | os.PathLike[str] | None +_VERIFY_UNSET = object() +_TLS_FALSE_VALUES = frozenset({"0", "false", "no"}) + +# Process-wide TLS override set from global CLI flags (``--ca-bundle`` / +# ``--verify-tls`` / ``--no-verify-tls``). It sits between an explicit +# ``verify=`` argument and the environment variables in :func:`resolve_verify`, +# so a single resolver serves every transport, including the schema discovery +# that runs before normal command dispatch. +_CLI_VERIFY_OVERRIDE: Any = _VERIFY_UNSET _SENSITIVE_HEADER_NAMES = {"authorization", "cloud-auth", "cookie", "x-api-key"} _SENSITIVE_KEY_RE = re.compile( r"(authorization|authentication|(^|[-_])auth($|[-_])|cloud[-_]?auth|cookie|x[-_]?api[-_]?key|token|secret|password|credential|pre[-_]?signed[-_]?url|signed[-_]?url)", @@ -40,6 +55,115 @@ def tangle_verbose_enabled() -> bool: return value.strip().lower() in {"1", "true", "yes", "on"} +def _parse_verify_flag(raw: str) -> bool: + """Interpret ``TANGLE_API_VERIFY_TLS``. + + Only the case/space-insensitive values ``0``, ``false``, and ``no`` disable + verification; any other nonempty value keeps it enabled. + """ + + return raw.strip().lower() not in _TLS_FALSE_VALUES + + +def _validate_ca_bundle(path: str, source: str) -> str: + """Return an existing CA bundle path or fail before any network request.""" + + candidate = Path(path).expanduser() + if not candidate.is_file(): + raise SystemExit( + f"{source} points to a CA bundle that does not exist: {path!r}. " + "Provide a path to an existing PEM file, or unset it to use the " + "system trust store." + ) + return str(candidate) + + +def _coerce_explicit_verify(value: bool | str | os.PathLike[str]) -> VerifyValue: + if isinstance(value, bool): + return value + if isinstance(value, (str, os.PathLike)): + return _validate_ca_bundle(os.fspath(value), "verify") + raise SystemExit("verify must be a bool or a path to a CA bundle file") + + +def configure_cli_verify( + ca_bundle: str | os.PathLike[str] | None = None, + verify_tls: bool | None = None, +) -> None: + """Install (or clear) the process-wide TLS override from global CLI flags. + + ``ca_bundle`` is a path to a PEM trust store; ``verify_tls`` is the tri-state + ``--verify-tls`` / ``--no-verify-tls`` flag where ``None`` means the flag was + not supplied. A ``--ca-bundle`` combined with an explicit + ``--no-verify-tls`` is contradictory and fails fast before any network + request. Passing neither clears any previously installed override. + + The resolved value is validated here so an invalid or missing CA bundle + fails before dynamic schema discovery. It is consulted by + :func:`resolve_verify` for every transport. + """ + + global _CLI_VERIFY_OVERRIDE + if ca_bundle is not None: + if verify_tls is False: + raise SystemExit( + "--ca-bundle cannot be combined with --no-verify-tls: a CA " + "bundle only takes effect when verification is enabled. Drop " + "one of the two flags." + ) + _CLI_VERIFY_OVERRIDE = _validate_ca_bundle(os.fspath(ca_bundle), "--ca-bundle") + return + if verify_tls is not None: + _CLI_VERIFY_OVERRIDE = bool(verify_tls) + return + _CLI_VERIFY_OVERRIDE = _VERIFY_UNSET + + +def resolve_verify(verify: Any = _VERIFY_UNSET) -> Any: + """Resolve the effective TLS verification setting. + + Precedence, highest to lowest: an explicit ``verify`` argument, the global + CLI override (``--ca-bundle`` / ``--verify-tls`` / ``--no-verify-tls``), a + nonempty ``TANGLE_API_CA_BUNDLE``, ``TANGLE_API_VERIFY_TLS``, then a secure + enabled default. When none of these are set, ``_VERIFY_UNSET`` is returned + so callers can preserve library and caller defaults (for example requests' + ``REQUESTS_CA_BUNDLE``/``CURL_CA_BUNDLE`` handling and a caller-supplied + ``Session.verify``). Empty environment values are treated as unset. + """ + + if verify is not _VERIFY_UNSET and verify is not None: + return _coerce_explicit_verify(verify) + if _CLI_VERIFY_OVERRIDE is not _VERIFY_UNSET: + return _CLI_VERIFY_OVERRIDE + ca_bundle = os.environ.get("TANGLE_API_CA_BUNDLE", "") + if ca_bundle.strip(): + return _validate_ca_bundle(ca_bundle.strip(), "TANGLE_API_CA_BUNDLE") + flag = os.environ.get("TANGLE_API_VERIFY_TLS", "") + if flag.strip(): + return _parse_verify_flag(flag) + return _VERIFY_UNSET + + +def resolve_verify_default(verify: Any = _VERIFY_UNSET) -> VerifyValue: + """Like :func:`resolve_verify`, but fall back to the secure ``True`` default.""" + + resolved = resolve_verify(verify) + return True if resolved is _VERIFY_UNSET else resolved + + +def httpx_verify(verify: Any = _VERIFY_UNSET) -> bool | ssl.SSLContext: + """Adapt the resolved verify value to what httpx expects. + + httpx 0.28 deprecates string CA-bundle paths, so a path is turned into an + :class:`ssl.SSLContext`; booleans pass through unchanged. + """ + + resolved = resolve_verify_default(verify) + if isinstance(resolved, bool): + return resolved + return ssl.create_default_context(cafile=resolved) + + def _redact_headers(headers: dict[str, Any] | None) -> dict[str, Any]: redacted: dict[str, Any] = {} for name, value in (headers or {}).items(): @@ -280,6 +404,7 @@ def request_operation( timeout: float = DEFAULT_TIMEOUT_SECONDS, allow_body_file_references: bool = False, include_env_credentials: bool = True, + verify: Any = _VERIFY_UNSET, ) -> httpx.Response: """Dispatch one normalized OpenAPI operation as an HTTP request. @@ -306,6 +431,7 @@ def request_operation( content=content, headers=request_headers, timeout=timeout, + verify=httpx_verify(verify), ) if tangle_verbose_enabled(): log_http_exchange( diff --git a/packages/tangle-cli/src/tangle_cli/cli.py b/packages/tangle-cli/src/tangle_cli/cli.py index e9c1fd4..ad9541c 100644 --- a/packages/tangle-cli/src/tangle_cli/cli.py +++ b/packages/tangle-cli/src/tangle_cli/cli.py @@ -1,4 +1,9 @@ -from cyclopts import App +from __future__ import annotations + +import sys +from typing import Annotated + +from cyclopts import App, Parameter from . import ( __version__, @@ -11,6 +16,8 @@ quickstart, secrets_cli, ) +from .api_transport import configure_cli_verify +from .cli_options import CaBundleOption, VerifyTlsOption def version() -> None: @@ -19,6 +26,41 @@ def version() -> None: print(__version__) +def _configure_tls_from_argv(argv: list[str]) -> None: + """Parse the global TLS flags and install the process-wide override. + + Runs before the `api` app is built so the override is in place for the + dynamic schema discovery that happens during command construction, ahead of + normal Cyclopts dispatch. Only tokens before the first subcommand are + considered; flag validation and conflict detection live in + :func:`configure_cli_verify`. + """ + + ca_bundle: str | None = None + verify_tls: bool | None = None + index = 1 + while index < len(argv): + arg = argv[index] + if arg == "--ca-bundle" and index + 1 < len(argv): + ca_bundle = argv[index + 1] + index += 2 + continue + if arg.startswith("--ca-bundle="): + ca_bundle = arg.split("=", 1)[1] + index += 1 + continue + if arg == "--verify-tls": + verify_tls = True + index += 1 + continue + if arg == "--no-verify-tls": + verify_tls = False + index += 1 + continue + break + configure_cli_verify(ca_bundle, verify_tls) + + def build_sdk_app() -> App: """Build the SDK command group.""" @@ -35,8 +77,15 @@ def build_sdk_app() -> App: return sdk_app -def build_app() -> App: - """Build the root CLI app lazily for the current invocation.""" +def build_app(argv: list[str] | None = None) -> App: + """Build the root CLI app lazily for the current invocation. + + Global TLS flags are parsed from *argv* (defaulting to ``sys.argv``) and + installed before the `api` app is built, because building it can trigger + dynamic OpenAPI schema discovery over the network. + """ + + _configure_tls_from_argv(sys.argv if argv is None else argv) app = App( help="CLI for Tangle, the open-source ML pipeline orchestration platform.", @@ -46,11 +95,23 @@ def build_app() -> App: app.command(quickstart.app) app.command(api_cli.build_app()) app.command(build_sdk_app()) + + @app.meta.default + def launcher( + *tokens: Annotated[str, Parameter(allow_leading_hyphen=True)], + ca_bundle: CaBundleOption = None, + verify_tls: VerifyTlsOption = None, + ) -> None: + """Apply global TLS options, then dispatch the requested command.""" + + configure_cli_verify(ca_bundle, verify_tls) + app(tokens) + return app def main() -> None: - build_app()() + build_app().meta() if __name__ == "__main__": diff --git a/packages/tangle-cli/src/tangle_cli/cli_options.py b/packages/tangle-cli/src/tangle_cli/cli_options.py index f6080ed..5edfbb2 100644 --- a/packages/tangle-cli/src/tangle_cli/cli_options.py +++ b/packages/tangle-cli/src/tangle_cli/cli_options.py @@ -2,6 +2,7 @@ from __future__ import annotations +from pathlib import Path from typing import Annotated from cyclopts import Parameter @@ -50,3 +51,26 @@ str, Parameter(help="Log output: console, none, file."), ] +CaBundleOption = Annotated[ + Path | None, + Parameter( + name="--ca-bundle", + help=( + "Path to a PEM CA bundle used as the TLS trust store for every " + "transport. Overrides TANGLE_API_CA_BUNDLE. Place before the " + "subcommand, e.g. `tangle --ca-bundle ca.pem api ...`." + ), + ), +] +VerifyTlsOption = Annotated[ + bool | None, + Parameter( + name="--verify-tls", + help=( + "Enable (--verify-tls) or disable (--no-verify-tls) TLS " + "certificate verification for every transport. Overrides " + "TANGLE_API_VERIFY_TLS. --no-verify-tls is for local development " + "only. Place before the subcommand." + ), + ), +] diff --git a/packages/tangle-cli/src/tangle_cli/client.py b/packages/tangle-cli/src/tangle_cli/client.py index 060c960..c0e494a 100644 --- a/packages/tangle-cli/src/tangle_cli/client.py +++ b/packages/tangle-cli/src/tangle_cli/client.py @@ -19,11 +19,14 @@ from .api_transport import ( DEFAULT_TIMEOUT_SECONDS, + VerifyArgument, _join_operation_url, _normalize_base_url, _request_headers, + _VERIFY_UNSET, default_base_url, log_http_exchange, + resolve_verify, tangle_verbose_enabled, ) from tangle_api.generated.operations import GeneratedTangleApiOperations @@ -67,6 +70,7 @@ def __init__( timeout: float = DEFAULT_TIMEOUT_SECONDS, session: requests.Session | None = None, include_env_credentials: bool = True, + verify: VerifyArgument = None, ) -> None: self.base_url = _normalize_base_url(base_url or default_base_url()) env_verbose = tangle_verbose_enabled() @@ -79,6 +83,7 @@ def __init__( self.timeout = timeout self.session = session or requests.Session() self.include_env_credentials = include_env_credentials + self._verify = resolve_verify(verify) def _response_model(self, model_name: str, default: Any) -> Any: """Use CLI-composed models for generated operation deserialization.""" @@ -224,6 +229,9 @@ def _request_with_same_origin_redirects( for _ in range(self._MAX_REDIRECTS + 1): request_headers = self._headers(extra_headers) + call_kwargs = dict(request_kwargs) + if self._verify is not _VERIFY_UNSET and "verify" not in call_kwargs: + call_kwargs["verify"] = self._verify response = self.session.request( current_method, current_url, @@ -232,7 +240,7 @@ def _request_with_same_origin_redirects( headers=request_headers, timeout=timeout, allow_redirects=False, - **request_kwargs, + **call_kwargs, ) if self.verbose: log_http_exchange( diff --git a/packages/tangle-cli/src/tangle_cli/dynamic_discovery_client.py b/packages/tangle-cli/src/tangle_cli/dynamic_discovery_client.py index d3a5582..3270919 100644 --- a/packages/tangle-cli/src/tangle_cli/dynamic_discovery_client.py +++ b/packages/tangle-cli/src/tangle_cli/dynamic_discovery_client.py @@ -18,6 +18,7 @@ ) from .api_transport import ( DEFAULT_TIMEOUT_SECONDS, + VerifyArgument, _normalize_base_url, default_base_url, request_operation, @@ -43,6 +44,7 @@ def __init__( auth_header: str | None = None, header: list[str] | str | None = None, timeout: float = DEFAULT_TIMEOUT_SECONDS, + verify: VerifyArgument = None, ) -> None: self.schema = schema self.base_url = _normalize_base_url(base_url or default_base_url()) @@ -51,6 +53,7 @@ def __init__( self.auth_header = auth_header self.header = _header_list(header) self.timeout = timeout + self.verify = verify self._operations = operation_map(schema) self._aliases = self._build_alias_map(self._operations) self._groups = self._build_groups(self._operations) @@ -66,6 +69,7 @@ def from_schema( auth_header: str | None = None, header: list[str] | str | None = None, timeout: float = DEFAULT_TIMEOUT_SECONDS, + verify: VerifyArgument = None, ) -> TangleDynamicDiscoveryClient: """Create a client from an already loaded OpenAPI schema.""" @@ -77,6 +81,7 @@ def from_schema( auth_header=auth_header, header=header, timeout=timeout, + verify=verify, ) @classmethod @@ -89,6 +94,7 @@ def from_cache( auth_header: str | None = None, header: list[str] | str | None = None, timeout: float = DEFAULT_TIMEOUT_SECONDS, + verify: VerifyArgument = None, ) -> TangleDynamicDiscoveryClient: """Create a client from the local schema cache without network access.""" @@ -107,6 +113,7 @@ def from_cache( auth_header=auth_header, header=header, timeout=timeout, + verify=verify, ) @classmethod @@ -119,6 +126,7 @@ def from_url( auth_header: str | None = None, header: list[str] | str | None = None, timeout: float = DEFAULT_TIMEOUT_SECONDS, + verify: VerifyArgument = None, ) -> TangleDynamicDiscoveryClient: """Fetch ``/openapi.json`` and create a client without writing the cache.""" @@ -129,6 +137,7 @@ def from_url( header=header, auth_header=auth_header, headers=headers, + verify=verify, ) return cls.from_schema( schema, @@ -138,6 +147,7 @@ def from_url( auth_header=auth_header, header=header, timeout=timeout, + verify=verify, ) @classmethod @@ -150,6 +160,7 @@ def from_cache_or_refresh( auth_header: str | None = None, header: list[str] | str | None = None, timeout: float = DEFAULT_TIMEOUT_SECONDS, + verify: VerifyArgument = None, ) -> TangleDynamicDiscoveryClient: """Create a client from cache, fetching and caching the schema on miss.""" @@ -162,6 +173,7 @@ def from_cache_or_refresh( header=header, auth_header=auth_header, headers=headers, + verify=verify, ) return cls.from_schema( schema, @@ -171,6 +183,7 @@ def from_cache_or_refresh( auth_header=auth_header, header=header, timeout=timeout, + verify=verify, ) @property @@ -197,6 +210,7 @@ def request(self, operation_name: str, **params: Any) -> httpx.Response: headers = {**self.headers, **dict(headers_override or {})} body = params.pop("body", None) timeout = params.pop("timeout", self.timeout) + verify = params.pop("verify", self.verify) return request_operation( operation, params, @@ -207,6 +221,7 @@ def request(self, operation_name: str, **params: Any) -> httpx.Response: headers=headers, body=body, timeout=timeout, + verify=verify, ) def call(self, operation_name: str, **params: Any) -> Any: diff --git a/tests/test_packaging.py b/tests/test_packaging.py index dd56d23..c3fc0cb 100644 --- a/tests/test_packaging.py +++ b/tests/test_packaging.py @@ -40,7 +40,7 @@ def _write_import_stubs(path: Path) -> None: _write_runtime_stubs(path) (path / "cyclopts.py").write_text( "class App:\n" - " def __init__(self, *args, **kwargs): pass\n" + " def __init__(self, *args, **kwargs): self._meta = None\n" " def command(self, obj=None, **kwargs):\n" " if obj is not None:\n" " return obj\n" @@ -48,6 +48,11 @@ def _write_import_stubs(path: Path) -> None: " return decorator\n" " def __call__(self, *args, **kwargs): pass\n" " def default(self, fn): return fn\n" + " @property\n" + " def meta(self):\n" + " if self._meta is None:\n" + " self._meta = App()\n" + " return self._meta\n" "\n" "def Parameter(*args, **kwargs): return object()\n", encoding="utf-8", diff --git a/tests/test_tls_verification.py b/tests/test_tls_verification.py new file mode 100644 index 0000000..74be544 --- /dev/null +++ b/tests/test_tls_verification.py @@ -0,0 +1,748 @@ +"""Tests for centralized TLS verification configuration across transports. + +Covers environment parsing, explicit precedence, path handling, preservation of +caller-supplied session settings, and propagation into the requests client, the +httpx schema/operation transport, and the dynamic-discovery client. A real +local HTTPS server with a generated private CA exercises the end-to-end +behavior of the secure default, ``TANGLE_API_CA_BUNDLE``, and +``TANGLE_API_VERIFY_TLS=0``. +""" + +from __future__ import annotations + +import http.server +import json +import os +import shutil +import ssl +import subprocess +import sys +import threading +from pathlib import Path +from types import SimpleNamespace +from typing import Any, Iterator + +import httpx +import pytest +import requests + +from tangle_cli import cli +from tangle_cli.api_cli import _api_argv_tail +from tangle_cli.api_schema import fetch_schema +from tangle_cli.api_transport import ( + _VERIFY_UNSET, + configure_cli_verify, + httpx_verify, + request_operation, + resolve_verify, + resolve_verify_default, +) +from tangle_cli.client import TangleApiClient +from tangle_cli.dynamic_discovery_client import TangleDynamicDiscoveryClient + +_TLS_ENV_VARS = ("TANGLE_API_VERIFY_TLS", "TANGLE_API_CA_BUNDLE") + + +@pytest.fixture(autouse=True) +def _clear_tls_env(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]: + for name in _TLS_ENV_VARS: + monkeypatch.delenv(name, raising=False) + configure_cli_verify() # ensure no leaked process-wide CLI override + yield + configure_cli_verify() + + +# --------------------------------------------------------------------------- # +# Environment parsing and precedence +# --------------------------------------------------------------------------- # + + +def test_resolve_verify_defaults_to_unset() -> None: + assert resolve_verify() is _VERIFY_UNSET + assert resolve_verify_default() is True + assert httpx_verify() is True + + +@pytest.mark.parametrize("value", ["0", "false", "FALSE", " no ", "No", "nO"]) +def test_verify_tls_false_values_disable_verification( + monkeypatch: pytest.MonkeyPatch, value: str +) -> None: + monkeypatch.setenv("TANGLE_API_VERIFY_TLS", value) + assert resolve_verify() is False + assert httpx_verify() is False + + +@pytest.mark.parametrize("value", ["1", "true", "yes", "on", "enabled", "anything"]) +def test_verify_tls_other_nonempty_values_keep_verification( + monkeypatch: pytest.MonkeyPatch, value: str +) -> None: + monkeypatch.setenv("TANGLE_API_VERIFY_TLS", value) + assert resolve_verify() is True + + +def test_empty_verify_tls_is_treated_as_unset(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("TANGLE_API_VERIFY_TLS", " ") + assert resolve_verify() is _VERIFY_UNSET + + +def test_ca_bundle_env_resolves_to_path( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + bundle = tmp_path / "ca.pem" + bundle.write_text("cert", encoding="utf-8") + monkeypatch.setenv("TANGLE_API_CA_BUNDLE", str(bundle)) + assert resolve_verify() == str(bundle) + + +def test_empty_ca_bundle_is_treated_as_unset(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("TANGLE_API_CA_BUNDLE", " ") + assert resolve_verify() is _VERIFY_UNSET + + +def test_missing_ca_bundle_fails_early(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("TANGLE_API_CA_BUNDLE", "/nonexistent/ca.pem") + with pytest.raises(SystemExit, match="TANGLE_API_CA_BUNDLE"): + resolve_verify() + + +def test_ca_bundle_wins_over_verify_tls( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + bundle = tmp_path / "ca.pem" + bundle.write_text("cert", encoding="utf-8") + monkeypatch.setenv("TANGLE_API_CA_BUNDLE", str(bundle)) + monkeypatch.setenv("TANGLE_API_VERIFY_TLS", "0") + # CA bundle wins and TLS stays verified against the bundle. + assert resolve_verify() == str(bundle) + + +def test_explicit_argument_wins_over_environment( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + bundle = tmp_path / "ca.pem" + bundle.write_text("cert", encoding="utf-8") + monkeypatch.setenv("TANGLE_API_CA_BUNDLE", str(bundle)) + assert resolve_verify(False) is False + assert resolve_verify(True) is True + + +def test_explicit_pathlike_argument_is_accepted(tmp_path: Path) -> None: + bundle = tmp_path / "ca.pem" + bundle.write_text("cert", encoding="utf-8") + assert resolve_verify(bundle) == str(bundle) + + +def test_explicit_missing_path_argument_fails_early(tmp_path: Path) -> None: + with pytest.raises(SystemExit, match="verify"): + resolve_verify(str(tmp_path / "missing.pem")) + + +def test_none_argument_falls_through_to_environment( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("TANGLE_API_VERIFY_TLS", "0") + assert resolve_verify(None) is False + + +# --------------------------------------------------------------------------- # +# httpx adapter +# --------------------------------------------------------------------------- # + + +def test_httpx_verify_converts_path_to_ssl_context(tmp_path: Path) -> None: + ca = _generate_private_ca(tmp_path) + context = httpx_verify(str(ca.ca_pem)) + assert isinstance(context, ssl.SSLContext) + + +def test_httpx_verify_passes_booleans_through() -> None: + assert httpx_verify(True) is True + assert httpx_verify(False) is False + + +# --------------------------------------------------------------------------- # +# Transport propagation (mocked) +# --------------------------------------------------------------------------- # + + +def _operation(path: str, *, method: str = "GET") -> SimpleNamespace: + return SimpleNamespace( + method=method, + path=path, + parameters=[], + group_name="test", + command_name="op", + has_request_body=False, + ) + + +def test_request_operation_passes_verify_to_httpx( + monkeypatch: pytest.MonkeyPatch, +) -> None: + captured: dict[str, Any] = {} + + def fake_request(method: str, url: str, **kwargs: Any) -> httpx.Response: + captured.update(kwargs) + return httpx.Response(200, json={}, request=httpx.Request(method, url)) + + monkeypatch.setattr("tangle_cli.api_transport.httpx.request", fake_request) + request_operation( + _operation("/api/ping"), + {}, + base_url="https://api.test", + verify=False, + ) + assert captured["verify"] is False + + +def test_request_operation_defaults_to_secure_verify( + monkeypatch: pytest.MonkeyPatch, +) -> None: + captured: dict[str, Any] = {} + + def fake_request(method: str, url: str, **kwargs: Any) -> httpx.Response: + captured.update(kwargs) + return httpx.Response(200, json={}, request=httpx.Request(method, url)) + + monkeypatch.setattr("tangle_cli.api_transport.httpx.request", fake_request) + request_operation(_operation("/api/ping"), {}, base_url="https://api.test") + assert captured["verify"] is True + + +def test_fetch_schema_passes_verify_to_httpx(monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict[str, Any] = {} + + def fake_get(url: str, **kwargs: Any) -> httpx.Response: + captured.update(kwargs) + return httpx.Response( + 200, + json={"openapi": "3.1.0", "paths": {}}, + request=httpx.Request("GET", url), + ) + + monkeypatch.setattr("tangle_cli.api_schema.httpx.get", fake_get) + fetch_schema("https://api.test", verify=False) + assert captured["verify"] is False + + +class _RecordingSession: + """Minimal session without a ``.verify`` attribute.""" + + def __init__(self) -> None: + self.calls: list[dict[str, Any]] = [] + + def request(self, method: str, url: str, **kwargs: Any) -> requests.Response: + self.calls.append({"method": method, "url": url, **kwargs}) + r = requests.Response() + r.status_code = 200 + r._content = b"{}" + r.headers["Content-Type"] = "application/json" + r.request = requests.Request(method, url).prepare() + return r + + +def test_static_client_injects_verify_when_configured( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("TANGLE_API_VERIFY_TLS", "0") + session = _RecordingSession() + client = TangleApiClient("https://api.test", session=session) + + client._make_request("GET", "/api/ping") + + assert session.calls[0]["verify"] is False + + +def test_static_client_preserves_session_when_unset() -> None: + session = _RecordingSession() + client = TangleApiClient("https://api.test", session=session) + + client._make_request("GET", "/api/ping") + + # No Tangle TLS setting: do not pass ``verify`` so requests' own + # session/environment handling (REQUESTS_CA_BUNDLE/CURL_CA_BUNDLE) applies. + assert "verify" not in session.calls[0] + + +def test_static_client_explicit_verify_argument_wins() -> None: + session = _RecordingSession() + client = TangleApiClient("https://api.test", session=session, verify=False) + + client._make_request("GET", "/api/ping") + + assert session.calls[0]["verify"] is False + + +# --------------------------------------------------------------------------- # +# Real local HTTPS end-to-end tests with a generated private CA +# --------------------------------------------------------------------------- # + + +class _GeneratedCa(SimpleNamespace): + ca_pem: Path + server_crt: Path + server_key: Path + + +def _generate_private_ca(directory: Path) -> _GeneratedCa: + """Generate a private CA and a localhost server cert using openssl.""" + + ca_key = directory / "ca.key" + ca_pem = directory / "ca.pem" + server_key = directory / "server.key" + server_csr = directory / "server.csr" + server_crt = directory / "server.crt" + ext = directory / "ext.cnf" + ext.write_text("subjectAltName=DNS:localhost,IP:127.0.0.1\n", encoding="utf-8") + + def _run(args: list[str]) -> None: + subprocess.run( + args, + check=True, + capture_output=True, + ) + + _run([ + "openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes", + "-keyout", str(ca_key), "-out", str(ca_pem), + "-subj", "/CN=Tangle Test CA", "-days", "2", + "-addext", "basicConstraints=critical,CA:TRUE", + "-addext", "keyUsage=critical,keyCertSign,cRLSign", + ]) + _run([ + "openssl", "req", "-newkey", "rsa:2048", "-nodes", + "-keyout", str(server_key), "-out", str(server_csr), + "-subj", "/CN=localhost", + ]) + _run([ + "openssl", "x509", "-req", "-in", str(server_csr), + "-CA", str(ca_pem), "-CAkey", str(ca_key), "-CAcreateserial", + "-out", str(server_crt), "-days", "2", "-extfile", str(ext), + ]) + return _GeneratedCa(ca_pem=ca_pem, server_crt=server_crt, server_key=server_key) + + +class _SchemaHandler(http.server.BaseHTTPRequestHandler): + def do_GET(self) -> None: # noqa: N802 - required by BaseHTTPRequestHandler + body = json.dumps({"openapi": "3.1.0", "info": {}, "paths": {}}).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, *args: Any) -> None: # silence test server logging + pass + + +@pytest.fixture(scope="module") +def https_server(tmp_path_factory: pytest.TempPathFactory) -> Iterator[dict[str, Any]]: + if shutil.which("openssl") is None: # pragma: no cover - environment guard + pytest.skip("openssl is required for the real HTTPS TLS tests") + + directory = tmp_path_factory.mktemp("tls") + ca = _generate_private_ca(directory) + + context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + context.load_cert_chain(certfile=str(ca.server_crt), keyfile=str(ca.server_key)) + + server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), _SchemaHandler) + server.socket = context.wrap_socket(server.socket, server_side=True) + port = server.server_address[1] + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield { + "base_url": f"https://localhost:{port}", + "ca_pem": str(ca.ca_pem), + } + finally: + server.shutdown() + thread.join(timeout=5) + server.server_close() + + +def test_httpx_schema_default_verification_fails( + https_server: dict[str, Any], +) -> None: + with pytest.raises(httpx.ConnectError): + fetch_schema(https_server["base_url"]) + + +def test_httpx_schema_ca_bundle_succeeds( + monkeypatch: pytest.MonkeyPatch, https_server: dict[str, Any] +) -> None: + monkeypatch.setenv("TANGLE_API_CA_BUNDLE", https_server["ca_pem"]) + schema = fetch_schema(https_server["base_url"]) + assert schema["paths"] == {} + + +def test_httpx_schema_verify_off_succeeds( + monkeypatch: pytest.MonkeyPatch, https_server: dict[str, Any] +) -> None: + monkeypatch.setenv("TANGLE_API_VERIFY_TLS", "0") + schema = fetch_schema(https_server["base_url"]) + assert schema["paths"] == {} + + +def test_dynamic_client_from_url_uses_ca_bundle( + monkeypatch: pytest.MonkeyPatch, https_server: dict[str, Any] +) -> None: + monkeypatch.setenv("TANGLE_API_CA_BUNDLE", https_server["ca_pem"]) + client = TangleDynamicDiscoveryClient.from_url(https_server["base_url"]) + assert client.operations == () + + +def test_dynamic_client_from_url_default_verification_fails( + https_server: dict[str, Any], +) -> None: + with pytest.raises(httpx.ConnectError): + TangleDynamicDiscoveryClient.from_url(https_server["base_url"]) + + +def test_requests_client_default_verification_fails( + https_server: dict[str, Any], +) -> None: + client = TangleApiClient(https_server["base_url"]) + with pytest.raises(requests.exceptions.SSLError): + client._make_request("GET", "/api/ping") + + +def test_requests_client_ca_bundle_succeeds( + monkeypatch: pytest.MonkeyPatch, https_server: dict[str, Any] +) -> None: + monkeypatch.setenv("TANGLE_API_CA_BUNDLE", https_server["ca_pem"]) + client = TangleApiClient(https_server["base_url"]) + response = client._make_request("GET", "/api/ping") + assert response.status_code == 200 + + +def test_requests_client_verify_off_succeeds( + monkeypatch: pytest.MonkeyPatch, https_server: dict[str, Any] +) -> None: + monkeypatch.setenv("TANGLE_API_VERIFY_TLS", "0") + client = TangleApiClient(https_server["base_url"]) + response = client._make_request("GET", "/api/ping") + assert response.status_code == 200 + + +# --------------------------------------------------------------------------- # +# Global CLI override: configure_cli_verify and resolver precedence +# --------------------------------------------------------------------------- # + + +def test_cli_override_disables_verification() -> None: + configure_cli_verify(verify_tls=False) + assert resolve_verify() is False + + +def test_cli_override_enables_verification() -> None: + configure_cli_verify(verify_tls=True) + assert resolve_verify() is True + + +def test_cli_override_ca_bundle_resolves_to_path(tmp_path: Path) -> None: + bundle = tmp_path / "ca.pem" + bundle.write_text("cert", encoding="utf-8") + configure_cli_verify(ca_bundle=bundle) + assert resolve_verify() == str(bundle) + + +def test_cli_override_wins_over_environment( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("TANGLE_API_VERIFY_TLS", "1") + configure_cli_verify(verify_tls=False) + assert resolve_verify() is False + + +def test_cli_ca_bundle_wins_over_environment( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + env_bundle = tmp_path / "env.pem" + env_bundle.write_text("cert", encoding="utf-8") + cli_bundle = tmp_path / "cli.pem" + cli_bundle.write_text("cert", encoding="utf-8") + monkeypatch.setenv("TANGLE_API_CA_BUNDLE", str(env_bundle)) + configure_cli_verify(ca_bundle=cli_bundle) + assert resolve_verify() == str(cli_bundle) + + +def test_explicit_python_argument_wins_over_cli_override() -> None: + configure_cli_verify(verify_tls=False) + # A library caller's explicit verify= stays highest precedence. + assert resolve_verify(True) is True + + +def test_cli_override_conflict_ca_bundle_and_no_verify_fails(tmp_path: Path) -> None: + bundle = tmp_path / "ca.pem" + bundle.write_text("cert", encoding="utf-8") + with pytest.raises(SystemExit, match="ca-bundle"): + configure_cli_verify(ca_bundle=bundle, verify_tls=False) + + +def test_cli_override_ca_bundle_and_verify_true_is_accepted(tmp_path: Path) -> None: + bundle = tmp_path / "ca.pem" + bundle.write_text("cert", encoding="utf-8") + configure_cli_verify(ca_bundle=bundle, verify_tls=True) + assert resolve_verify() == str(bundle) + + +def test_cli_override_missing_ca_bundle_fails_early() -> None: + with pytest.raises(SystemExit, match="ca-bundle"): + configure_cli_verify(ca_bundle="/nonexistent/ca.pem") + + +def test_cli_override_clears_back_to_unset(monkeypatch: pytest.MonkeyPatch) -> None: + configure_cli_verify(verify_tls=False) + assert resolve_verify() is False + configure_cli_verify() + assert resolve_verify() is _VERIFY_UNSET + + +# --------------------------------------------------------------------------- # +# argv pre-parse and placement +# --------------------------------------------------------------------------- # + + +@pytest.mark.parametrize( + "argv, expected", + [ + (["tangle", "--no-verify-tls"], False), + (["tangle", "--verify-tls"], True), + ], +) +def test_configure_tls_from_argv_parses_boolean_flags( + argv: list[str], expected: bool +) -> None: + cli._configure_tls_from_argv(argv) + assert resolve_verify() is expected + + +def test_configure_tls_from_argv_parses_ca_bundle(tmp_path: Path) -> None: + bundle = tmp_path / "ca.pem" + bundle.write_text("cert", encoding="utf-8") + cli._configure_tls_from_argv(["tangle", "--ca-bundle", str(bundle), "api", "ping"]) + assert resolve_verify() == str(bundle) + + +def test_configure_tls_from_argv_parses_ca_bundle_equals(tmp_path: Path) -> None: + bundle = tmp_path / "ca.pem" + bundle.write_text("cert", encoding="utf-8") + cli._configure_tls_from_argv(["tangle", f"--ca-bundle={bundle}", "sdk"]) + assert resolve_verify() == str(bundle) + + +def test_configure_tls_from_argv_stops_at_subcommand( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("TANGLE_API_VERIFY_TLS", "1") + # A --no-verify-tls after the subcommand is not a global flag: it must not + # be consumed here, leaving the env default to apply. + cli._configure_tls_from_argv(["tangle", "api", "ping", "--no-verify-tls"]) + assert resolve_verify() is True + + +@pytest.mark.parametrize( + "argv, expected_tail", + [ + (["tangle", "--no-verify-tls", "api", "ping"], ["ping"]), + (["tangle", "--ca-bundle", "ca.pem", "api", "foo", "bar"], ["foo", "bar"]), + (["tangle", "--ca-bundle=ca.pem", "api", "foo"], ["foo"]), + (["tangle", "--verify-tls", "api"], []), + ], +) +def test_api_argv_tail_skips_global_tls_flags( + argv: list[str], expected_tail: list[str] +) -> None: + assert _api_argv_tail(argv) == expected_tail + + +# --------------------------------------------------------------------------- # +# Global override propagation into every transport (mocked) +# --------------------------------------------------------------------------- # + + +def test_cli_override_propagates_to_static_requests_client() -> None: + configure_cli_verify(verify_tls=False) + session = _RecordingSession() + client = TangleApiClient("https://api.test", session=session) + client._make_request("GET", "/api/ping") + assert session.calls[0]["verify"] is False + + +def test_cli_override_propagates_to_request_operation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + captured: dict[str, Any] = {} + + def fake_request(method: str, url: str, **kwargs: Any) -> httpx.Response: + captured.update(kwargs) + return httpx.Response(200, json={}, request=httpx.Request(method, url)) + + monkeypatch.setattr("tangle_cli.api_transport.httpx.request", fake_request) + configure_cli_verify(verify_tls=False) + request_operation(_operation("/api/ping"), {}, base_url="https://api.test") + assert captured["verify"] is False + + +def test_cli_override_propagates_to_schema_fetch( + monkeypatch: pytest.MonkeyPatch, +) -> None: + captured: dict[str, Any] = {} + + def fake_get(url: str, **kwargs: Any) -> httpx.Response: + captured.update(kwargs) + return httpx.Response( + 200, + json={"openapi": "3.1.0", "paths": {}}, + request=httpx.Request("GET", url), + ) + + monkeypatch.setattr("tangle_cli.api_schema.httpx.get", fake_get) + configure_cli_verify(verify_tls=False) + fetch_schema("https://api.test") + assert captured["verify"] is False + + +# --------------------------------------------------------------------------- # +# Real local HTTPS end-to-end tests driven by the global CLI override +# --------------------------------------------------------------------------- # + + +def test_cli_override_ca_bundle_schema_fetch_succeeds( + https_server: dict[str, Any], +) -> None: + configure_cli_verify(ca_bundle=https_server["ca_pem"]) + schema = fetch_schema(https_server["base_url"]) + assert schema["paths"] == {} + + +def test_cli_override_verify_off_schema_fetch_succeeds( + https_server: dict[str, Any], +) -> None: + configure_cli_verify(verify_tls=False) + schema = fetch_schema(https_server["base_url"]) + assert schema["paths"] == {} + + +def test_cli_override_default_schema_fetch_fails( + https_server: dict[str, Any], +) -> None: + configure_cli_verify() + with pytest.raises(httpx.ConnectError): + fetch_schema(https_server["base_url"]) + + +def test_cli_override_ca_bundle_dynamic_client_succeeds( + https_server: dict[str, Any], +) -> None: + configure_cli_verify(ca_bundle=https_server["ca_pem"]) + client = TangleDynamicDiscoveryClient.from_url(https_server["base_url"]) + assert client.operations == () + + +def test_cli_override_ca_bundle_requests_client_succeeds( + https_server: dict[str, Any], +) -> None: + configure_cli_verify(ca_bundle=https_server["ca_pem"]) + client = TangleApiClient(https_server["base_url"]) + response = client._make_request("GET", "/api/ping") + assert response.status_code == 200 + + +# --------------------------------------------------------------------------- # +# Live CLI subprocess: global flags reach `tangle api refresh` over real HTTPS +# --------------------------------------------------------------------------- # + + +def _run_tangle( + args: list[str], cache_dir: Path, extra_env: dict[str, str] | None = None +) -> subprocess.CompletedProcess[str]: + env = {**os.environ, "TANGLE_CLI_CACHE_DIR": str(cache_dir)} + if extra_env: + env.update(extra_env) + return subprocess.run( + [ + sys.executable, + "-c", + "import sys; from tangle_cli.cli import main; " + "sys.argv = ['tangle', *sys.argv[1:]]; main()", + *args, + ], + capture_output=True, + text=True, + env=env, + ) + + +def test_cli_subprocess_ca_bundle_refresh_succeeds( + https_server: dict[str, Any], tmp_path: Path +) -> None: + result = _run_tangle( + [ + "--ca-bundle", + https_server["ca_pem"], + "api", + "refresh", + "--base-url", + https_server["base_url"], + ], + tmp_path, + ) + assert result.returncode == 0, result.stderr + assert "Cached OpenAPI schema" in result.stdout + + +def test_cli_subprocess_default_refresh_fails( + https_server: dict[str, Any], tmp_path: Path +) -> None: + result = _run_tangle( + ["api", "refresh", "--base-url", https_server["base_url"]], + tmp_path, + ) + assert result.returncode != 0 + assert "Failed to fetch" in result.stderr + + +def test_cli_subprocess_verify_off_refresh_succeeds( + https_server: dict[str, Any], tmp_path: Path +) -> None: + result = _run_tangle( + [ + "--no-verify-tls", + "api", + "refresh", + "--base-url", + https_server["base_url"], + ], + tmp_path, + ) + assert result.returncode == 0, result.stderr + assert "Cached OpenAPI schema" in result.stdout + + +def test_cli_subprocess_ca_bundle_and_no_verify_conflict_fails( + https_server: dict[str, Any], tmp_path: Path +) -> None: + result = _run_tangle( + [ + "--ca-bundle", + https_server["ca_pem"], + "--no-verify-tls", + "api", + "refresh", + "--base-url", + https_server["base_url"], + ], + tmp_path, + ) + assert result.returncode != 0 + assert "ca-bundle" in result.stderr + + +def test_cli_root_help_lists_global_tls_flags(tmp_path: Path) -> None: + result = _run_tangle(["--help"], tmp_path) + assert result.returncode == 0 + assert "--ca-bundle" in result.stdout + assert "--no-verify-tls" in result.stdout