Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 92f7feaa32 |
@@ -0,0 +1,189 @@
|
||||
"""ASV benchmarks for Auth and Team CLI command throughput.
|
||||
|
||||
Measures the performance of:
|
||||
- Auth login command (token store + CLI rendering)
|
||||
- Auth logout command
|
||||
- Auth status command
|
||||
- Team list command
|
||||
- Team use command
|
||||
- TokenStore store/get/clear cycle
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
import shutil
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from unittest.mock import patch
|
||||
|
||||
# Ensure the local *source* tree is importable.
|
||||
_SRC = str(Path(__file__).resolve().parents[1] / "src")
|
||||
if _SRC not in sys.path:
|
||||
sys.path.insert(0, _SRC)
|
||||
|
||||
import cleveragents # noqa: E402
|
||||
|
||||
importlib.reload(cleveragents)
|
||||
|
||||
from typer.testing import CliRunner # noqa: E402
|
||||
|
||||
from cleveragents.cli.commands import auth as auth_mod # noqa: E402
|
||||
from cleveragents.cli.commands import team as team_mod # noqa: E402
|
||||
from cleveragents.cli.commands.auth import app as auth_app # noqa: E402
|
||||
from cleveragents.cli.commands.team import app as team_app # noqa: E402
|
||||
from cleveragents.infrastructure.auth.token_store import TokenStore # noqa: E402
|
||||
|
||||
_runner = CliRunner()
|
||||
_config_data: dict[str, Any] = {}
|
||||
|
||||
|
||||
def _make_store() -> tuple[TokenStore, Path]:
|
||||
tmpdir = Path(tempfile.mkdtemp())
|
||||
return TokenStore(token_dir=tmpdir, use_keyring=False), tmpdir
|
||||
|
||||
|
||||
def _patched_read_team() -> str | None:
|
||||
val = _config_data.get("active_team")
|
||||
if val is not None and str(val).strip():
|
||||
return str(val).strip()
|
||||
return None
|
||||
|
||||
|
||||
def _patched_write_team(team: str | None) -> None:
|
||||
global _config_data
|
||||
if team is None:
|
||||
_config_data.pop("active_team", None)
|
||||
else:
|
||||
_config_data["active_team"] = team
|
||||
|
||||
|
||||
class AuthLoginSuite:
|
||||
"""Benchmark auth login throughput."""
|
||||
|
||||
def setup(self) -> None:
|
||||
global _config_data
|
||||
_config_data = {}
|
||||
self._store, self._tmpdir = _make_store()
|
||||
|
||||
def teardown(self) -> None:
|
||||
shutil.rmtree(str(self._tmpdir), ignore_errors=True)
|
||||
|
||||
def time_auth_login(self) -> None:
|
||||
def _ps() -> TokenStore:
|
||||
return self._store
|
||||
|
||||
with (
|
||||
patch.object(auth_mod, "_get_token_store", _ps),
|
||||
patch.object(auth_mod, "_read_active_team", _patched_read_team),
|
||||
patch.object(auth_mod, "_write_active_team", _patched_write_team),
|
||||
):
|
||||
_runner.invoke(auth_app, ["login", "--token", "bench-tok"])
|
||||
|
||||
|
||||
class AuthLogoutSuite:
|
||||
"""Benchmark auth logout throughput."""
|
||||
|
||||
def setup(self) -> None:
|
||||
global _config_data
|
||||
_config_data = {}
|
||||
self._store, self._tmpdir = _make_store()
|
||||
self._store.store_token("bench-tok")
|
||||
|
||||
def teardown(self) -> None:
|
||||
shutil.rmtree(str(self._tmpdir), ignore_errors=True)
|
||||
|
||||
def time_auth_logout(self) -> None:
|
||||
def _ps() -> TokenStore:
|
||||
return self._store
|
||||
|
||||
with (
|
||||
patch.object(auth_mod, "_get_token_store", _ps),
|
||||
patch.object(auth_mod, "_read_active_team", _patched_read_team),
|
||||
patch.object(auth_mod, "_write_active_team", _patched_write_team),
|
||||
):
|
||||
_runner.invoke(auth_app, ["logout"])
|
||||
|
||||
|
||||
class AuthStatusSuite:
|
||||
"""Benchmark auth status throughput."""
|
||||
|
||||
def setup(self) -> None:
|
||||
global _config_data
|
||||
_config_data = {}
|
||||
self._store, self._tmpdir = _make_store()
|
||||
|
||||
def teardown(self) -> None:
|
||||
shutil.rmtree(str(self._tmpdir), ignore_errors=True)
|
||||
|
||||
def time_auth_status(self) -> None:
|
||||
def _ps() -> TokenStore:
|
||||
return self._store
|
||||
|
||||
with (
|
||||
patch.object(auth_mod, "_get_token_store", _ps),
|
||||
patch.object(auth_mod, "_read_active_team", _patched_read_team),
|
||||
patch.object(auth_mod, "_write_active_team", _patched_write_team),
|
||||
):
|
||||
_runner.invoke(auth_app, ["status"])
|
||||
|
||||
|
||||
class TeamListSuite:
|
||||
"""Benchmark team list throughput."""
|
||||
|
||||
def setup(self) -> None:
|
||||
global _config_data
|
||||
_config_data = {}
|
||||
self._store, self._tmpdir = _make_store()
|
||||
|
||||
def teardown(self) -> None:
|
||||
shutil.rmtree(str(self._tmpdir), ignore_errors=True)
|
||||
|
||||
def time_team_list(self) -> None:
|
||||
with (
|
||||
patch.object(team_mod, "_read_active_team", _patched_read_team),
|
||||
patch.object(team_mod, "_write_active_team", _patched_write_team),
|
||||
):
|
||||
_runner.invoke(team_app, ["list"])
|
||||
|
||||
|
||||
class TeamUseSuite:
|
||||
"""Benchmark team use throughput."""
|
||||
|
||||
def setup(self) -> None:
|
||||
global _config_data
|
||||
_config_data = {}
|
||||
self._store, self._tmpdir = _make_store()
|
||||
|
||||
def teardown(self) -> None:
|
||||
shutil.rmtree(str(self._tmpdir), ignore_errors=True)
|
||||
|
||||
def time_team_use(self) -> None:
|
||||
with (
|
||||
patch.object(team_mod, "_read_active_team", _patched_read_team),
|
||||
patch.object(team_mod, "_write_active_team", _patched_write_team),
|
||||
):
|
||||
_runner.invoke(team_app, ["use", "bench-team"])
|
||||
|
||||
|
||||
class TokenStoreSuite:
|
||||
"""Benchmark token store operations."""
|
||||
|
||||
def setup(self) -> None:
|
||||
self._store, self._tmpdir = _make_store()
|
||||
|
||||
def teardown(self) -> None:
|
||||
shutil.rmtree(str(self._tmpdir), ignore_errors=True)
|
||||
|
||||
def time_store_token(self) -> None:
|
||||
self._store.store_token("bench-token-value")
|
||||
|
||||
def time_get_token(self) -> None:
|
||||
self._store.store_token("bench-token-value")
|
||||
self._store.get_token()
|
||||
|
||||
def time_clear_token(self) -> None:
|
||||
self._store.store_token("bench-token-value")
|
||||
self._store.clear_token()
|
||||
@@ -0,0 +1,147 @@
|
||||
@phase7 @auth @cli
|
||||
Feature: Auth and Team CLI Commands
|
||||
As a CleverAgents user
|
||||
I want auth login/logout/status and team list/use commands
|
||||
So that I can authenticate and manage team context
|
||||
|
||||
# -------------------------------------------------------------------
|
||||
# Auth Login
|
||||
# -------------------------------------------------------------------
|
||||
|
||||
Scenario: Auth login stores token with stubbed server
|
||||
Given a clean token store
|
||||
When I run auth login with token "test-bearer-token-abc123"
|
||||
Then the auth result status should be "logged_in"
|
||||
And the auth result should indicate server not validated
|
||||
And the auth result should contain a warning about server mode
|
||||
And the token should be stored
|
||||
And the token in output should be redacted
|
||||
|
||||
Scenario: Auth login with empty token fails
|
||||
Given a clean token store
|
||||
When I run auth login with empty token
|
||||
Then the auth command should fail with exit code 1
|
||||
|
||||
Scenario: Auth login with JSON format
|
||||
Given a clean token store
|
||||
When I run auth login with token "test-bearer-token-abc123" and format "json"
|
||||
Then the auth result should be valid JSON
|
||||
And the JSON result should contain key "status" with value "logged_in"
|
||||
And the JSON result should contain key "token" with value "***REDACTED***"
|
||||
|
||||
# -------------------------------------------------------------------
|
||||
# Auth Logout
|
||||
# -------------------------------------------------------------------
|
||||
|
||||
Scenario: Auth logout clears token
|
||||
Given a token store with token "existing-token-xyz"
|
||||
When I run auth logout
|
||||
Then the auth result status should be "logged_out"
|
||||
And the auth result should indicate token was cleared
|
||||
And the token should not be stored
|
||||
And the active team should be cleared
|
||||
|
||||
Scenario: Auth logout when not logged in
|
||||
Given a clean token store
|
||||
When I run auth logout
|
||||
Then the auth result status should be "logged_out"
|
||||
And the auth result should indicate no token was cleared
|
||||
|
||||
Scenario: Auth logout with JSON format
|
||||
Given a token store with token "existing-token-xyz"
|
||||
When I run auth logout with format "json"
|
||||
Then the auth result should be valid JSON
|
||||
And the JSON result should contain key "status" with value "logged_out"
|
||||
|
||||
# -------------------------------------------------------------------
|
||||
# Auth Status
|
||||
# -------------------------------------------------------------------
|
||||
|
||||
Scenario: Auth status when not logged in
|
||||
Given a clean token store
|
||||
When I run auth status
|
||||
Then the auth result should show not authenticated
|
||||
And the auth result should show no active team
|
||||
|
||||
Scenario: Auth status when logged in
|
||||
Given a token store with token "my-auth-token-456"
|
||||
When I run auth status
|
||||
Then the auth result should show authenticated
|
||||
And the auth result should indicate server not validated
|
||||
|
||||
Scenario: Auth status with active team
|
||||
Given a token store with token "my-auth-token-456"
|
||||
And the active team is set to "acme-corp"
|
||||
When I run auth status with format "json"
|
||||
Then the auth result should be valid JSON
|
||||
And the JSON result should contain key "active_team" with value "acme-corp"
|
||||
|
||||
# -------------------------------------------------------------------
|
||||
# Team List
|
||||
# -------------------------------------------------------------------
|
||||
|
||||
Scenario: Team list with stubbed server
|
||||
When I run team list
|
||||
Then the team result should contain a server warning
|
||||
And the team result should show no available teams
|
||||
|
||||
Scenario: Team list with JSON format
|
||||
When I run team list with format "json"
|
||||
Then the team result should be valid JSON
|
||||
And the JSON result should contain key "teams"
|
||||
And the JSON result should contain key "warning"
|
||||
|
||||
# -------------------------------------------------------------------
|
||||
# Team Use
|
||||
# -------------------------------------------------------------------
|
||||
|
||||
Scenario: Team use sets active team
|
||||
When I run team use with name "my-team"
|
||||
Then the team result should show active team "my-team"
|
||||
And the team result should indicate server not validated
|
||||
|
||||
Scenario: Team use switches team
|
||||
Given the active team is set to "old-team"
|
||||
When I run team use with name "new-team"
|
||||
Then the team result should show active team "new-team"
|
||||
And the team result should show previous team "old-team"
|
||||
|
||||
Scenario: Team use with JSON format
|
||||
When I run team use with name "json-team" and format "json"
|
||||
Then the team result should be valid JSON
|
||||
And the JSON result should contain key "active_team" with value "json-team"
|
||||
|
||||
# -------------------------------------------------------------------
|
||||
# Token Storage
|
||||
# -------------------------------------------------------------------
|
||||
|
||||
Scenario: Token store roundtrip via file backend
|
||||
Given a clean token store using file backend
|
||||
When I store token "roundtrip-test-token"
|
||||
Then retrieving the token should return "roundtrip-test-token"
|
||||
And the token store should report file backend
|
||||
|
||||
Scenario: Token store clear removes token
|
||||
Given a token store with token "to-be-cleared"
|
||||
When I clear the token store
|
||||
Then the token store should report no token
|
||||
|
||||
Scenario: Token store rejects empty token
|
||||
Given a clean token store
|
||||
When I try to store an empty token
|
||||
Then a token validation error should be raised
|
||||
|
||||
Scenario: Token store repr redacts token
|
||||
Given a token store with token "secret-value-999"
|
||||
Then the token store repr should contain REDACTED
|
||||
And the token store repr should not contain "secret-value-999"
|
||||
|
||||
# -------------------------------------------------------------------
|
||||
# Token Redaction
|
||||
# -------------------------------------------------------------------
|
||||
|
||||
Scenario: Auth login output never reveals token
|
||||
Given a clean token store
|
||||
When I run auth login with token "sk-ant-sensitive-key-1234567890"
|
||||
Then the raw CLI output should not contain "sk-ant-sensitive-key-1234567890"
|
||||
And the raw CLI output should contain "REDACTED"
|
||||
@@ -0,0 +1,446 @@
|
||||
"""Step definitions for auth and team CLI command scenarios."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import shutil
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from unittest.mock import patch
|
||||
|
||||
from behave import given, then, use_step_matcher, when
|
||||
from behave.runner import Context
|
||||
from typer.testing import CliRunner
|
||||
|
||||
use_step_matcher("re")
|
||||
|
||||
from cleveragents.cli.commands import auth as auth_mod # noqa: E402
|
||||
from cleveragents.cli.commands import team as team_mod # noqa: E402
|
||||
from cleveragents.cli.commands.auth import app as auth_app # noqa: E402
|
||||
from cleveragents.cli.commands.team import app as team_app # noqa: E402
|
||||
from cleveragents.infrastructure.auth.token_store import TokenStore # noqa: E402
|
||||
from cleveragents.shared.redaction import REDACTED # noqa: E402
|
||||
|
||||
runner = CliRunner()
|
||||
|
||||
|
||||
def _make_temp_store(context: Context) -> TokenStore:
|
||||
"""Create a temp-backed TokenStore and store cleanup info."""
|
||||
tmpdir = Path(tempfile.mkdtemp())
|
||||
store = TokenStore(token_dir=tmpdir, use_keyring=False)
|
||||
if not hasattr(context, "_tmp_dirs"):
|
||||
context._tmp_dirs = []
|
||||
context._tmp_dirs.append(tmpdir)
|
||||
return store
|
||||
|
||||
|
||||
def _cleanup_temps(context: Context) -> None:
|
||||
"""Remove temp dirs created during the scenario."""
|
||||
for d in getattr(context, "_tmp_dirs", []):
|
||||
shutil.rmtree(str(d), ignore_errors=True)
|
||||
|
||||
|
||||
def _run_auth(context: Context, args: list[str]) -> Any:
|
||||
"""Run auth CLI command with patched token store."""
|
||||
store: TokenStore = context.token_store
|
||||
|
||||
def _patched_store() -> TokenStore:
|
||||
return store
|
||||
|
||||
def _patched_read_team() -> str | None:
|
||||
data = getattr(context, "_config_data", {})
|
||||
val = data.get("active_team")
|
||||
if val is not None and str(val).strip():
|
||||
return str(val).strip()
|
||||
return None
|
||||
|
||||
def _patched_write_team(team: str | None) -> None:
|
||||
if not hasattr(context, "_config_data"):
|
||||
context._config_data = {}
|
||||
if team is None:
|
||||
context._config_data.pop("active_team", None)
|
||||
else:
|
||||
context._config_data["active_team"] = team
|
||||
|
||||
with (
|
||||
patch.object(auth_mod, "_get_token_store", _patched_store),
|
||||
patch.object(auth_mod, "_read_active_team", _patched_read_team),
|
||||
patch.object(auth_mod, "_write_active_team", _patched_write_team),
|
||||
):
|
||||
result = runner.invoke(auth_app, args)
|
||||
context.result = result
|
||||
context.raw_output = result.output
|
||||
return result
|
||||
|
||||
|
||||
def _run_team(context: Context, args: list[str]) -> Any:
|
||||
"""Run team CLI command with patched helpers."""
|
||||
|
||||
def _patched_read_team() -> str | None:
|
||||
data = getattr(context, "_config_data", {})
|
||||
val = data.get("active_team")
|
||||
if val is not None and str(val).strip():
|
||||
return str(val).strip()
|
||||
return None
|
||||
|
||||
def _patched_write_team(team: str | None) -> None:
|
||||
if not hasattr(context, "_config_data"):
|
||||
context._config_data = {}
|
||||
if team is None:
|
||||
context._config_data.pop("active_team", None)
|
||||
else:
|
||||
context._config_data["active_team"] = team
|
||||
|
||||
with (
|
||||
patch.object(team_mod, "_read_active_team", _patched_read_team),
|
||||
patch.object(team_mod, "_write_active_team", _patched_write_team),
|
||||
):
|
||||
result = runner.invoke(team_app, args)
|
||||
context.result = result
|
||||
context.raw_output = result.output
|
||||
return result
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Given
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@given(r"a clean token store")
|
||||
def step_clean_token_store(context: Context) -> None:
|
||||
context.token_store = _make_temp_store(context)
|
||||
context._config_data = {}
|
||||
|
||||
|
||||
@given(r'a token store with token "(?P<token>[^"]*)"')
|
||||
def step_token_store_with_token(context: Context, token: str) -> None:
|
||||
store = _make_temp_store(context)
|
||||
store.store_token(token)
|
||||
context.token_store = store
|
||||
context._config_data = {}
|
||||
|
||||
|
||||
@given(r"a clean token store using file backend")
|
||||
def step_clean_token_store_file(context: Context) -> None:
|
||||
context.token_store = _make_temp_store(context)
|
||||
context._config_data = {}
|
||||
|
||||
|
||||
@given(r'the active team is set to "(?P<team>[^"]*)"')
|
||||
def step_active_team_set(context: Context, team: str) -> None:
|
||||
if not hasattr(context, "_config_data"):
|
||||
context._config_data = {}
|
||||
context._config_data["active_team"] = team
|
||||
if not hasattr(context, "token_store"):
|
||||
context.token_store = _make_temp_store(context)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# When
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@when(r'I run auth login with token "(?P<token>[^"]*)" and format "(?P<fmt>[^"]*)"')
|
||||
def step_run_auth_login_fmt(context: Context, token: str, fmt: str) -> None:
|
||||
_run_auth(context, ["login", "--token", token, "--format", fmt])
|
||||
|
||||
|
||||
@when(r'I run auth login with token "(?P<token>[^"]*)"')
|
||||
def step_run_auth_login(context: Context, token: str) -> None:
|
||||
_run_auth(context, ["login", "--token", token])
|
||||
|
||||
|
||||
@when(r"I run auth login with empty token")
|
||||
def step_run_auth_login_empty(context: Context) -> None:
|
||||
if not hasattr(context, "token_store"):
|
||||
context.token_store = _make_temp_store(context)
|
||||
_run_auth(context, ["login", "--token", ""])
|
||||
|
||||
|
||||
@when(r"I run auth logout")
|
||||
def step_run_auth_logout(context: Context) -> None:
|
||||
_run_auth(context, ["logout"])
|
||||
|
||||
|
||||
@when(r'I run auth logout with format "(?P<fmt>[^"]*)"')
|
||||
def step_run_auth_logout_fmt(context: Context, fmt: str) -> None:
|
||||
_run_auth(context, ["logout", "--format", fmt])
|
||||
|
||||
|
||||
@when(r"I run auth status")
|
||||
def step_run_auth_status(context: Context) -> None:
|
||||
if not hasattr(context, "token_store"):
|
||||
context.token_store = _make_temp_store(context)
|
||||
_run_auth(context, ["status"])
|
||||
|
||||
|
||||
@when(r'I run auth status with format "(?P<fmt>[^"]*)"')
|
||||
def step_run_auth_status_fmt(context: Context, fmt: str) -> None:
|
||||
if not hasattr(context, "token_store"):
|
||||
context.token_store = _make_temp_store(context)
|
||||
_run_auth(context, ["status", "--format", fmt])
|
||||
|
||||
|
||||
@when(r"I run team list")
|
||||
def step_run_team_list(context: Context) -> None:
|
||||
if not hasattr(context, "token_store"):
|
||||
context.token_store = _make_temp_store(context)
|
||||
_run_team(context, ["list"])
|
||||
|
||||
|
||||
@when(r'I run team list with format "(?P<fmt>[^"]*)"')
|
||||
def step_run_team_list_fmt(context: Context, fmt: str) -> None:
|
||||
if not hasattr(context, "token_store"):
|
||||
context.token_store = _make_temp_store(context)
|
||||
_run_team(context, ["list", "--format", fmt])
|
||||
|
||||
|
||||
@when(r'I run team use with name "(?P<name>[^"]*)" and format "(?P<fmt>[^"]*)"')
|
||||
def step_run_team_use_fmt(context: Context, name: str, fmt: str) -> None:
|
||||
if not hasattr(context, "token_store"):
|
||||
context.token_store = _make_temp_store(context)
|
||||
_run_team(context, ["use", name, "--format", fmt])
|
||||
|
||||
|
||||
@when(r'I run team use with name "(?P<name>[^"]*)"')
|
||||
def step_run_team_use(context: Context, name: str) -> None:
|
||||
if not hasattr(context, "token_store"):
|
||||
context.token_store = _make_temp_store(context)
|
||||
_run_team(context, ["use", name])
|
||||
|
||||
|
||||
@when(r'I store token "(?P<token>[^"]*)"')
|
||||
def step_store_token(context: Context, token: str) -> None:
|
||||
context.token_store.store_token(token)
|
||||
|
||||
|
||||
@when(r"I clear the token store")
|
||||
def step_clear_token_store(context: Context) -> None:
|
||||
context.token_store.clear_token()
|
||||
|
||||
|
||||
@when(r"I try to store an empty token")
|
||||
def step_try_store_empty(context: Context) -> None:
|
||||
try:
|
||||
context.token_store.store_token("")
|
||||
context.token_error = None
|
||||
except ValueError as exc:
|
||||
context.token_error = exc
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Then
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@then(r'the auth result status should be "(?P<status>[^"]*)"')
|
||||
def step_auth_result_status(context: Context, status: str) -> None:
|
||||
assert context.result.exit_code == 0, (
|
||||
f"Expected exit 0, got {context.result.exit_code}: {context.result.output}"
|
||||
)
|
||||
assert (
|
||||
status in context.result.output.lower().replace("_", " ").replace("-", " ")
|
||||
or status.replace("_", " ") in context.result.output.lower()
|
||||
), f"Status '{status}' not found in output: {context.result.output}"
|
||||
|
||||
|
||||
@then(r"the auth result should indicate server not validated")
|
||||
def step_server_not_validated(context: Context) -> None:
|
||||
out = context.result.output.lower()
|
||||
assert "false" in out or "not validated" in out or "server mode" in out, (
|
||||
f"Expected server-not-validated indicator in: {context.result.output}"
|
||||
)
|
||||
|
||||
|
||||
@then(r"the auth result should contain a warning about server mode")
|
||||
def step_server_warning(context: Context) -> None:
|
||||
out = context.result.output
|
||||
assert "Server mode not enabled" in out or "server" in out.lower(), (
|
||||
f"Missing server mode warning in: {out}"
|
||||
)
|
||||
|
||||
|
||||
@then(r"the token should be stored")
|
||||
def step_token_stored(context: Context) -> None:
|
||||
assert context.token_store.has_token(), "Token should be stored"
|
||||
|
||||
|
||||
@then(r"the token in output should be redacted")
|
||||
def step_token_redacted(context: Context) -> None:
|
||||
assert REDACTED in context.result.output, (
|
||||
f"Expected {REDACTED} in output: {context.result.output}"
|
||||
)
|
||||
|
||||
|
||||
@then(r"the auth command should fail with exit code 1")
|
||||
def step_auth_fail(context: Context) -> None:
|
||||
assert context.result.exit_code == 1, (
|
||||
f"Expected exit 1, got {context.result.exit_code}"
|
||||
)
|
||||
|
||||
|
||||
@then(r"the auth result should be valid JSON")
|
||||
def step_auth_valid_json(context: Context) -> None:
|
||||
assert context.result.exit_code == 0, (
|
||||
f"Expected exit 0, got {context.result.exit_code}: {context.result.output}"
|
||||
)
|
||||
context.json_result = json.loads(context.result.output)
|
||||
|
||||
|
||||
@then(
|
||||
r'the JSON result should contain key "(?P<key>[^"]*)" with value "(?P<value>[^"]*)"'
|
||||
)
|
||||
def step_json_key_value(context: Context, key: str, value: str) -> None:
|
||||
data = context.json_result
|
||||
assert key in data, f"Key '{key}' not in JSON: {data}"
|
||||
assert str(data[key]) == value, f"Expected {key}={value!r}, got {data[key]!r}"
|
||||
|
||||
|
||||
@then(r'the JSON result should contain key "(?P<key>[^"]*)"')
|
||||
def step_json_key(context: Context, key: str) -> None:
|
||||
data = context.json_result
|
||||
assert key in data, f"Key '{key}' not in JSON: {data}"
|
||||
|
||||
|
||||
@then(r"the auth result should indicate token was cleared")
|
||||
def step_token_cleared(context: Context) -> None:
|
||||
out = context.result.output.lower()
|
||||
assert "true" in out or "cleared" in out, (
|
||||
f"Expected token-cleared indicator: {context.result.output}"
|
||||
)
|
||||
|
||||
|
||||
@then(r"the token should not be stored")
|
||||
def step_token_not_stored(context: Context) -> None:
|
||||
assert not context.token_store.has_token(), "Token should NOT be stored"
|
||||
|
||||
|
||||
@then(r"the active team should be cleared")
|
||||
def step_active_team_cleared(context: Context) -> None:
|
||||
data = getattr(context, "_config_data", {})
|
||||
assert data.get("active_team") is None, (
|
||||
f"Active team should be cleared, got: {data.get('active_team')}"
|
||||
)
|
||||
|
||||
|
||||
@then(r"the auth result should indicate no token was cleared")
|
||||
def step_no_token_cleared(context: Context) -> None:
|
||||
out = context.result.output.lower()
|
||||
assert "false" in out or "no token" in out or "cleared" in out, (
|
||||
f"Expected no-token-cleared indicator: {context.result.output}"
|
||||
)
|
||||
|
||||
|
||||
@then(r"the auth result should show not authenticated")
|
||||
def step_not_authenticated(context: Context) -> None:
|
||||
assert context.result.exit_code == 0
|
||||
out = context.result.output.lower()
|
||||
assert "not authenticated" in out or "false" in out, (
|
||||
f"Expected not-authenticated: {context.result.output}"
|
||||
)
|
||||
|
||||
|
||||
@then(r"the auth result should show no active team")
|
||||
def step_no_active_team(context: Context) -> None:
|
||||
out = context.result.output.lower()
|
||||
assert "(none)" in out or "none" in out, (
|
||||
f"Expected no active team: {context.result.output}"
|
||||
)
|
||||
|
||||
|
||||
@then(r"the auth result should show authenticated")
|
||||
def step_authenticated(context: Context) -> None:
|
||||
assert context.result.exit_code == 0
|
||||
out = context.result.output.lower()
|
||||
assert "authenticated" in out, f"Expected authenticated: {context.result.output}"
|
||||
|
||||
|
||||
@then(r"the team result should contain a server warning")
|
||||
def step_team_server_warning(context: Context) -> None:
|
||||
assert context.result.exit_code == 0
|
||||
out = context.result.output
|
||||
assert "Server mode not enabled" in out or "server" in out.lower(), (
|
||||
f"Expected server warning: {out}"
|
||||
)
|
||||
|
||||
|
||||
@then(r"the team result should show no available teams")
|
||||
def step_no_teams(context: Context) -> None:
|
||||
out = context.result.output.lower()
|
||||
assert "none" in out or "no" in out or "[]" in out, (
|
||||
f"Expected no teams: {context.result.output}"
|
||||
)
|
||||
|
||||
|
||||
@then(r"the team result should be valid JSON")
|
||||
def step_team_valid_json(context: Context) -> None:
|
||||
assert context.result.exit_code == 0, (
|
||||
f"Expected exit 0, got {context.result.exit_code}: {context.result.output}"
|
||||
)
|
||||
context.json_result = json.loads(context.result.output)
|
||||
|
||||
|
||||
@then(r'the team result should show active team "(?P<name>[^"]*)"')
|
||||
def step_team_active(context: Context, name: str) -> None:
|
||||
assert context.result.exit_code == 0
|
||||
out = context.result.output
|
||||
assert name in out, f"Expected team '{name}' in: {out}"
|
||||
|
||||
|
||||
@then(r"the team result should indicate server not validated")
|
||||
def step_team_server_not_validated(context: Context) -> None:
|
||||
out = context.result.output.lower()
|
||||
assert "false" in out or "not validated" in out or "server mode" in out
|
||||
|
||||
|
||||
@then(r'the team result should show previous team "(?P<name>[^"]*)"')
|
||||
def step_team_previous(context: Context, name: str) -> None:
|
||||
out = context.result.output
|
||||
assert name in out, f"Expected previous team '{name}' in: {out}"
|
||||
|
||||
|
||||
@then(r'retrieving the token should return "(?P<expected>[^"]*)"')
|
||||
def step_retrieve_token(context: Context, expected: str) -> None:
|
||||
token = context.token_store.get_token()
|
||||
assert token == expected, f"Expected {expected!r}, got {token!r}"
|
||||
|
||||
|
||||
@then(r"the token store should report file backend")
|
||||
def step_file_backend(context: Context) -> None:
|
||||
assert context.token_store.backend == "file"
|
||||
|
||||
|
||||
@then(r"the token store should report no token")
|
||||
def step_no_token(context: Context) -> None:
|
||||
assert not context.token_store.has_token()
|
||||
|
||||
|
||||
@then(r"a token validation error should be raised")
|
||||
def step_token_validation_error(context: Context) -> None:
|
||||
assert context.token_error is not None, "Expected ValueError"
|
||||
assert isinstance(context.token_error, ValueError)
|
||||
|
||||
|
||||
@then(r"the token store repr should contain REDACTED")
|
||||
def step_repr_redacted(context: Context) -> None:
|
||||
r = repr(context.token_store)
|
||||
assert REDACTED in r, f"Expected REDACTED in repr: {r}"
|
||||
|
||||
|
||||
@then(r'the token store repr should not contain "(?P<value>[^"]*)"')
|
||||
def step_repr_no_value(context: Context, value: str) -> None:
|
||||
r = repr(context.token_store)
|
||||
assert value not in r, f"Value '{value}' should not appear in repr: {r}"
|
||||
|
||||
|
||||
@then(r'the raw CLI output should not contain "(?P<text>[^"]*)"')
|
||||
def step_raw_no_text(context: Context, text: str) -> None:
|
||||
assert text not in context.raw_output, f"Text '{text}' should not appear in output"
|
||||
|
||||
|
||||
@then(r'the raw CLI output should contain "(?P<text>[^"]*)"')
|
||||
def step_raw_has_text(context: Context, text: str) -> None:
|
||||
assert text in context.raw_output, (
|
||||
f"Expected '{text}' in output: {context.raw_output}"
|
||||
)
|
||||
@@ -78,6 +78,9 @@ tests = [
|
||||
"robotframework>=7.3.2",
|
||||
"robotframework-pabot>=4.0.0",
|
||||
]
|
||||
keyring = [
|
||||
"keyring>=25.0.0",
|
||||
]
|
||||
docs = [
|
||||
"mkdocs>=1.6.1",
|
||||
"mkdocs-material>=9.6.0",
|
||||
|
||||
@@ -0,0 +1,81 @@
|
||||
*** Settings ***
|
||||
Documentation Smoke tests for Auth and Team CLI commands
|
||||
Resource ${CURDIR}/common.resource
|
||||
Suite Setup Setup Test Environment
|
||||
Suite Teardown Cleanup Test Environment
|
||||
|
||||
*** Variables ***
|
||||
${HELPER} ${CURDIR}/helper_auth_commands.py
|
||||
|
||||
*** Test Cases ***
|
||||
Auth Login With Token
|
||||
[Documentation] Verify that auth login stores a token and returns status
|
||||
${result}= Run Process ${PYTHON} ${HELPER} login cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} auth-login-ok
|
||||
|
||||
Auth Login JSON Format
|
||||
[Documentation] Verify auth login JSON output
|
||||
${result}= Run Process ${PYTHON} ${HELPER} login-json cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} auth-login-json-ok
|
||||
|
||||
Auth Logout
|
||||
[Documentation] Verify auth logout clears token
|
||||
${result}= Run Process ${PYTHON} ${HELPER} logout cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} auth-logout-ok
|
||||
|
||||
Auth Status
|
||||
[Documentation] Verify auth status reports state
|
||||
${result}= Run Process ${PYTHON} ${HELPER} status cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} auth-status-ok
|
||||
|
||||
Auth Token Redaction
|
||||
[Documentation] Verify tokens are redacted in output
|
||||
${result}= Run Process ${PYTHON} ${HELPER} redaction cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} auth-redaction-ok
|
||||
|
||||
Team List
|
||||
[Documentation] Verify team list returns stubbed response
|
||||
${result}= Run Process ${PYTHON} ${HELPER} team-list cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} team-list-ok
|
||||
|
||||
Team Use
|
||||
[Documentation] Verify team use switches active team
|
||||
${result}= Run Process ${PYTHON} ${HELPER} team-use cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} team-use-ok
|
||||
|
||||
Team Use JSON Format
|
||||
[Documentation] Verify team use JSON output
|
||||
${result}= Run Process ${PYTHON} ${HELPER} team-use-json cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} team-use-json-ok
|
||||
|
||||
Token Store Roundtrip
|
||||
[Documentation] Verify token store/retrieve/clear cycle
|
||||
${result}= Run Process ${PYTHON} ${HELPER} token-roundtrip cwd=${WORKSPACE}
|
||||
Log ${result.stdout}
|
||||
Log ${result.stderr}
|
||||
Should Be Equal As Integers ${result.rc} 0
|
||||
Should Contain ${result.stdout} token-roundtrip-ok
|
||||
@@ -0,0 +1,267 @@
|
||||
"""Helper script for auth_commands.robot smoke tests.
|
||||
|
||||
Each subcommand is a self-contained check that prints a sentinel on success.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import shutil
|
||||
import sys
|
||||
import tempfile
|
||||
from collections.abc import Callable
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from unittest.mock import patch
|
||||
|
||||
# Ensure local source tree is importable
|
||||
_SRC = str(Path(__file__).resolve().parents[1] / "src")
|
||||
if _SRC not in sys.path:
|
||||
sys.path.insert(0, _SRC)
|
||||
|
||||
from typer.testing import CliRunner # noqa: E402
|
||||
|
||||
from cleveragents.cli.commands import auth as auth_mod # noqa: E402
|
||||
from cleveragents.cli.commands import team as team_mod # noqa: E402
|
||||
from cleveragents.cli.commands.auth import app as auth_app # noqa: E402
|
||||
from cleveragents.cli.commands.team import app as team_app # noqa: E402
|
||||
from cleveragents.infrastructure.auth.token_store import TokenStore # noqa: E402
|
||||
|
||||
runner = CliRunner()
|
||||
|
||||
|
||||
def _make_tmp_store() -> tuple[TokenStore, Path]:
|
||||
"""Create a temp-backed TokenStore and return (store, tmpdir)."""
|
||||
tmpdir = Path(tempfile.mkdtemp())
|
||||
store = TokenStore(token_dir=tmpdir, use_keyring=False)
|
||||
return store, tmpdir
|
||||
|
||||
|
||||
_config_data: dict[str, Any] = {}
|
||||
|
||||
|
||||
def _run_auth(store: TokenStore, args: list[str]) -> Any:
|
||||
"""Run auth CLI with patched helpers."""
|
||||
global _config_data
|
||||
|
||||
def _patched_store() -> TokenStore:
|
||||
return store
|
||||
|
||||
def _patched_read_team() -> str | None:
|
||||
val = _config_data.get("active_team")
|
||||
if val is not None and str(val).strip():
|
||||
return str(val).strip()
|
||||
return None
|
||||
|
||||
def _patched_write_team(team: str | None) -> None:
|
||||
global _config_data
|
||||
if team is None:
|
||||
_config_data.pop("active_team", None)
|
||||
else:
|
||||
_config_data["active_team"] = team
|
||||
|
||||
with (
|
||||
patch.object(auth_mod, "_get_token_store", _patched_store),
|
||||
patch.object(auth_mod, "_read_active_team", _patched_read_team),
|
||||
patch.object(auth_mod, "_write_active_team", _patched_write_team),
|
||||
):
|
||||
return runner.invoke(auth_app, args)
|
||||
|
||||
|
||||
def _run_team(store: TokenStore, args: list[str]) -> Any:
|
||||
"""Run team CLI with patched helpers."""
|
||||
global _config_data
|
||||
|
||||
def _patched_read_team() -> str | None:
|
||||
val = _config_data.get("active_team")
|
||||
if val is not None and str(val).strip():
|
||||
return str(val).strip()
|
||||
return None
|
||||
|
||||
def _patched_write_team(team: str | None) -> None:
|
||||
global _config_data
|
||||
if team is None:
|
||||
_config_data.pop("active_team", None)
|
||||
else:
|
||||
_config_data["active_team"] = team
|
||||
|
||||
with (
|
||||
patch.object(team_mod, "_read_active_team", _patched_read_team),
|
||||
patch.object(team_mod, "_write_active_team", _patched_write_team),
|
||||
):
|
||||
return runner.invoke(team_app, args)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Subcommands
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def auth_login() -> None:
|
||||
"""Verify auth login stores token."""
|
||||
global _config_data
|
||||
_config_data = {}
|
||||
store, tmpdir = _make_tmp_store()
|
||||
result = _run_auth(store, ["login", "--token", "test-token-abc"])
|
||||
if result.exit_code == 0 and store.has_token():
|
||||
print("auth-login-ok")
|
||||
else:
|
||||
print(f"FAIL: login rc={result.exit_code}", file=sys.stderr)
|
||||
print(result.output, file=sys.stderr)
|
||||
sys.exit(1)
|
||||
shutil.rmtree(str(tmpdir), ignore_errors=True)
|
||||
|
||||
|
||||
def auth_login_json() -> None:
|
||||
"""Verify auth login JSON format."""
|
||||
global _config_data
|
||||
_config_data = {}
|
||||
store, tmpdir = _make_tmp_store()
|
||||
result = _run_auth(store, ["login", "--token", "t", "--format", "json"])
|
||||
if result.exit_code != 0:
|
||||
print(f"FAIL: login json rc={result.exit_code}", file=sys.stderr)
|
||||
print(result.output, file=sys.stderr)
|
||||
sys.exit(1)
|
||||
try:
|
||||
data = json.loads(result.output)
|
||||
if data.get("status") == "logged_in":
|
||||
print("auth-login-json-ok")
|
||||
else:
|
||||
print(f"FAIL: unexpected status {data}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
except json.JSONDecodeError as exc:
|
||||
print(f"FAIL: invalid JSON: {exc}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
shutil.rmtree(str(tmpdir), ignore_errors=True)
|
||||
|
||||
|
||||
def auth_logout() -> None:
|
||||
"""Verify auth logout clears token."""
|
||||
global _config_data
|
||||
_config_data = {}
|
||||
store, tmpdir = _make_tmp_store()
|
||||
store.store_token("to-clear")
|
||||
result = _run_auth(store, ["logout"])
|
||||
if result.exit_code == 0 and not store.has_token():
|
||||
print("auth-logout-ok")
|
||||
else:
|
||||
print(f"FAIL: logout rc={result.exit_code}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
shutil.rmtree(str(tmpdir), ignore_errors=True)
|
||||
|
||||
|
||||
def auth_status() -> None:
|
||||
"""Verify auth status shows state."""
|
||||
global _config_data
|
||||
_config_data = {}
|
||||
store, tmpdir = _make_tmp_store()
|
||||
result = _run_auth(store, ["status"])
|
||||
if result.exit_code == 0 and "not authenticated" in result.output.lower():
|
||||
print("auth-status-ok")
|
||||
else:
|
||||
print(f"FAIL: status rc={result.exit_code}", file=sys.stderr)
|
||||
print(result.output, file=sys.stderr)
|
||||
sys.exit(1)
|
||||
shutil.rmtree(str(tmpdir), ignore_errors=True)
|
||||
|
||||
|
||||
def auth_redaction() -> None:
|
||||
"""Verify tokens are redacted in output."""
|
||||
global _config_data
|
||||
_config_data = {}
|
||||
store, tmpdir = _make_tmp_store()
|
||||
result = _run_auth(store, ["login", "--token", "secret-tok"])
|
||||
if result.exit_code == 0 and "secret-tok" not in result.output:
|
||||
print("auth-redaction-ok")
|
||||
else:
|
||||
print(f"FAIL: redaction rc={result.exit_code}", file=sys.stderr)
|
||||
print(result.output, file=sys.stderr)
|
||||
sys.exit(1)
|
||||
shutil.rmtree(str(tmpdir), ignore_errors=True)
|
||||
|
||||
|
||||
def team_list() -> None:
|
||||
"""Verify team list returns stubbed response."""
|
||||
global _config_data
|
||||
_config_data = {}
|
||||
store, tmpdir = _make_tmp_store()
|
||||
result = _run_team(store, ["list"])
|
||||
if result.exit_code == 0 and "server" in result.output.lower():
|
||||
print("team-list-ok")
|
||||
else:
|
||||
print(f"FAIL: team list rc={result.exit_code}", file=sys.stderr)
|
||||
print(result.output, file=sys.stderr)
|
||||
sys.exit(1)
|
||||
shutil.rmtree(str(tmpdir), ignore_errors=True)
|
||||
|
||||
|
||||
def team_use() -> None:
|
||||
"""Verify team use switches active team."""
|
||||
global _config_data
|
||||
_config_data = {}
|
||||
store, tmpdir = _make_tmp_store()
|
||||
result = _run_team(store, ["use", "my-team"])
|
||||
if result.exit_code == 0 and "my-team" in result.output:
|
||||
print("team-use-ok")
|
||||
else:
|
||||
print(f"FAIL: team use rc={result.exit_code}", file=sys.stderr)
|
||||
print(result.output, file=sys.stderr)
|
||||
sys.exit(1)
|
||||
shutil.rmtree(str(tmpdir), ignore_errors=True)
|
||||
|
||||
|
||||
def team_use_json() -> None:
|
||||
"""Verify team use JSON output."""
|
||||
global _config_data
|
||||
_config_data = {}
|
||||
store, tmpdir = _make_tmp_store()
|
||||
result = _run_team(store, ["use", "j-team", "--format", "json"])
|
||||
if result.exit_code != 0:
|
||||
print(f"FAIL: team use json rc={result.exit_code}", file=sys.stderr)
|
||||
print(result.output, file=sys.stderr)
|
||||
sys.exit(1)
|
||||
try:
|
||||
data = json.loads(result.output)
|
||||
if data.get("active_team") == "j-team":
|
||||
print("team-use-json-ok")
|
||||
else:
|
||||
print(f"FAIL: unexpected data {data}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
except json.JSONDecodeError as exc:
|
||||
print(f"FAIL: invalid JSON: {exc}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
shutil.rmtree(str(tmpdir), ignore_errors=True)
|
||||
|
||||
|
||||
def token_roundtrip() -> None:
|
||||
"""Verify token store/retrieve/clear cycle."""
|
||||
store, tmpdir = _make_tmp_store()
|
||||
store.store_token("roundtrip-tok")
|
||||
assert store.get_token() == "roundtrip-tok"
|
||||
store.clear_token()
|
||||
assert store.get_token() is None
|
||||
print("token-roundtrip-ok")
|
||||
shutil.rmtree(str(tmpdir), ignore_errors=True)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Dispatcher
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_COMMANDS: dict[str, Callable[[], None]] = {
|
||||
"login": auth_login,
|
||||
"login-json": auth_login_json,
|
||||
"logout": auth_logout,
|
||||
"status": auth_status,
|
||||
"redaction": auth_redaction,
|
||||
"team-list": team_list,
|
||||
"team-use": team_use,
|
||||
"team-use-json": team_use_json,
|
||||
"token-roundtrip": token_roundtrip,
|
||||
}
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
|
||||
print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>", file=sys.stderr)
|
||||
sys.exit(2)
|
||||
_COMMANDS[sys.argv[1]]()
|
||||
@@ -0,0 +1,268 @@
|
||||
"""Authentication CLI commands.
|
||||
|
||||
Provides ``agents auth login``, ``agents auth logout``, and
|
||||
``agents auth status`` commands. When server mode is disabled
|
||||
(stub clients), commands return informative messages rather than
|
||||
attempting real authentication.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
from pathlib import Path
|
||||
from typing import Annotated, Any
|
||||
|
||||
import typer
|
||||
from rich.console import Console
|
||||
from rich.panel import Panel
|
||||
|
||||
from cleveragents.acp.clients import StubAuthClient
|
||||
from cleveragents.cli.formatting import OutputFormat, format_output
|
||||
from cleveragents.infrastructure.auth.token_store import TokenStore
|
||||
from cleveragents.shared.redaction import REDACTED
|
||||
|
||||
app: Any = typer.Typer(help="Authentication management.")
|
||||
console: Console = Console()
|
||||
|
||||
_SERVER_DISABLED_MSG: str = (
|
||||
"Server mode not enabled. "
|
||||
"Authentication requires a running CleverAgents server. "
|
||||
"Configure a server connection with 'agents server connect <url>' first."
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Internal helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _get_token_store() -> TokenStore:
|
||||
"""Return a ``TokenStore`` using the default storage location."""
|
||||
return TokenStore(token_dir=Path.home() / ".cleveragents")
|
||||
|
||||
|
||||
def _get_auth_client() -> StubAuthClient:
|
||||
"""Return the current auth client (always the stub for now)."""
|
||||
return StubAuthClient()
|
||||
|
||||
|
||||
def _read_active_team() -> str | None:
|
||||
"""Read the active team from the config file, or ``None``."""
|
||||
from cleveragents.application.services.config_service import ConfigService
|
||||
|
||||
config_dir = Path.home() / ".cleveragents"
|
||||
config_path = config_dir / "config.toml"
|
||||
svc = ConfigService(config_dir=config_dir, config_path=config_path)
|
||||
data = svc.read_config()
|
||||
value = data.get("active_team")
|
||||
if value is not None and str(value).strip():
|
||||
return str(value).strip()
|
||||
return None
|
||||
|
||||
|
||||
def _write_active_team(team: str | None) -> None:
|
||||
"""Persist or clear the active team in the config file."""
|
||||
from cleveragents.application.services.config_service import ConfigService
|
||||
|
||||
config_dir = Path.home() / ".cleveragents"
|
||||
config_path = config_dir / "config.toml"
|
||||
svc = ConfigService(config_dir=config_dir, config_path=config_path)
|
||||
data = svc.read_config()
|
||||
if team is None:
|
||||
data.pop("active_team", None)
|
||||
else:
|
||||
data["active_team"] = team
|
||||
svc.write_config(data)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Commands
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@app.command("login")
|
||||
def auth_login(
|
||||
token: Annotated[
|
||||
str | None,
|
||||
typer.Option(
|
||||
"--token",
|
||||
"-t",
|
||||
help="Bearer token (omit to be prompted securely)",
|
||||
),
|
||||
] = None,
|
||||
fmt: Annotated[
|
||||
str,
|
||||
typer.Option("--format", "-f", help="Output format: rich, plain, json, yaml"),
|
||||
] = "rich",
|
||||
) -> None:
|
||||
"""Authenticate with a CleverAgents server.
|
||||
|
||||
Provide a bearer token via ``--token`` or omit the flag to be prompted
|
||||
securely. The token is validated against the server and stored in the
|
||||
OS keyring (or encrypted file fallback).
|
||||
|
||||
When server mode is not enabled the command stores the token locally
|
||||
but warns that no server validation occurred.
|
||||
|
||||
Examples::
|
||||
|
||||
agents auth login --token <bearer-token>
|
||||
agents auth login
|
||||
"""
|
||||
if token is None:
|
||||
token = typer.prompt("Enter your auth token", hide_input=True)
|
||||
|
||||
if not token or not token.strip():
|
||||
console.print("[red]Error:[/red] Token must not be empty.")
|
||||
raise typer.Exit(code=1)
|
||||
|
||||
token = token.strip()
|
||||
store = _get_token_store()
|
||||
client = _get_auth_client()
|
||||
|
||||
# Attempt server-side validation
|
||||
server_validated = False
|
||||
try:
|
||||
client.authenticate(token)
|
||||
server_validated = True
|
||||
except NotImplementedError:
|
||||
pass
|
||||
|
||||
# Persist token regardless
|
||||
store.store_token(token)
|
||||
|
||||
result: dict[str, Any] = {
|
||||
"status": "logged_in",
|
||||
"server_validated": server_validated,
|
||||
"token": REDACTED,
|
||||
"backend": store.backend,
|
||||
}
|
||||
|
||||
if not server_validated:
|
||||
result["warning"] = _SERVER_DISABLED_MSG
|
||||
|
||||
if fmt != OutputFormat.RICH.value:
|
||||
typer.echo(format_output(result, fmt))
|
||||
return
|
||||
|
||||
body = (
|
||||
f"[bold]Status:[/bold] logged in\n"
|
||||
f"[bold]Token:[/bold] {REDACTED}\n"
|
||||
f"[bold]Storage:[/bold] {store.backend}\n"
|
||||
f"[bold]Server validated:[/bold] {server_validated}"
|
||||
)
|
||||
if not server_validated:
|
||||
body += f"\n\n[yellow]{_SERVER_DISABLED_MSG}[/yellow]"
|
||||
|
||||
console.print(Panel(body, title="Authentication", expand=False))
|
||||
|
||||
|
||||
@app.command("logout")
|
||||
def auth_logout(
|
||||
fmt: Annotated[
|
||||
str,
|
||||
typer.Option("--format", "-f", help="Output format: rich, plain, json, yaml"),
|
||||
] = "rich",
|
||||
) -> None:
|
||||
"""Remove stored authentication tokens.
|
||||
|
||||
Clears the token from the OS keyring and/or file-backed store and
|
||||
resets the active team selection.
|
||||
|
||||
Examples::
|
||||
|
||||
agents auth logout
|
||||
"""
|
||||
store = _get_token_store()
|
||||
had_token = store.has_token()
|
||||
store.clear_token()
|
||||
_write_active_team(None)
|
||||
|
||||
result: dict[str, Any] = {
|
||||
"status": "logged_out",
|
||||
"token_cleared": had_token,
|
||||
"active_team_cleared": True,
|
||||
}
|
||||
|
||||
if fmt != OutputFormat.RICH.value:
|
||||
typer.echo(format_output(result, fmt))
|
||||
return
|
||||
|
||||
console.print(
|
||||
Panel(
|
||||
f"[bold]Status:[/bold] logged out\n"
|
||||
f"[bold]Token cleared:[/bold] {had_token}\n"
|
||||
f"[bold]Active team cleared:[/bold] True",
|
||||
title="Authentication",
|
||||
expand=False,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
@app.command("status")
|
||||
def auth_status(
|
||||
fmt: Annotated[
|
||||
str,
|
||||
typer.Option("--format", "-f", help="Output format: rich, plain, json, yaml"),
|
||||
] = "rich",
|
||||
) -> None:
|
||||
"""Show current authentication state.
|
||||
|
||||
Displays whether a token is stored, the active user (when server mode
|
||||
is enabled), the active team, and the storage backend.
|
||||
|
||||
Examples::
|
||||
|
||||
agents auth status
|
||||
agents auth status --format json
|
||||
"""
|
||||
store = _get_token_store()
|
||||
has_token = store.has_token()
|
||||
active_team = _read_active_team()
|
||||
|
||||
# Try server-side token validation
|
||||
server_ok = False
|
||||
client = _get_auth_client()
|
||||
if has_token:
|
||||
token = store.get_token()
|
||||
if token is not None:
|
||||
with contextlib.suppress(NotImplementedError):
|
||||
server_ok = client.validate_token(token)
|
||||
|
||||
status_label = "authenticated" if has_token else "not authenticated"
|
||||
|
||||
result: dict[str, Any] = {
|
||||
"status": status_label,
|
||||
"logged_in": has_token,
|
||||
"server_validated": server_ok,
|
||||
"active_team": active_team,
|
||||
"backend": store.backend,
|
||||
}
|
||||
|
||||
if has_token and not server_ok:
|
||||
result["warning"] = _SERVER_DISABLED_MSG
|
||||
|
||||
if fmt != OutputFormat.RICH.value:
|
||||
typer.echo(format_output(result, fmt))
|
||||
return
|
||||
|
||||
team_display = active_team if active_team else "(none)"
|
||||
body = (
|
||||
f"[bold]Status:[/bold] {status_label}\n"
|
||||
f"[bold]Logged in:[/bold] {has_token}\n"
|
||||
f"[bold]Server validated:[/bold] {server_ok}\n"
|
||||
f"[bold]Active team:[/bold] {team_display}\n"
|
||||
f"[bold]Storage:[/bold] {store.backend}"
|
||||
)
|
||||
if has_token and not server_ok:
|
||||
body += f"\n\n[yellow]{_SERVER_DISABLED_MSG}[/yellow]"
|
||||
|
||||
console.print(Panel(body, title="Auth Status", expand=False))
|
||||
|
||||
|
||||
__all__ = [
|
||||
"app",
|
||||
"auth_login",
|
||||
"auth_logout",
|
||||
"auth_status",
|
||||
]
|
||||
@@ -0,0 +1,185 @@
|
||||
"""Team management CLI commands.
|
||||
|
||||
Provides ``agents team list`` and ``agents team use`` commands.
|
||||
When server mode is disabled (stub clients), commands return
|
||||
informative stubbed responses.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
from typing import Annotated, Any
|
||||
|
||||
import typer
|
||||
from rich.console import Console
|
||||
from rich.panel import Panel
|
||||
from rich.table import Table
|
||||
|
||||
from cleveragents.acp.clients import StubServerClient
|
||||
from cleveragents.cli.commands.auth import (
|
||||
_read_active_team,
|
||||
_write_active_team,
|
||||
)
|
||||
from cleveragents.cli.formatting import OutputFormat, format_output
|
||||
|
||||
app: Any = typer.Typer(help="Team / organization management.")
|
||||
console: Console = Console()
|
||||
|
||||
_SERVER_DISABLED_MSG: str = (
|
||||
"Server mode not enabled. "
|
||||
"Team management requires a running CleverAgents server. "
|
||||
"Configure a server connection with 'agents server connect <url>' first."
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Internal helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _get_server_client() -> StubServerClient:
|
||||
"""Return the current server client (always the stub for now)."""
|
||||
return StubServerClient()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Commands
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@app.command("list")
|
||||
def team_list(
|
||||
fmt: Annotated[
|
||||
str,
|
||||
typer.Option("--format", "-f", help="Output format: rich, plain, json, yaml"),
|
||||
] = "rich",
|
||||
) -> None:
|
||||
"""List available teams / organizations.
|
||||
|
||||
When server mode is enabled, queries the server for the user's
|
||||
teams. In local-only mode a stubbed response is returned.
|
||||
|
||||
Examples::
|
||||
|
||||
agents team list
|
||||
agents team list --format json
|
||||
"""
|
||||
client = _get_server_client()
|
||||
|
||||
# Attempt server query
|
||||
teams_from_server: list[dict[str, str]] | None = None
|
||||
with contextlib.suppress(NotImplementedError):
|
||||
client.health_check()
|
||||
# If we get here the server is live (unreachable with stub)
|
||||
|
||||
active_team = _read_active_team()
|
||||
|
||||
if teams_from_server is None:
|
||||
# Stub response
|
||||
result: dict[str, Any] = {
|
||||
"teams": [],
|
||||
"active_team": active_team,
|
||||
"warning": _SERVER_DISABLED_MSG,
|
||||
}
|
||||
|
||||
if fmt != OutputFormat.RICH.value:
|
||||
typer.echo(format_output(result, fmt))
|
||||
return
|
||||
|
||||
body = (
|
||||
f"[bold]Active team:[/bold] {active_team or '(none)'}\n"
|
||||
f"[bold]Available teams:[/bold] (none — server not connected)\n"
|
||||
f"\n[yellow]{_SERVER_DISABLED_MSG}[/yellow]"
|
||||
)
|
||||
console.print(Panel(body, title="Teams", expand=False))
|
||||
return
|
||||
|
||||
# Future: render real team list from server
|
||||
result = {
|
||||
"teams": teams_from_server,
|
||||
"active_team": active_team,
|
||||
}
|
||||
|
||||
if fmt != OutputFormat.RICH.value:
|
||||
typer.echo(format_output(result, fmt))
|
||||
return
|
||||
|
||||
table = Table(title="Teams")
|
||||
table.add_column("Name", style="cyan")
|
||||
table.add_column("Active", justify="center")
|
||||
for team in teams_from_server:
|
||||
name = team.get("name", "")
|
||||
is_active = "[green]yes[/green]" if name == active_team else ""
|
||||
table.add_row(name, is_active)
|
||||
console.print(table)
|
||||
|
||||
|
||||
@app.command("use")
|
||||
def team_use(
|
||||
team_name: Annotated[
|
||||
str,
|
||||
typer.Argument(help="Name of the team to switch to"),
|
||||
],
|
||||
fmt: Annotated[
|
||||
str,
|
||||
typer.Option("--format", "-f", help="Output format: rich, plain, json, yaml"),
|
||||
] = "rich",
|
||||
) -> None:
|
||||
"""Switch the active team / namespace.
|
||||
|
||||
Persists the selection to the global config file. When server mode
|
||||
is disabled the team name is stored but no server-side validation
|
||||
occurs.
|
||||
|
||||
Examples::
|
||||
|
||||
agents team use my-team
|
||||
agents team use acme-corp --format json
|
||||
"""
|
||||
if not team_name or not team_name.strip():
|
||||
console.print("[red]Error:[/red] Team name must not be empty.")
|
||||
raise typer.Exit(code=1)
|
||||
|
||||
team_name = team_name.strip()
|
||||
previous = _read_active_team()
|
||||
_write_active_team(team_name)
|
||||
|
||||
# Attempt server validation
|
||||
server_validated = False
|
||||
client = _get_server_client()
|
||||
try:
|
||||
client.health_check()
|
||||
server_validated = True
|
||||
except NotImplementedError:
|
||||
pass
|
||||
|
||||
result: dict[str, Any] = {
|
||||
"active_team": team_name,
|
||||
"previous_team": previous,
|
||||
"server_validated": server_validated,
|
||||
}
|
||||
|
||||
if not server_validated:
|
||||
result["warning"] = _SERVER_DISABLED_MSG
|
||||
|
||||
if fmt != OutputFormat.RICH.value:
|
||||
typer.echo(format_output(result, fmt))
|
||||
return
|
||||
|
||||
prev_display = previous if previous else "(none)"
|
||||
body = (
|
||||
f"[bold]Active team:[/bold] {team_name}\n"
|
||||
f"[bold]Previous:[/bold] {prev_display}\n"
|
||||
f"[bold]Server validated:[/bold] {server_validated}"
|
||||
)
|
||||
if not server_validated:
|
||||
body += f"\n\n[yellow]{_SERVER_DISABLED_MSG}[/yellow]"
|
||||
|
||||
console.print(Panel(body, title="Team Switch", expand=False))
|
||||
|
||||
|
||||
__all__ = [
|
||||
"app",
|
||||
"team_list",
|
||||
"team_use",
|
||||
]
|
||||
@@ -94,9 +94,11 @@ def _register_subcommands() -> None:
|
||||
tool,
|
||||
validation,
|
||||
)
|
||||
from cleveragents.cli.commands.auth import app as auth_app
|
||||
from cleveragents.cli.commands.auto_debug import app as auto_debug_app
|
||||
from cleveragents.cli.commands.repl import _repl_app
|
||||
from cleveragents.cli.commands.server import app as server_app
|
||||
from cleveragents.cli.commands.team import app as team_app
|
||||
except Exception as exc: # pragma: no cover
|
||||
import traceback
|
||||
|
||||
@@ -194,6 +196,16 @@ def _register_subcommands() -> None:
|
||||
name="server",
|
||||
help="Server connection management (stub)",
|
||||
)
|
||||
app.add_typer(
|
||||
auth_app,
|
||||
name="auth",
|
||||
help="Authentication management (login, logout, status)",
|
||||
)
|
||||
app.add_typer(
|
||||
team_app,
|
||||
name="team",
|
||||
help="Team / organization management",
|
||||
)
|
||||
|
||||
_subcommands_registered = True
|
||||
|
||||
@@ -224,6 +236,8 @@ def _print_basic_help() -> None:
|
||||
typer.echo(" tell Create a plan (shortcut)")
|
||||
typer.echo(" build Build the current plan")
|
||||
typer.echo(" apply Apply plan changes")
|
||||
typer.echo(" auth Authentication (login, logout, status)")
|
||||
typer.echo(" team Team / organization management")
|
||||
typer.echo(" auto-debug Auto-debug operations")
|
||||
typer.echo(" repl Interactive REPL session")
|
||||
typer.echo(" version Show version")
|
||||
@@ -611,6 +625,8 @@ def main(args: list[str] | None = None) -> int:
|
||||
"invariant", # Invariant constraint management
|
||||
"repl", # Interactive REPL
|
||||
"server", # Server connection management
|
||||
"auth", # Authentication management
|
||||
"team", # Team / organization management
|
||||
"tell", # Shortcut for plan tell
|
||||
"build", # Shortcut for plan build
|
||||
"apply", # Shortcut for plan apply
|
||||
|
||||
@@ -435,6 +435,23 @@ class Settings(BaseSettings):
|
||||
description="Completed job retention in seconds before cleanup.",
|
||||
)
|
||||
|
||||
# Auth / team configuration (M7 - post-auth)
|
||||
auth_token: str | None = Field(
|
||||
default=None,
|
||||
validation_alias=AliasChoices("CLEVERAGENTS_AUTH_TOKEN"),
|
||||
description="Bearer token for server authentication.",
|
||||
)
|
||||
active_team: str | None = Field(
|
||||
default=None,
|
||||
validation_alias=AliasChoices("CLEVERAGENTS_ACTIVE_TEAM"),
|
||||
description="Currently active team/org namespace.",
|
||||
)
|
||||
default_namespace: str | None = Field(
|
||||
default=None,
|
||||
validation_alias=AliasChoices("CLEVERAGENTS_DEFAULT_NAMESPACE"),
|
||||
description="Default namespace used when no team is selected.",
|
||||
)
|
||||
|
||||
# Mock providers flag (M4 - provider fixes)
|
||||
mock_providers: bool = Field(
|
||||
default=False,
|
||||
|
||||
@@ -0,0 +1,9 @@
|
||||
"""Authentication infrastructure — token storage and retrieval.
|
||||
|
||||
Provides :class:`TokenStore` which persists authentication tokens using
|
||||
the OS keyring when available, falling back to an obfuscated local file.
|
||||
"""
|
||||
|
||||
from cleveragents.infrastructure.auth.token_store import TokenStore
|
||||
|
||||
__all__ = ["TokenStore"]
|
||||
@@ -0,0 +1,236 @@
|
||||
"""Secure token storage with OS keyring and file fallback.
|
||||
|
||||
The :class:`TokenStore` persists bearer tokens using the OS keyring
|
||||
(via the ``keyring`` library) when available. When keyring is not
|
||||
installed or is unavailable, tokens are stored in an obfuscated
|
||||
local file with restricted permissions (``0600``).
|
||||
|
||||
Token values are validated (non-empty, printable ASCII) and always
|
||||
redacted in string representations.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import contextlib
|
||||
import logging
|
||||
import re
|
||||
import stat
|
||||
from pathlib import Path
|
||||
from typing import Any, Final
|
||||
|
||||
from cleveragents.shared.redaction import REDACTED
|
||||
|
||||
logger: logging.Logger = logging.getLogger(__name__)
|
||||
|
||||
_SERVICE_NAME: Final[str] = "cleveragents"
|
||||
_USERNAME: Final[str] = "auth_token"
|
||||
_TOKEN_PATTERN: re.Pattern[str] = re.compile(r"^[\x20-\x7E]+$")
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _default_token_dir() -> Path:
|
||||
"""Return the default directory for file-backed token storage."""
|
||||
return Path.home() / ".cleveragents"
|
||||
|
||||
|
||||
def _validate_token(token: str) -> None:
|
||||
"""Raise ``ValueError`` when *token* is not a valid bearer token."""
|
||||
if not token:
|
||||
raise ValueError("Token must not be empty")
|
||||
if not _TOKEN_PATTERN.match(token):
|
||||
raise ValueError("Token contains invalid characters")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Keyring helpers (optional dependency)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _import_keyring() -> Any:
|
||||
"""Import and return the ``keyring`` module, or ``None``."""
|
||||
try:
|
||||
import keyring # type: ignore[import-not-found,import-untyped]
|
||||
|
||||
return keyring
|
||||
except ImportError: # pragma: no cover
|
||||
return None
|
||||
|
||||
|
||||
def _keyring_available() -> bool:
|
||||
"""Return ``True`` when the ``keyring`` library is importable."""
|
||||
try:
|
||||
kr: Any = _import_keyring()
|
||||
if kr is None:
|
||||
return False
|
||||
backend = kr.get_keyring()
|
||||
name = type(backend).__name__.lower()
|
||||
return "fail" not in name and "null" not in name
|
||||
except Exception: # pragma: no cover
|
||||
return False
|
||||
|
||||
|
||||
def _keyring_store(token: str) -> None:
|
||||
"""Store *token* in the OS keyring."""
|
||||
kr: Any = _import_keyring()
|
||||
kr.set_password(_SERVICE_NAME, _USERNAME, token)
|
||||
|
||||
|
||||
def _keyring_get() -> str | None:
|
||||
"""Retrieve the token from the OS keyring, or ``None``."""
|
||||
kr: Any = _import_keyring()
|
||||
result: str | None = kr.get_password(_SERVICE_NAME, _USERNAME)
|
||||
return result
|
||||
|
||||
|
||||
def _keyring_clear() -> None:
|
||||
"""Remove the token from the OS keyring."""
|
||||
kr: Any = _import_keyring()
|
||||
with contextlib.suppress(Exception):
|
||||
kr.delete_password(_SERVICE_NAME, _USERNAME)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# File-backed helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _file_path(token_dir: Path) -> Path:
|
||||
"""Return the path to the file-backed token store."""
|
||||
return token_dir / ".auth_token"
|
||||
|
||||
|
||||
def _file_store(token: str, token_dir: Path) -> None:
|
||||
"""Write *token* (base64-encoded) to the local file store."""
|
||||
token_dir.mkdir(parents=True, exist_ok=True)
|
||||
path = _file_path(token_dir)
|
||||
encoded = base64.b64encode(token.encode("utf-8")).decode("ascii")
|
||||
path.write_text(encoded, encoding="ascii")
|
||||
# Restrict permissions to owner-only
|
||||
with contextlib.suppress(OSError): # pragma: no cover - Windows
|
||||
path.chmod(stat.S_IRUSR | stat.S_IWUSR)
|
||||
|
||||
|
||||
def _file_get(token_dir: Path) -> str | None:
|
||||
"""Read the token from the local file store, or ``None``."""
|
||||
path = _file_path(token_dir)
|
||||
if not path.exists():
|
||||
return None
|
||||
try:
|
||||
encoded = path.read_text(encoding="ascii").strip()
|
||||
return base64.b64decode(encoded).decode("utf-8")
|
||||
except Exception:
|
||||
logger.warning("Corrupt token file at %s - ignoring", path)
|
||||
return None
|
||||
|
||||
|
||||
def _file_clear(token_dir: Path) -> None:
|
||||
"""Remove the file-backed token store."""
|
||||
path = _file_path(token_dir)
|
||||
if path.exists():
|
||||
path.unlink()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Public API
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TokenStore:
|
||||
"""Secure token storage with keyring and file fallback.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
token_dir:
|
||||
Directory used for file-backed storage when the OS keyring is
|
||||
unavailable. Defaults to ``~/.cleveragents``.
|
||||
use_keyring:
|
||||
Explicit override; when ``None`` (default) the store auto-detects
|
||||
keyring availability.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
token_dir: Path | None = None,
|
||||
*,
|
||||
use_keyring: bool | None = None,
|
||||
) -> None:
|
||||
self._token_dir: Path = token_dir or _default_token_dir()
|
||||
if use_keyring is not None:
|
||||
self._use_keyring: bool = use_keyring
|
||||
else:
|
||||
self._use_keyring = _keyring_available()
|
||||
self._backend: str = "keyring" if self._use_keyring else "file"
|
||||
|
||||
# -- public interface ---------------------------------------------------
|
||||
|
||||
@property
|
||||
def backend(self) -> str:
|
||||
"""Return ``"keyring"`` or ``"file"`` depending on active backend."""
|
||||
return self._backend
|
||||
|
||||
def store_token(self, token: str) -> None:
|
||||
"""Persist *token* securely.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
token:
|
||||
A non-empty printable-ASCII bearer token.
|
||||
|
||||
Raises
|
||||
------
|
||||
ValueError:
|
||||
If the token is empty or contains non-printable characters.
|
||||
"""
|
||||
_validate_token(token)
|
||||
if self._use_keyring:
|
||||
try:
|
||||
_keyring_store(token)
|
||||
logger.debug("Token stored via OS keyring")
|
||||
return
|
||||
except Exception:
|
||||
logger.warning("Keyring write failed - falling back to file storage")
|
||||
self._use_keyring = False
|
||||
self._backend = "file"
|
||||
_file_store(token, self._token_dir)
|
||||
logger.debug("Token stored via file backend at %s", self._token_dir)
|
||||
|
||||
def get_token(self) -> str | None:
|
||||
"""Retrieve the stored token, or ``None`` if not present."""
|
||||
if self._use_keyring:
|
||||
try:
|
||||
token = _keyring_get()
|
||||
if token is not None:
|
||||
return token
|
||||
except Exception:
|
||||
logger.warning("Keyring read failed - falling back to file storage")
|
||||
self._use_keyring = False
|
||||
self._backend = "file"
|
||||
return _file_get(self._token_dir)
|
||||
|
||||
def clear_token(self) -> None:
|
||||
"""Remove any stored token from both backends."""
|
||||
if self._use_keyring:
|
||||
with contextlib.suppress(Exception):
|
||||
_keyring_clear()
|
||||
_file_clear(self._token_dir)
|
||||
logger.debug("Token cleared from all backends")
|
||||
|
||||
def has_token(self) -> bool:
|
||||
"""Return ``True`` if a token is currently stored."""
|
||||
return self.get_token() is not None
|
||||
|
||||
def __repr__(self) -> str:
|
||||
"""Safe repr that never reveals the token."""
|
||||
has = self.has_token()
|
||||
return (
|
||||
f"TokenStore(backend={self._backend!r}, "
|
||||
f"has_token={has}, "
|
||||
f"token={REDACTED if has else None!r})"
|
||||
)
|
||||
|
||||
|
||||
__all__ = ["TokenStore"]
|
||||
Reference in New Issue
Block a user