mirror of
https://hubproxy.babadafafafafa.cn/https://github.com/usestrix/strix.git
synced 2026-09-21 00:23:52 +08:00
Compare commits
1 Commits
fix/writab
...
feat/cli-f
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
9d14cf96bd |
@@ -17,6 +17,8 @@ strix cloud session # verify the remote session and consen
|
||||
strix cloud logout # revoke remotely, then remove the local token
|
||||
```
|
||||
|
||||
On your first sign-in in a terminal, the CLI asks for a workspace name and offers to open the GitHub App installation page. Press Enter to keep the generated name. After each sign-in, the CLI lists the next steps for the workspace, for example the rename command, the GitHub installation link, or the first scan. Without a terminal, the CLI shows no prompts and only prints the steps.
|
||||
|
||||
A browser sign-in creates one reusable credential for each CLI installation. A second sign-in on the same installation replaces the secret instead of adding another key. `strix cloud logout` revokes the server session before it deletes the local token. Use `--local-only` when you cannot reach the server.
|
||||
|
||||
## Scopes
|
||||
|
||||
@@ -39,6 +39,16 @@ _MAX_POLL_INTERVAL_S = 60
|
||||
_MAX_EXPIRES_IN_S = 30 * 60
|
||||
|
||||
_ROLE_RANK = {"viewer": 0, "analyst": 1, "admin": 2}
|
||||
_MAX_WORKSPACE_NAME_LENGTH = 100
|
||||
|
||||
# Sign-in guidance is shown once and never written to the credential file.
|
||||
_LOGIN_GUIDANCE_KEYS = (
|
||||
"is_new_user",
|
||||
"onboarding",
|
||||
"next_steps",
|
||||
"next_steps_hint",
|
||||
"dashboard_url",
|
||||
)
|
||||
|
||||
|
||||
class PlatformAuthError(Exception):
|
||||
@@ -172,6 +182,7 @@ def _login(console: Console, argv: list[str]) -> int:
|
||||
console.print("\n[yellow]Sign-in cancelled.[/]")
|
||||
return 130
|
||||
|
||||
record, guidance = _split_login_guidance(record)
|
||||
try:
|
||||
save_record(record)
|
||||
except OSError as exc:
|
||||
@@ -184,10 +195,18 @@ def _login(console: Console, argv: list[str]) -> int:
|
||||
)
|
||||
return 1
|
||||
_revoke_replaced_legacy_session(previous_record, record)
|
||||
_print_success(console, record)
|
||||
_print_success(console, record, guidance)
|
||||
_offer_github_install(console, guidance, open_browser=not args.no_browser)
|
||||
return 0
|
||||
|
||||
|
||||
def _split_login_guidance(record: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
|
||||
"""Separate the one-time setup guidance from the credential that is stored."""
|
||||
guidance = {key: record[key] for key in _LOGIN_GUIDANCE_KEYS if key in record}
|
||||
stored = {key: value for key, value in record.items() if key not in guidance}
|
||||
return stored, guidance
|
||||
|
||||
|
||||
def _run_device_flow( # noqa: PLR0912, PLR0915
|
||||
console: Console,
|
||||
*,
|
||||
@@ -395,6 +414,9 @@ def _complete_selection(
|
||||
|
||||
chosen_org = _choose_workspace(console, organizations, workspace)
|
||||
role = str(chosen_org.get("role") or "admin")
|
||||
workspace_name: str | None = None
|
||||
if sys.stdin.isatty() and selection.get("is_new_user") and chosen_org.get("has_default_name"):
|
||||
workspace_name = _choose_workspace_name(console, str(chosen_org.get("name") or ""))
|
||||
chosen_scopes = scopes
|
||||
chosen_profile = scope_profile
|
||||
if chosen_scopes is None and chosen_profile is None and sys.stdin.isatty():
|
||||
@@ -404,6 +426,8 @@ def _complete_selection(
|
||||
"selection_token": selection_token,
|
||||
"organization_id": chosen_org.get("id"),
|
||||
}
|
||||
if workspace_name is not None:
|
||||
body["workspace_name"] = workspace_name
|
||||
if chosen_scopes is not None:
|
||||
body["scopes"] = chosen_scopes
|
||||
body["scope_profile"] = "custom"
|
||||
@@ -476,6 +500,29 @@ def _choose_workspace(
|
||||
console.print("[yellow]Enter a number from the list.[/]")
|
||||
|
||||
|
||||
def _choose_workspace_name(console: Console, current: str) -> str | None:
|
||||
"""Prompt a first-time user for a workspace name. None keeps the generated name."""
|
||||
console.print()
|
||||
console.print(
|
||||
"[bold]Name your workspace.[/] "
|
||||
"[dim]Teammates see this name. Press Enter to keep the default. "
|
||||
"You can change it later with `strix cloud org update --name`.[/]"
|
||||
)
|
||||
while True:
|
||||
try:
|
||||
answer = console.input(f"Workspace name ({_terminal_markup(current)}): ").strip()
|
||||
except EOFError:
|
||||
return None
|
||||
if not answer or answer == current:
|
||||
return None
|
||||
if len(answer) > _MAX_WORKSPACE_NAME_LENGTH:
|
||||
console.print(
|
||||
f"[yellow]Use a name with {_MAX_WORKSPACE_NAME_LENGTH} characters or less.[/]"
|
||||
)
|
||||
continue
|
||||
return answer
|
||||
|
||||
|
||||
def _choose_scopes(
|
||||
console: Console, catalog: list[dict[str, Any]], role: str
|
||||
) -> tuple[str, list[str] | None]:
|
||||
@@ -632,19 +679,30 @@ def _revoke_replaced_legacy_session(
|
||||
)
|
||||
|
||||
|
||||
def _print_success(console: Console, record: dict[str, Any]) -> None:
|
||||
def _print_success(
|
||||
console: Console, record: dict[str, Any], guidance: dict[str, Any] | None = None
|
||||
) -> None:
|
||||
guidance = guidance or {}
|
||||
onboarding = _onboarding_state(guidance)
|
||||
email = record.get("email", "")
|
||||
organization = record.get("organization_name") or record.get("organization_id", "")
|
||||
console.print()
|
||||
console.print("[green]✓ Signed in to the Strix platform.[/]")
|
||||
if guidance.get("is_new_user"):
|
||||
console.print("[green]✓ Welcome to Strix. Your account and workspace are ready.[/]")
|
||||
else:
|
||||
console.print("[green]✓ Signed in to the Strix platform.[/]")
|
||||
if email:
|
||||
console.print(f" Account: [bold]{_terminal_markup(email)}[/]")
|
||||
if organization:
|
||||
console.print(f" Workspace: [bold]{_terminal_markup(organization)}[/]")
|
||||
default_note = (
|
||||
" [dim](default name)[/]" if onboarding.get("workspace_named") is False else ""
|
||||
)
|
||||
console.print(f" Workspace: [bold]{_terminal_markup(organization)}[/]{default_note}")
|
||||
scopes = record.get("scopes")
|
||||
if isinstance(scopes, list) and scopes:
|
||||
console.print(f" Access: [dim]{_terminal_markup(_scope_summary(record))}[/]")
|
||||
console.print(f" Token: stored in [dim]{_terminal_markup(AUTH_PATH)}[/]")
|
||||
_print_next_steps(console, guidance)
|
||||
console.print()
|
||||
console.print(
|
||||
"[dim]The managed platform is ready. Run `strix cloud` to list the commands. "
|
||||
@@ -652,6 +710,63 @@ def _print_success(console: Console, record: dict[str, Any]) -> None:
|
||||
)
|
||||
|
||||
|
||||
def _onboarding_state(guidance: dict[str, Any]) -> dict[str, Any]:
|
||||
onboarding = guidance.get("onboarding")
|
||||
if not isinstance(onboarding, dict):
|
||||
return {}
|
||||
return cast("dict[str, Any]", onboarding)
|
||||
|
||||
|
||||
def _github_install_url(guidance: dict[str, Any]) -> str | None:
|
||||
url = _onboarding_state(guidance).get("github_install_url")
|
||||
return url if is_safe_web_url(url) else None
|
||||
|
||||
|
||||
def _print_next_steps(console: Console, guidance: dict[str, Any]) -> None:
|
||||
"""Render the server-provided setup steps for a person."""
|
||||
steps = [
|
||||
step
|
||||
for step in _dict_items(guidance.get("next_steps"))
|
||||
if isinstance(step.get("action"), str) and step.get("action")
|
||||
]
|
||||
github_url = _github_install_url(guidance)
|
||||
if not steps and github_url is None:
|
||||
return
|
||||
console.print()
|
||||
console.print("[bold]Next steps:[/]")
|
||||
for index, step in enumerate(steps, start=1):
|
||||
console.print(f" [cyan]{index}[/]. {_terminal_markup(step['action'])}")
|
||||
cli = step.get("cli")
|
||||
if isinstance(cli, str) and cli:
|
||||
console.print(f" [dim]{_terminal_markup(cli)}[/]")
|
||||
if github_url is not None:
|
||||
console.print()
|
||||
console.print("Install the Strix GitHub App to connect repositories:")
|
||||
console.print(sanitize_terminal_text(github_url), markup=False, soft_wrap=True)
|
||||
|
||||
|
||||
def _offer_github_install(
|
||||
console: Console, guidance: dict[str, Any], *, open_browser: bool
|
||||
) -> None:
|
||||
"""On a first sign-in in a terminal, offer to open the GitHub App installation page."""
|
||||
github_url = _github_install_url(guidance)
|
||||
if github_url is None or not open_browser or not guidance.get("is_new_user"):
|
||||
return
|
||||
if not sys.stdin.isatty() or not sys.stdout.isatty():
|
||||
return
|
||||
console.print()
|
||||
try:
|
||||
answer = console.input("Open the GitHub App installation page now? [y/N]: ").strip()
|
||||
except EOFError:
|
||||
return
|
||||
if answer.casefold() not in {"y", "yes"}:
|
||||
console.print("[dim]Run `strix cloud integrations install github` when you are ready.[/]")
|
||||
return
|
||||
with contextlib.suppress(Exception):
|
||||
webbrowser.open(github_url)
|
||||
console.print("[dim]Approve the installation in the browser. Then run `strix cloud repos`.[/]")
|
||||
|
||||
|
||||
def _status(console: Console, argv: list[str]) -> int: # noqa: PLR0912
|
||||
parser = _SessionArgumentParser(
|
||||
prog="strix cloud whoami",
|
||||
|
||||
@@ -7,6 +7,7 @@ import io
|
||||
import json
|
||||
import sys
|
||||
import time
|
||||
import webbrowser
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
import pytest
|
||||
@@ -862,6 +863,281 @@ def test_login_workspace_selector_prefers_ids_and_rejects_duplicate_names() -> N
|
||||
platform_cli._choose_workspace(console, organizations, "Strix")
|
||||
|
||||
|
||||
_NEW_USER_ONBOARDING: dict[str, Any] = {
|
||||
"workspace_named": False,
|
||||
"repositories_connected": False,
|
||||
"domains_added": False,
|
||||
"github_install_url": "https://github.com/apps/strix/installations/new?state=abc",
|
||||
}
|
||||
|
||||
_NEW_USER_LOGIN: dict[str, Any] = {
|
||||
"api_token": "strix_pat_test",
|
||||
"email": "alex@example.com",
|
||||
"organization_id": "org-1",
|
||||
"organization_name": "Alex's Workspace",
|
||||
"scopes": ["scans:read"],
|
||||
"is_new_user": True,
|
||||
"onboarding": _NEW_USER_ONBOARDING,
|
||||
"next_steps_hint": "Run the steps in order.",
|
||||
"next_steps": [
|
||||
{
|
||||
"action": "Name your workspace",
|
||||
"cli": 'strix cloud org update --name "Acme Security"',
|
||||
"method": "PATCH",
|
||||
"path": "/api/v1/organization",
|
||||
},
|
||||
{
|
||||
"action": "Add a domain",
|
||||
"cli": "strix cloud domains add --domain example.com --asset-type web_app",
|
||||
"method": "POST",
|
||||
"path": "/api/v1/domains",
|
||||
},
|
||||
],
|
||||
"dashboard_url": "https://app.strix.ai",
|
||||
}
|
||||
|
||||
|
||||
def test_login_guidance_is_shown_once_and_never_stored() -> None:
|
||||
stored, guidance = platform_cli._split_login_guidance(dict(_NEW_USER_LOGIN))
|
||||
|
||||
assert stored["api_token"] == _NEW_USER_LOGIN["api_token"]
|
||||
assert stored["organization_name"] == "Alex's Workspace"
|
||||
assert set(guidance) == {
|
||||
"is_new_user",
|
||||
"onboarding",
|
||||
"next_steps",
|
||||
"next_steps_hint",
|
||||
"dashboard_url",
|
||||
}
|
||||
assert not any(key in stored for key in guidance)
|
||||
|
||||
|
||||
def test_new_user_success_renders_setup_steps_and_github_link() -> None:
|
||||
output = io.StringIO()
|
||||
console = Console(file=output, width=120)
|
||||
stored, guidance = platform_cli._split_login_guidance(dict(_NEW_USER_LOGIN))
|
||||
|
||||
platform_cli._print_success(console, stored, guidance)
|
||||
|
||||
rendered = output.getvalue()
|
||||
assert "Welcome to Strix" in rendered
|
||||
assert "Alex's Workspace" in rendered
|
||||
assert "(default name)" in rendered
|
||||
assert "Next steps:" in rendered
|
||||
assert "1. Name your workspace" in rendered
|
||||
assert 'strix cloud org update --name "Acme Security"' in rendered
|
||||
assert "2. Add a domain" in rendered
|
||||
assert "https://github.com/apps/strix/installations/new?state=abc" in rendered
|
||||
|
||||
|
||||
def test_returning_user_success_stays_compact_without_guidance() -> None:
|
||||
output = io.StringIO()
|
||||
console = Console(file=output, width=120)
|
||||
|
||||
platform_cli._print_success(
|
||||
console,
|
||||
{"email": "alex@example.com", "organization_name": "Acme", "scopes": ["scans:read"]},
|
||||
)
|
||||
|
||||
rendered = output.getvalue()
|
||||
assert "Signed in to the Strix platform." in rendered
|
||||
assert "Welcome" not in rendered
|
||||
assert "Next steps" not in rendered
|
||||
assert "default name" not in rendered
|
||||
|
||||
|
||||
def test_success_ignores_unsafe_github_link_and_malformed_steps() -> None:
|
||||
output = io.StringIO()
|
||||
console = Console(file=output, width=120)
|
||||
|
||||
platform_cli._print_success(
|
||||
console,
|
||||
{"organization_name": "Acme"},
|
||||
{
|
||||
"onboarding": {"github_install_url": "javascript:alert(1)"},
|
||||
"next_steps": ["not a step", {"cli": "strix cloud"}],
|
||||
},
|
||||
)
|
||||
|
||||
rendered = output.getvalue()
|
||||
assert "javascript:" not in rendered
|
||||
assert "Next steps" not in rendered
|
||||
|
||||
|
||||
def test_first_login_prompts_for_a_workspace_name_and_sends_it(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
monkeypatch.setattr(sys.stdin, "isatty", lambda: True)
|
||||
sent: dict[str, Any] = {}
|
||||
|
||||
def post(_url: str, **kwargs: Any) -> FakeResponse:
|
||||
sent.update(kwargs["json"])
|
||||
return FakeResponse(_NEW_USER_LOGIN)
|
||||
|
||||
monkeypatch.setattr(requests, "post", post)
|
||||
console = Console(file=io.StringIO(), width=120)
|
||||
answers = iter(["Acme Security", "1"])
|
||||
console.input = lambda *_args, **_kwargs: next(answers) # type: ignore[method-assign]
|
||||
|
||||
record = platform_cli._complete_selection(
|
||||
console,
|
||||
"https://example.test",
|
||||
{
|
||||
"selection_token": "sel-1",
|
||||
"is_new_user": True,
|
||||
"organizations": [
|
||||
{
|
||||
"id": "org-1",
|
||||
"name": "Alex's Workspace",
|
||||
"role": "admin",
|
||||
"has_default_name": True,
|
||||
}
|
||||
],
|
||||
"scopes": [{"scope": "scans:read", "min_role": "viewer", "minimum": True}],
|
||||
},
|
||||
scopes=None,
|
||||
scope_profile=None,
|
||||
workspace=None,
|
||||
)
|
||||
|
||||
assert sent["workspace_name"] == "Acme Security"
|
||||
assert sent["organization_id"] == "org-1"
|
||||
assert record["api_token"] == _NEW_USER_LOGIN["api_token"]
|
||||
|
||||
|
||||
def test_first_login_keeps_the_default_name_on_enter(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(sys.stdin, "isatty", lambda: True)
|
||||
sent: dict[str, Any] = {}
|
||||
|
||||
def post(_url: str, **kwargs: Any) -> FakeResponse:
|
||||
sent.update(kwargs["json"])
|
||||
return FakeResponse(_NEW_USER_LOGIN)
|
||||
|
||||
monkeypatch.setattr(requests, "post", post)
|
||||
console = Console(file=io.StringIO(), width=120)
|
||||
console.input = lambda *_args, **_kwargs: "" # type: ignore[method-assign]
|
||||
|
||||
platform_cli._complete_selection(
|
||||
console,
|
||||
"https://example.test",
|
||||
{
|
||||
"selection_token": "sel-1",
|
||||
"is_new_user": True,
|
||||
"organizations": [
|
||||
{
|
||||
"id": "org-1",
|
||||
"name": "Alex's Workspace",
|
||||
"role": "admin",
|
||||
"has_default_name": True,
|
||||
}
|
||||
],
|
||||
"scopes": [],
|
||||
},
|
||||
scopes=None,
|
||||
scope_profile="recommended",
|
||||
workspace=None,
|
||||
)
|
||||
|
||||
assert "workspace_name" not in sent
|
||||
|
||||
|
||||
def test_workspace_name_prompt_rejects_overlong_names() -> None:
|
||||
output = io.StringIO()
|
||||
console = Console(file=output, width=120)
|
||||
answers = iter(["x" * 101, "Acme"])
|
||||
console.input = lambda *_args, **_kwargs: next(answers) # type: ignore[method-assign]
|
||||
|
||||
assert platform_cli._choose_workspace_name(console, "Alex's Workspace") == "Acme"
|
||||
assert "100 characters or less" in output.getvalue()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("is_new_user", "has_default_name", "tty"),
|
||||
[(False, True, True), (True, False, True), (True, True, False)],
|
||||
)
|
||||
def test_workspace_name_prompt_is_skipped_for_returning_named_or_noninteractive_logins(
|
||||
monkeypatch: pytest.MonkeyPatch, is_new_user: bool, has_default_name: bool, tty: bool
|
||||
) -> None:
|
||||
monkeypatch.setattr(sys.stdin, "isatty", lambda: tty)
|
||||
sent: dict[str, Any] = {}
|
||||
|
||||
def post(_url: str, **kwargs: Any) -> FakeResponse:
|
||||
sent.update(kwargs["json"])
|
||||
return FakeResponse(_NEW_USER_LOGIN)
|
||||
|
||||
monkeypatch.setattr(requests, "post", post)
|
||||
console = Console(file=io.StringIO(), width=120)
|
||||
console.input = lambda *_args, **_kwargs: pytest.fail("must not prompt") # type: ignore[method-assign]
|
||||
|
||||
platform_cli._complete_selection(
|
||||
console,
|
||||
"https://example.test",
|
||||
{
|
||||
"selection_token": "sel-1",
|
||||
"is_new_user": is_new_user,
|
||||
"organizations": [
|
||||
{
|
||||
"id": "org-1",
|
||||
"name": "Alex's Workspace",
|
||||
"role": "admin",
|
||||
"has_default_name": has_default_name,
|
||||
}
|
||||
],
|
||||
"scopes": [],
|
||||
},
|
||||
scopes=["scans:read"],
|
||||
scope_profile=None,
|
||||
workspace=None,
|
||||
)
|
||||
|
||||
assert "workspace_name" not in sent
|
||||
|
||||
|
||||
def test_github_install_offer_opens_the_browser_only_on_yes(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
monkeypatch.setattr(sys.stdin, "isatty", lambda: True)
|
||||
monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
|
||||
opened: list[str] = []
|
||||
|
||||
def fake_open(url: str) -> bool:
|
||||
opened.append(url)
|
||||
return True
|
||||
|
||||
monkeypatch.setattr(webbrowser, "open", fake_open)
|
||||
guidance: dict[str, Any] = {"is_new_user": True, "onboarding": dict(_NEW_USER_ONBOARDING)}
|
||||
|
||||
output = io.StringIO()
|
||||
console = Console(file=output, width=120)
|
||||
console.input = lambda *_args, **_kwargs: "n" # type: ignore[method-assign]
|
||||
platform_cli._offer_github_install(console, guidance, open_browser=True)
|
||||
assert opened == []
|
||||
assert "strix cloud integrations install github" in output.getvalue()
|
||||
|
||||
console.input = lambda *_args, **_kwargs: "y" # type: ignore[method-assign]
|
||||
platform_cli._offer_github_install(console, guidance, open_browser=True)
|
||||
assert opened == ["https://github.com/apps/strix/installations/new?state=abc"]
|
||||
|
||||
|
||||
def test_github_install_offer_never_prompts_without_a_terminal_or_browser(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
monkeypatch.setattr(webbrowser, "open", lambda _url: pytest.fail("must not open"))
|
||||
guidance: dict[str, Any] = {"is_new_user": True, "onboarding": dict(_NEW_USER_ONBOARDING)}
|
||||
console = Console(file=io.StringIO(), width=120)
|
||||
console.input = lambda *_args, **_kwargs: pytest.fail("must not prompt") # type: ignore[method-assign]
|
||||
|
||||
monkeypatch.setattr(sys.stdin, "isatty", lambda: False)
|
||||
platform_cli._offer_github_install(console, guidance, open_browser=True)
|
||||
|
||||
monkeypatch.setattr(sys.stdin, "isatty", lambda: True)
|
||||
monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
|
||||
platform_cli._offer_github_install(console, guidance, open_browser=False)
|
||||
platform_cli._offer_github_install(
|
||||
console, {**guidance, "is_new_user": False}, open_browser=True
|
||||
)
|
||||
|
||||
|
||||
def test_device_flow_slow_down_never_exceeds_the_poll_interval_cap(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
|
||||
Reference in New Issue
Block a user