From 099a24e019144ed7fe54d84068babc7d6030e98f Mon Sep 17 00:00:00 2001 From: jon-trust Date: Fri, 9 Oct 2026 06:13:47 +1300 Subject: [PATCH] Note when an exported key goes to a stored endpoint The key and the endpoint are resolved independently, so a key exported in TOC_API_KEY is sent to whatever non-default endpoint `tmxcaliber init` stored, unless TOC_API_URL is also set. That precedence is unchanged. Commands now print a note on stderr, once per run and before the first request, naming the endpoint and the credentials file it came from and pointing at TOC_API_URL. Stdout stays the command's output. Settings records where the endpoint came from and owns the condition, so the library still prints nothing. API commands and ThreatModel references given as a document source both build their client through remote_cli.connect, the one place the note is written. Bump to 0.5.4. --- README.md | 5 + pyproject.toml | 2 +- test/test_remote_cli.py | 207 ++++++++++++++++++++++++++++++- test/test_remote_config.py | 115 +++++++++++++++++ tmxcaliber/cli.py | 8 +- tmxcaliber/lib/remote/config.py | 57 ++++++++- tmxcaliber/lib/remote/resolve.py | 20 ++- tmxcaliber/remote_cli.py | 74 ++++++++++- uv.lock | 2 +- 9 files changed, 471 insertions(+), 19 deletions(-) diff --git a/README.md b/README.md index e77dae8..dc852c5 100644 --- a/README.md +++ b/README.md @@ -140,6 +140,11 @@ The environment wins over the file, which is also why there are no profiles: one stored key, and an environment variable for the "a different one, right now" case. +The key and the endpoint are each looked up on their own. If `init` stored an +endpoint other than the default, a key you export in `TOC_API_KEY` is sent to +that stored endpoint unless you also set `TOC_API_URL`. Commands print a note +on stderr when that happens, naming the endpoint and the file it came from. + The key is a TrustOnCloud credential rather than a tmxcaliber one, which is why it is named and stored this way: the same key works with `curl` or any other client. There is deliberately no `--api-key` flag, so the secret never diff --git a/pyproject.toml b/pyproject.toml index b561df1..7e86eb8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "tmxcaliber" -version = "0.5.3" +version = "0.5.4" description = "CLI utility to filter down one or more TrustOnCloud ThreatModels and get more refined information." readme = "README.md" requires-python = ">=3.10" diff --git a/test/test_remote_cli.py b/test/test_remote_cli.py index 08f9208..3767971 100644 --- a/test/test_remote_cli.py +++ b/test/test_remote_cli.py @@ -7,16 +7,26 @@ import sys from argparse import Namespace from collections.abc import Iterator, Mapping -from typing import Any +from typing import Any, NoReturn import pytest from tmxcaliber import cli as cli_module +from tmxcaliber import remote_cli +from tmxcaliber.lib.remote.config import Settings from tmxcaliber.lib.remote.contract import route_for +from tmxcaliber.lib.remote.errors import AuthenticationError from tmxcaliber.lib.remote.resolve import resolve_source -from tmxcaliber.remote_cli import run_api_command +from tmxcaliber.remote_cli import connect, resolve_command_source, run_api_command from .test_remote_assemble import DETAIL, DFD, PARTS, _at_release +from .test_remote_config import ( + HAZARD, + NOTED_CASES, + QUIET_CASES, + STORED_URL, + EndpointCase, +) class StubClient: @@ -445,3 +455,196 @@ def test_an_unpinned_fetch_satisfies_a_later_pinned_request( assert pinned.calls == [], "the pinned request repeated the assembly" assert path.endswith("1611187200.json") + + +#: How the stored-endpoint note starts, wherever a test only needs to find it. +NOTE = "Note: the key in TOC_API_KEY will be sent to" + + +@pytest.fixture(autouse=True) +def fresh_notes(monkeypatch: pytest.MonkeyPatch) -> None: + """Forget the stored-endpoint notes earlier tests wrote. + + The CLI says it once per process, which is once per invocation in real + use but once per whole session under pytest. + + Args: + monkeypatch: Replaces the record for this test only. + """ + monkeypatch.setattr(remote_cli, "_noted", set()) + + +class SettingsStub(StubClient): + """A StubClient built the way `connect` builds a real client.""" + + def __init__(self, settings: Settings) -> None: + """Keep the settings `connect` resolved. + + Args: + settings: The resolved settings. + """ + super().__init__() + self.settings = settings + + +class RefusingClient(SettingsStub): + """Refuses every call, as an API that does not know the key would.""" + + def get(self, path: str, params: Mapping[str, str] | None = None) -> dict[str, Any]: + """Refuse the call. + + Args: + path: The path. + params: The query. + + Raises: + AuthenticationError: Always. + """ + raise AuthenticationError("The API key was not accepted.", status=401) + + +def _use_environment( + monkeypatch: pytest.MonkeyPatch, env: Mapping[str, str], tmp_path: pathlib.Path +) -> None: + """Make `env` the whole of the credential environment, for the real CLI. + + Args: + monkeypatch: Sets and clears the variables. + env: The variables to set. + tmp_path: Holds the document cache and is the working directory, so + a reference cannot collide with a file of the same name. + """ + for name in ("TOC_API_KEY", "TOC_API_URL", "TOC_CONFIG_FILE"): + monkeypatch.delenv(name, raising=False) + for name, value in env.items(): + monkeypatch.setenv(name, value) + monkeypatch.setenv("TMXCALIBER_CACHE_DIR", str(tmp_path / "cache")) + monkeypatch.chdir(tmp_path) + + +@pytest.mark.parametrize("case", NOTED_CASES) +def test_a_stored_endpoint_under_an_environment_key_is_noted_on_stderr( + tmp_path: pathlib.Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], + case: EndpointCase, +) -> None: + env = case.environment(tmp_path) + monkeypatch.setattr(remote_cli, "TocClient", SettingsStub) + + connect(env=env) + + captured = capsys.readouterr() + assert ( + f"Note: the key in TOC_API_KEY will be sent to {STORED_URL}, the " + f"endpoint stored in {env['TOC_CONFIG_FILE']}. Set TOC_API_URL to " + "choose a different endpoint." + ) in captured.err + assert captured.out == "" + + +@pytest.mark.parametrize("case", QUIET_CASES) +def test_nothing_is_said_about_an_endpoint_the_caller_chose_or_expects( + tmp_path: pathlib.Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], + case: EndpointCase, +) -> None: + monkeypatch.setattr(remote_cli, "TocClient", SettingsStub) + + connect(env=case.environment(tmp_path)) + + captured = capsys.readouterr() + assert captured.err == "" + assert captured.out == "" + + +def test_an_api_command_notes_on_stderr_and_leaves_stdout_to_the_json( + tmp_path: pathlib.Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + _use_environment(monkeypatch, HAZARD.environment(tmp_path), tmp_path) + monkeypatch.setattr(remote_cli, "TocClient", SettingsStub) + monkeypatch.setattr(sys, "argv", ["tmxcaliber", "me"]) + + cli_module.main() + + captured = capsys.readouterr() + assert json.loads(captured.out)["tenantId"] == "t-1" + assert NOTE in captured.err + + +def test_the_note_is_written_even_when_the_request_then_fails( + tmp_path: pathlib.Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + # The caller most needs to know which API refused the key. + _use_environment(monkeypatch, HAZARD.environment(tmp_path), tmp_path) + monkeypatch.setattr(remote_cli, "TocClient", RefusingClient) + monkeypatch.setattr(sys, "argv", ["tmxcaliber", "me"]) + + with pytest.raises(SystemExit) as caught: + cli_module.main() + + captured = capsys.readouterr() + assert caught.value.code == 1 + assert NOTE in captured.err + assert NOTE not in captured.out + + +def test_a_reference_given_as_a_document_source_is_noted_too( + tmp_path: pathlib.Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + _use_environment(monkeypatch, HAZARD.environment(tmp_path), tmp_path) + monkeypatch.setattr(remote_cli, "TocClient", SettingsStub) + argv = ["tmxcaliber", "list", "feature-classes", "aws-s3", "--format", "json"] + monkeypatch.setattr(sys, "argv", argv) + + cli_module.main() + + captured = capsys.readouterr() + assert isinstance(json.loads(captured.out), list) + assert NOTE in captured.err + + +def test_the_note_is_written_once_per_run( + tmp_path: pathlib.Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + # `create-change-log` resolves two references in one run. + _use_environment(monkeypatch, HAZARD.environment(tmp_path), tmp_path) + monkeypatch.setattr(remote_cli, "TocClient", SettingsStub) + + resolve_command_source("aws-s3") + resolve_command_source("aws-s3") + + assert capsys.readouterr().err.count(NOTE) == 1 + + +def test_a_local_file_never_resolves_a_credential( + tmp_path: pathlib.Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + # Laziness is what lets a caller work on a file with no key configured. + _use_environment(monkeypatch, HAZARD.environment(tmp_path), tmp_path) + model = tmp_path / "aws-s3" + model.write_text("{}", encoding="utf-8") + monkeypatch.setattr(remote_cli, "connect", _never_called) + + assert resolve_command_source(str(model)) == str(model) + assert capsys.readouterr().err == "" + + +def _never_called() -> NoReturn: + """Stand in for `connect` where a source must not reach the API. + + Raises: + AssertionError: Always. + """ + raise AssertionError("a local source reached for a credential") diff --git a/test/test_remote_config.py b/test/test_remote_config.py index 2b017d5..103ec02 100644 --- a/test/test_remote_config.py +++ b/test/test_remote_config.py @@ -3,6 +3,7 @@ from __future__ import annotations import pathlib +from dataclasses import dataclass import pytest @@ -38,6 +39,72 @@ def write_config(path: pathlib.Path, body: str) -> pathlib.Path: return path +#: An endpoint other than the default, as `init` would have stored it. +STORED_URL = "https://api.example.com" + + +@dataclass(frozen=True) +class EndpointCase: + """Where the key and the endpoint come from in one test. + + Attributes: + environment_key: Whether TOC_API_KEY is set. + environment_url: TOC_API_URL's value, or empty to leave it unset. + stored_url: The api_url the credentials file holds, or empty for none. + """ + + environment_key: bool + environment_url: str + stored_url: str + + def environment(self, tmp_path: pathlib.Path) -> dict[str, str]: + """Write the credentials file and build the environment for this case. + + The file always holds a key, and a different one from the + environment's, so a test can tell which of the two was used. + + Args: + tmp_path: Where to write the credentials file. + + Returns: + The environment to resolve settings from. + """ + body = f"[default]\napi_key = {OTHER}\n" + if self.stored_url: + body += f"api_url = {self.stored_url}\n" + config = write_config(tmp_path / "credentials", body) + env = {"TOC_CONFIG_FILE": str(config)} + if self.environment_key: + env["TOC_API_KEY"] = KEY + if self.environment_url: + env["TOC_API_URL"] = self.environment_url + return env + + +#: A key from the environment going to an endpoint from the file, which is +#: the one combination a reader is told about. +HAZARD = EndpointCase(True, "", STORED_URL) + +#: The hazard, and the same with a blank TOC_API_URL, which counts as unset. +NOTED_CASES = [ + pytest.param(HAZARD, id="TOC_API_URL unset"), + pytest.param(EndpointCase(True, " ", STORED_URL), id="TOC_API_URL blank"), +] + +#: An endpoint the caller chose, or the default, so there is nothing to say. +QUIET_CASES = [ + pytest.param( + EndpointCase(True, "https://other.example", STORED_URL), id="TOC_API_URL set" + ), + pytest.param(EndpointCase(False, "", STORED_URL), id="key from the file"), + pytest.param(EndpointCase(True, "", ""), id="no endpoint stored"), + pytest.param(EndpointCase(True, "", DEFAULT_BASE_URL), id="stored default"), + pytest.param( + EndpointCase(True, "", DEFAULT_BASE_URL + "/"), id="stored default with slash" + ), +] + + def test_the_environment_supplies_the_key() -> None: found = load_settings(env={"TOC_API_KEY": KEY}) @@ -329,3 +396,51 @@ def test_an_environment_endpoint_still_wins_over_the_stored_one( ) assert found.base_url == "https://other.example" + + +@pytest.mark.parametrize("case", NOTED_CASES) +def test_an_environment_key_going_to_a_stored_endpoint_is_flagged( + tmp_path: pathlib.Path, case: EndpointCase +) -> None: + """The hazard the CLI warns about, recorded rather than printed. + + A key exported for another API goes to the endpoint `init` stored, + because the endpoint is resolved independently of the key. The + precedence stays; the settings say where the endpoint came from. + """ + env = case.environment(tmp_path) + + found = load_settings(env=env) + + assert found.credentials.api_key == KEY + assert found.base_url == STORED_URL + assert found.endpoint_source == env["TOC_CONFIG_FILE"] + assert found.sends_environment_key_to_stored_endpoint is True + + +@pytest.mark.parametrize("case", QUIET_CASES) +def test_an_endpoint_the_caller_chose_or_expects_is_not_flagged( + tmp_path: pathlib.Path, case: EndpointCase +) -> None: + found = load_settings(env=case.environment(tmp_path)) + + assert found.sends_environment_key_to_stored_endpoint is False + + +@pytest.mark.parametrize( + ("case", "expected"), + [ + (EndpointCase(True, "https://other.example", STORED_URL), "TOC_API_URL"), + (EndpointCase(True, "", STORED_URL), "file"), + (EndpointCase(True, "", ""), ""), + ], +) +def test_the_endpoint_records_where_it_came_from( + tmp_path: pathlib.Path, case: EndpointCase, expected: str +) -> None: + env = case.environment(tmp_path) + + found = load_settings(env=env) + + wanted = env["TOC_CONFIG_FILE"] if expected == "file" else expected + assert found.endpoint_source == wanted diff --git a/tmxcaliber/cli.py b/tmxcaliber/cli.py index 6e9ad17..278646c 100644 --- a/tmxcaliber/cli.py +++ b/tmxcaliber/cli.py @@ -22,7 +22,6 @@ from .lib.filter_applier import FilterApplier from .lib.remote.errors import RemoteError from .lib.remote.ref import is_remote_ref -from .lib.remote.resolve import resolve_source from .lib.scf import get_scf_data from .lib.threatmodel_data import ( ThreatModelData, @@ -44,6 +43,7 @@ INCOMPLETE_EXIT, add_api_parsers, add_init_parser, + resolve_command_source, run_api_command, run_init, selected_route, @@ -376,7 +376,7 @@ def get_recursive_json_file_paths(source: str) -> list[str]: def get_service_rows(source: str) -> list[dict[str, str]]: service_rows: list[dict[str, str]] = [] - for json_file_path in get_recursive_json_file_paths(resolve_source(source)): + for json_file_path in get_recursive_json_file_paths(resolve_command_source(source)): data = load_json_data(json_file_path) metadata_block = data.get("metadata", {}) if not isinstance(metadata_block, dict): @@ -405,7 +405,7 @@ def get_service_rows(source: str) -> list[dict[str, str]]: def get_feature_class_rows(source: str) -> list[dict[str, str]]: - data = load_json_data(resolve_source(source)) + data = load_json_data(resolve_command_source(source)) feature_classes = data.get("feature_classes", {}) if not isinstance(feature_classes, dict): return [] @@ -449,7 +449,7 @@ def get_input_data( # A reference becomes a cached path here; a path is handed back # unchanged, so every existing invocation reaches the same code it # always did. - source = resolve_source(source) + source = resolve_command_source(source) if not os.path.exists(source): print(f"File or directory not found: {source}") sys.exit(1) diff --git a/tmxcaliber/lib/remote/config.py b/tmxcaliber/lib/remote/config.py index 2426ac3..c19e467 100644 --- a/tmxcaliber/lib/remote/config.py +++ b/tmxcaliber/lib/remote/config.py @@ -28,6 +28,12 @@ #: The API this client talks to, unless the environment or the file says otherwise. DEFAULT_BASE_URL: Final[str] = "https://api.trustoncloud.com" +#: The environment variable holding a key, and the source a key from it reports. +KEY_VARIABLE: Final[str] = "TOC_API_KEY" + +#: The environment variable holding an endpoint, and the source it reports. +URL_VARIABLE: Final[str] = "TOC_API_URL" + #: Seconds to wait on a single request before giving up. #: #: `lib/cache.py` sets no timeout at all, which is the mistake not to repeat: @@ -103,11 +109,37 @@ class Settings: credentials: The resolved API key. base_url: The API root, without a trailing slash. timeout: Seconds to wait on one request. + endpoint_source: Where ``base_url`` came from, for diagnostics: + ``TOC_API_URL``, the credentials file's path, or empty for the + default. """ credentials: Credentials base_url: str = DEFAULT_BASE_URL timeout: float = DEFAULT_TIMEOUT + endpoint_source: str = "" + + @property + def sends_environment_key_to_stored_endpoint(self) -> bool: + """Whether a key from the environment goes to an endpoint from the file. + + The key and the endpoint are resolved independently (see + `load_settings`), so exporting ``TOC_API_KEY`` without ``TOC_API_URL`` + keeps an endpoint that `init` stored earlier. That is deliberate, and + also easy to miss when the key was meant for another API, so this is + the condition a caller should be told about. A stored endpoint equal to + the default is excluded: ``base_url`` carries no trailing slash, so + ``https://api.trustoncloud.com/`` compares equal to the default. + + Returns: + True when the key came from ``TOC_API_KEY``, the endpoint came + from the credentials file, and that endpoint is not the default. + """ + return ( + self.credentials.source == KEY_VARIABLE + and self.endpoint_source not in ("", URL_VARIABLE) + and self.base_url != DEFAULT_BASE_URL + ) def config_path(env: Mapping[str, str]) -> pathlib.Path: @@ -305,16 +337,27 @@ def load_settings( # silently moved a configured custom endpoint back to the default, so # a caller could verify one API and then call another. stored = _from_file(environ) - url = environ.get("TOC_API_URL", "").strip() or (stored[1] if stored else "") - - key = environ.get("TOC_API_KEY", "").strip() - source = "TOC_API_KEY" + url = environ.get(URL_VARIABLE, "").strip() + endpoint_source = URL_VARIABLE if url else "" + if not url and stored is not None and stored[1]: + # Where the endpoint came from is recorded rather than printed, so + # the CLI can name the file that chose it. See `Settings`. + url, endpoint_source = stored[1], stored[2] + + key = environ.get(KEY_VARIABLE, "").strip() + source = KEY_VARIABLE if not key: if stored is None: raise ConfigurationError(_HELP, code="no_credential") key, _, source = stored - return settings_for(key, api_url=url, source=source, timeout=timeout) + return settings_for( + key, + api_url=url, + source=source, + endpoint_source=endpoint_source, + timeout=timeout, + ) def settings_for( @@ -322,6 +365,7 @@ def settings_for( *, api_url: str = "", source: str = "", + endpoint_source: str = "", timeout: float = DEFAULT_TIMEOUT, ) -> Settings: """Build settings from a credential given explicitly. @@ -335,6 +379,8 @@ def settings_for( api_key: The key. api_url: The endpoint, or empty for the default. source: Where the key came from, for diagnostics. + endpoint_source: Where the endpoint came from, for diagnostics, or + empty when ``api_url`` is. timeout: Seconds to wait on one request. Returns: @@ -358,6 +404,7 @@ def settings_for( credentials=Credentials(api_key=api_key, source=source), base_url=base_url, timeout=timeout, + endpoint_source=endpoint_source, ) diff --git a/tmxcaliber/lib/remote/resolve.py b/tmxcaliber/lib/remote/resolve.py index 02f92b7..46b242b 100644 --- a/tmxcaliber/lib/remote/resolve.py +++ b/tmxcaliber/lib/remote/resolve.py @@ -14,7 +14,7 @@ from __future__ import annotations import os -from collections.abc import Mapping +from collections.abc import Callable, Mapping from .assemble import fetch_document from .cache import DEFAULT_MAX_AGE, PINNED_MAX_AGE, cached_path, read, write @@ -42,6 +42,7 @@ def fetch_to_cache( ref: TmRef, *, client: TocClient | None = None, + connect: Callable[[], TocClient] | None = None, refresh: bool = False, env: Mapping[str, str] | None = None, ) -> str: @@ -55,7 +56,10 @@ def fetch_to_cache( Args: ref: The ThreatModel, optionally pinned to a release. - client: A client to use, built from the environment when omitted. + client: A client to use. Wins over ``connect``. + connect: Builds the client when ``client`` is omitted, so a caller + can see the resolved settings before any request. Resolved from + ``env`` when both are omitted. refresh: Fetch even when a cached copy would do. env: The environment to read. @@ -65,7 +69,7 @@ def fetch_to_cache( Raises: RemoteError: On any transport, credential or server failure. """ - api = client or _client(env) + api = client or (connect() if connect is not None else _client(env)) if ref.release: path = cached_path(ref, ref.release, api.base_url, api.key_id, env) @@ -96,6 +100,7 @@ def resolve_source( source: str, *, client: TocClient | None = None, + connect: Callable[[], TocClient] | None = None, refresh: bool = False, env: Mapping[str, str] | None = None, ) -> str: @@ -103,7 +108,10 @@ def resolve_source( Args: source: A filesystem path, a directory, or a ThreatModel reference. - client: A client to use, built from the environment when omitted. + client: A client to use. Wins over ``connect``. + connect: Builds the client when ``client`` is omitted and the source + is a reference. Never called for a path, so a local file still + needs no credential. Resolved from ``env`` when both are omitted. refresh: Fetch even when a cached copy would do. env: The environment to read. @@ -123,4 +131,6 @@ def resolve_source( # Not a path and not a reference. Left exactly as it came, so the # caller's own validation produces the message it always produced. return source - return fetch_to_cache(parse_ref(source), client=client, refresh=refresh, env=env) + return fetch_to_cache( + parse_ref(source), client=client, connect=connect, refresh=refresh, env=env + ) diff --git a/tmxcaliber/remote_cli.py b/tmxcaliber/remote_cli.py index 382fc06..72da658 100644 --- a/tmxcaliber/remote_cli.py +++ b/tmxcaliber/remote_cli.py @@ -48,6 +48,7 @@ ) from .lib.remote.errors import ConfigurationError, RemoteError from .lib.remote.ref import parse_ref +from .lib.remote.resolve import resolve_source #: The namespace key each nesting level of the command tree parks its word in. #: @@ -283,6 +284,77 @@ def _report_envelope(envelope: Mapping[str, Any]) -> None: ) +#: Stored-endpoint notes this process has already written. A process is one +#: invocation, and `create-change-log` can resolve two references in it; +#: the same note twice would read like two different problems. +_noted: set[tuple[str, str]] = set() + + +def _note_stored_endpoint(settings: Settings) -> None: + """Say, on stderr, that an environment key is going to a stored endpoint. + + The endpoint is resolved independently of the key, so a key exported in + TOC_API_KEY goes to whatever endpoint `init` stored unless TOC_API_URL + names another. That precedence is deliberate and stays; this only makes + it visible. Written before any request, so the reader learns it even + when the call then fails, and on stderr, so JSON on stdout is untouched. + + Args: + settings: The settings a command is about to use. + """ + if not settings.sends_environment_key_to_stored_endpoint: + return + said = (settings.base_url, settings.endpoint_source) + if said in _noted: + return + _noted.add(said) + print( + Fore.YELLOW + + f"Note: the key in TOC_API_KEY will be sent to {settings.base_url}, " + f"the endpoint stored in {settings.endpoint_source}. Set TOC_API_URL " + "to choose a different endpoint." + Fore.RESET, + file=sys.stderr, + ) + + +def connect(env: Mapping[str, str] | None = None) -> TocClient: + """Build the client a command uses when none was given. + + **Every command that resolves settings from the environment and the file + comes through here**, the API commands and a ThreatModel reference given + as a document source alike, so what the reader is told about those + settings is said in one place. `init` is the deliberate exception: it + verifies the key it just stored rather than what resolution would pick. + + Args: + env: The environment to read, defaulting to the real one. + + Returns: + A client over the resolved settings. + + Raises: + ConfigurationError: If no usable credential is configured. + """ + settings = load_settings(env=env) + _note_stored_endpoint(settings) + return TocClient(settings) + + +def resolve_command_source(source: str) -> str: + """Resolve a document command's source, reaching the API through `connect`. + + Args: + source: A filesystem path, a directory, or a ThreatModel reference. + + Returns: + A path. The input unchanged when it was already one. + + Raises: + RemoteError: On any transport, credential or server failure. + """ + return resolve_source(source, connect=connect) + + def run_api_command( params: Namespace, *, @@ -307,7 +379,7 @@ def run_api_command( """ route = selected_route(params) assert route is not None, "run_api_command called for a non-API command" - api = client or TocClient(load_settings()) + api = client or connect() path = _path_for(route, params) query = {name: str(getattr(params, name, "") or "") for name in route.filters} diff --git a/uv.lock b/uv.lock index 54bddbd..4230bc6 100644 --- a/uv.lock +++ b/uv.lock @@ -929,7 +929,7 @@ wheels = [ [[package]] name = "tmxcaliber" -version = "0.5.3" +version = "0.5.4" source = { editable = "." } dependencies = [ { name = "beautifulsoup4" },