From 6a8c89f128cde41a338a61acdb2d3d415406e002 Mon Sep 17 00:00:00 2001 From: bearsyankees Date: Fri, 28 Aug 2026 09:22:18 -0400 Subject: [PATCH] Harden cloud CLI type boundaries --- strix/interface/cloud/http.py | 2 +- strix/interface/cloud/runner.py | 11 +++++++---- strix/interface/cloud/source_scan.py | 9 +++++---- strix/interface/cloud/workspaces.py | 27 ++++++++++++++++++--------- strix/interface/main.py | 4 ++-- strix/interface/platform_cli.py | 16 +++++++++++----- strix/report/dedupe.py | 4 ++-- 7 files changed, 46 insertions(+), 27 deletions(-) diff --git a/strix/interface/cloud/http.py b/strix/interface/cloud/http.py index 56f3a4d3..6c4da3ff 100644 --- a/strix/interface/cloud/http.py +++ b/strix/interface/cloud/http.py @@ -71,7 +71,7 @@ def app_url() -> str: viewer = load_settings().viewer configured = viewer.app_url.rstrip("/") explicitly_configured = bool(os.environ.get("STRIX_APP_URL")) or "app_url" in getattr( - viewer, "model_fields_set", set() + viewer, "model_fields_set", set[str]() ) if explicitly_configured or _token_override_active or os.environ.get("STRIX_API_TOKEN"): return configured diff --git a/strix/interface/cloud/runner.py b/strix/interface/cloud/runner.py index 717ac0fa..1d813f87 100644 --- a/strix/interface/cloud/runner.py +++ b/strix/interface/cloud/runner.py @@ -809,8 +809,10 @@ def _ambiguous_scan_launch_error( exit_code = http.EXIT_ERROR if isinstance(error, http.CloudError): exit_code = error.exit_code - if isinstance(error.payload, dict): - payload.update(cast("dict[str, Any]", error.payload)) + raw_payload: Any = error.payload + error_payload = cast("dict[str, Any]", raw_payload) + if isinstance(raw_payload, dict): + payload.update(error_payload) payload["error"] = message _attach_idempotency_recovery(payload, idempotency_key) return http.CloudError(message, exit_code=exit_code, payload=payload) @@ -1011,8 +1013,9 @@ def _emit_error( console: Console, exc: http.CloudError, *, as_json: bool, to_stderr: bool = False ) -> None: if as_json: - if isinstance(exc.payload, dict): - error_payload = cast("dict[str, Any]", exc.payload) + raw_payload: Any = exc.payload + error_payload = cast("dict[str, Any]", raw_payload) + if isinstance(raw_payload, dict): payload = dict(error_payload) payload.setdefault("error", str(exc)) if payload.get("detail") == payload.get("error"): diff --git a/strix/interface/cloud/source_scan.py b/strix/interface/cloud/source_scan.py index 408c10d1..a3698c96 100644 --- a/strix/interface/cloud/source_scan.py +++ b/strix/interface/cloud/source_scan.py @@ -349,11 +349,12 @@ def _retained_source_upload_error( exit_code = http.EXIT_ERROR if isinstance(error, http.CloudError): exit_code = error.exit_code - if isinstance(error.payload, dict): - error_payload = cast("dict[str, Any]", error.payload) + raw_payload: Any = error.payload + error_payload = cast("dict[str, Any]", raw_payload) + if isinstance(raw_payload, dict): payload.update(error_payload) - elif error.payload is not None: - payload["detail"] = error.payload + elif raw_payload is not None: + payload["detail"] = raw_payload payload.update( { "error": message, diff --git a/strix/interface/cloud/workspaces.py b/strix/interface/cloud/workspaces.py index 133ec92a..62e97136 100644 --- a/strix/interface/cloud/workspaces.py +++ b/strix/interface/cloud/workspaces.py @@ -86,12 +86,13 @@ def _use(console: Console, args: argparse.Namespace, *, as_json: bool) -> int: body["scopes"] = args.scopes elif not external_token: stored_scopes = stored_record.get("requested_scopes", stored_record.get("scopes")) + stored_scope_items = cast("list[Any]", cast("Any", stored_scopes)) if ( isinstance(stored_scopes, list) - and stored_scopes - and all(isinstance(scope, str) for scope in stored_scopes) + and stored_scope_items + and all(isinstance(scope, str) for scope in stored_scope_items) ): - body["scopes"] = [scope for scope in stored_scopes if isinstance(scope, str)] + body["scopes"] = [scope for scope in stored_scope_items if isinstance(scope, str)] switched = _switch_workspace_token( str(workspace["id"]), token=args.token, @@ -104,10 +105,12 @@ def _use(console: Console, args: argparse.Namespace, *, as_json: bool) -> int: if not isinstance(switched_token, str) or not switched_token.strip(): raise _workspace_switch_unknown("the platform response omitted the token") switched_scopes = switched_record.get("scopes") + switched_scope_items = cast("list[Any]", cast("Any", switched_scopes)) if not isinstance(switched_scopes, list) or not all( - isinstance(scope, str) for scope in switched_scopes + isinstance(scope, str) for scope in switched_scope_items ): raise _workspace_switch_unknown("the platform response contained invalid scopes") + validated_scopes = cast("list[str]", switched_scope_items) record.update( { @@ -117,7 +120,7 @@ def _use(console: Console, args: argparse.Namespace, *, as_json: bool) -> int: "organization_name", workspace.get("name", "") ), "expires_at": switched_record.get("expires_at"), - "scopes": switched_scopes, + "scopes": validated_scopes, "requested_scopes": ( list(args.scopes) if args.scopes @@ -127,7 +130,7 @@ def _use(console: Console, args: argparse.Namespace, *, as_json: bool) -> int: stored_record.get("scopes", []), ) if not external_token - else switched_scopes + else validated_scopes ) ), "app_url": http.app_url(), @@ -161,7 +164,8 @@ def _use(console: Console, args: argparse.Namespace, *, as_json: bool) -> int: console.print(f"[green]✓ Switched to workspace [bold]{workspace_name}[/].[/]") scopes = record.get("scopes") if isinstance(scopes, list) and scopes: - scope_names = [scope for scope in scopes if isinstance(scope, str)] + scope_items = cast("list[Any]", cast("Any", scopes)) + scope_names = [scope for scope in scope_items if isinstance(scope, str)] if scope_names: rendered_scopes = escape(sanitize_terminal_text(" ".join(scope_names))) console.print(f" Scopes: [dim]{rendered_scopes}[/]") @@ -211,7 +215,9 @@ def _workspace_switch_unknown(detail: str) -> http.CloudError: def _emit_cloud_error(console: Console, error: http.CloudError, *, as_json: bool) -> None: if as_json: - payload = dict(error.payload) if isinstance(error.payload, dict) else {} + raw_payload: Any = error.payload + error_payload = cast("dict[str, Any]", raw_payload) + payload = dict(error_payload) if isinstance(raw_payload, dict) else {} payload["error"] = str(error) emit(console, payload, as_json=True) return @@ -222,7 +228,10 @@ def _find_workspace(selector: str, *, token: str | None) -> dict[str, Any]: listed = http.check(http.request("GET", "/workspaces", token=token)) listed_record = cast("dict[str, Any]", listed) if isinstance(listed, dict) else {} items = listed_record.get("workspaces") - workspaces = [cast("dict[str, Any]", item) for item in (items or []) if isinstance(item, dict)] + item_values = cast("list[Any]", cast("Any", items)) if isinstance(items, list) else [] + workspaces = [ + cast("dict[str, Any]", cast("Any", item)) for item in item_values if isinstance(item, dict) + ] if not workspaces: raise http.CloudError("no workspaces found for this account.") wanted = selector.strip() diff --git a/strix/interface/main.py b/strix/interface/main.py index 1f3d21ad..4885cc8f 100644 --- a/strix/interface/main.py +++ b/strix/interface/main.py @@ -217,12 +217,12 @@ async def warm_up_llm(show_model_warning: bool = True) -> None: logger.info("LLM warm-up succeeded for model %s", (llm.model or "").strip()) if settings.dedupe.model: - from strix.report.dedupe import _dedupe_extra_args + from strix.report.dedupe import dedupe_extra_args dedupe_model = settings.dedupe.model.strip() raw_model = dedupe_model deduper = StrixProvider().get_model(dedupe_model) - deduper_extra = _dedupe_extra_args(settings.dedupe) + deduper_extra = dedupe_extra_args(settings.dedupe) # A dedicated dedupe model may route to another provider, which must # never receive the main endpoint's headers; it has its own # DEDUPE_LLM_EXTRA_HEADERS. diff --git a/strix/interface/platform_cli.py b/strix/interface/platform_cli.py index 7ae8b739..8c05d17e 100644 --- a/strix/interface/platform_cli.py +++ b/strix/interface/platform_cli.py @@ -333,8 +333,9 @@ def _bind_login_record( (parsed.scheme.lower(), parsed.netloc.lower(), parsed.path.rstrip("/"), "", "") ) preference: Any = requested_scopes if requested_scopes is not None else record.get("scopes") - if isinstance(preference, list) and all(isinstance(scope, str) for scope in preference): - bound["requested_scopes"] = list(dict.fromkeys(cast("list[str]", preference))) + preference_items = cast("list[Any]", preference) + if isinstance(preference, list) and all(isinstance(scope, str) for scope in preference_items): + bound["requested_scopes"] = list(dict.fromkeys(cast("list[str]", preference_items))) return bound @@ -385,7 +386,8 @@ def _complete_selection( def _dict_items(value: Any) -> list[dict[str, Any]]: if not isinstance(value, list): return [] - return [item for item in value if isinstance(item, dict)] + items = cast("list[Any]", cast("Any", value)) + return [cast("dict[str, Any]", cast("Any", item)) for item in items if isinstance(item, dict)] def _choose_workspace( @@ -543,7 +545,8 @@ def _print_success(console: Console, record: dict[str, Any]) -> None: console.print(f" Workspace: [bold]{_terminal_markup(organization)}[/]") scopes = record.get("scopes") if isinstance(scopes, list) and scopes: - rendered_scopes = _terminal_markup(" ".join(str(s) for s in scopes)) + scope_items = cast("list[Any]", cast("Any", scopes)) + rendered_scopes = _terminal_markup(" ".join(str(scope) for scope in scope_items)) console.print(f" Scopes: [dim]{rendered_scopes}[/]") console.print(f" Token: stored in [dim]{_terminal_markup(AUTH_PATH)}[/]") console.print() @@ -603,7 +606,10 @@ def _status(console: Console, argv: list[str]) -> int: console.print(f" Platform: {_terminal_markup(record['app_url'])}") scopes = record.get("scopes") if isinstance(scopes, list) and scopes: - console.print(f" Scopes: {_terminal_markup(' '.join(str(scope) for scope in scopes))}") + scope_items = cast("list[Any]", cast("Any", scopes)) + console.print( + f" Scopes: {_terminal_markup(' '.join(str(scope) for scope in scope_items))}" + ) return 0 diff --git a/strix/report/dedupe.py b/strix/report/dedupe.py index 1cc0a66a..2d528e40 100644 --- a/strix/report/dedupe.py +++ b/strix/report/dedupe.py @@ -29,7 +29,7 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) -def _dedupe_extra_args(dedupe: DedupeSettings) -> dict[str, str]: +def dedupe_extra_args(dedupe: DedupeSettings) -> dict[str, str]: """Per-call credential + endpoint for the dedupe model. Provider env vars and the global base URL are process-wide, so a @@ -64,7 +64,7 @@ def _dedupe_model_settings( extra_headers=dedupe.extra_headers if dedupe.model else llm.extra_headers, has_tools=False, ) - extra = _dedupe_extra_args(dedupe) + extra = dedupe_extra_args(dedupe) if extra: settings = settings.resolve(ModelSettings(extra_args=extra)) return settings