Compare commits

...

1 Commits

Author SHA1 Message Date
freemo 92f7feaa32 feat(cli): add auth and team commands
CI / benchmark-publish (pull_request) Has been skipped
CI / lint (pull_request) Successful in 15s
CI / build (pull_request) Successful in 15s
CI / quality (pull_request) Successful in 17s
CI / security (pull_request) Successful in 34s
CI / typecheck (pull_request) Successful in 36s
CI / unit_tests (pull_request) Successful in 2m54s
CI / integration_tests (pull_request) Successful in 3m15s
CI / docker (pull_request) Successful in 41s
CI / coverage (pull_request) Successful in 5m45s
CI / benchmark-regression (pull_request) Successful in 31m1s
Implement auth login/logout/status and team list/use CLI commands
with stubbed responses when server is disabled. Includes token
storage via OS keyring with encrypted file fallback.

Key changes:
- Add agents auth login/logout/status commands
- Add agents team list/use commands
- Add TokenStore with keyring + encrypted file fallback
- Add auth_token, active_team, default_namespace settings
- Wire AuthClient and ServerClient stubs
- Add Behave BDD tests, Robot integration tests, ASV benchmarks

ISSUES CLOSED: #340
2026-03-10 06:13:59 +00:00
12 changed files with 1864 additions and 0 deletions
+189
View File
@@ -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()
+147
View File
@@ -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"
+446
View File
@@ -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}"
)
+3
View File
@@ -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",
+81
View File
@@ -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
+267
View File
@@ -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]]()
+268
View File
@@ -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",
]
+185
View File
@@ -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",
]
+16
View File
@@ -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
+17
View File
@@ -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"]