scriptoriumadmin/strix
1
0
Fork 0
mirror of https://github.com/usestrix/strix.git synced 2026-09-09 22:31:07 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Harden cloud CLI type boundaries

Browse source
This commit is contained in:
bearsyankees 2026-08-28 09:22:18 -04:00
parent 45353743b8
commit 6a8c89f128
7 changed files with 46 additions and 27 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
strix/interface/cloud/http.py
View file

@ -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

11
strix/interface/cloud/runner.py
View file

@ -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"):

9
strix/interface/cloud/source_scan.py
View file

@ -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,

27
strix/interface/cloud/workspaces.py
View file

@ -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()

4
strix/interface/main.py
View file

@ -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.

16
strix/interface/platform_cli.py
View file

@ -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

4
strix/report/dedupe.py
View file

@ -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